延迟队列任务调度器

功能说明

重试退避、定时提醒和缓存预热都需要延迟执行。这个调度器把任务存入最小堆，堆顶始终是最早到期的一项；每次取出所有执行时间不晚于当前时刻的任务，返回名称列表。查询下一个截止时间不需要弹出元素，适合驱动定时器或计算睡眠时长。

任务带上自增序号，执行时间相同时先调度的先执行，避免堆顺序不稳定导致的随机性。调度器不启动线程，也不睡眠，时间由调用方传入，因此可以在测试中直接跳到未来。

关键代码

ScheduledTask 手写 Ord，把 run_at 反向比较，使 BinaryHeap 变成最小堆；时间相同时比较序号。DelayQueue::schedule 用 Instant::now 加延迟生成执行时间。pop_ready 先 peek 判断堆顶是否到期，再弹出，避免借用冲突。next_deadline 只读取堆顶时间，不修改队列。main 用同一个基准时间加上偏移模拟时钟推进。

使用注意

调度器本身不是线程安全的，多线程环境需要加锁或让单个线程独占。Instant 不会回退，但重启后不持久，跨重启的任务应保存绝对时间。堆会随任务数增长，取消任务需要延迟删除标记或重建堆。任务执行时间到了不代表已经执行，只是可以取出。延迟为零的任务在下一次 pop_ready 时立即返回。

验证方式

开始时没有任何任务到期，输出空列表。推进到一百二十毫秒后，预热缓存和刷新令牌两项按序返回；到两百毫秒时，发送邮件也被取出。has deadline 在取出部分任务后仍为 true，直到所有任务清空。可以把两个任务设成相同延迟，验证序号决定先后。

代码示例

use std::cmp::Ordering;
use std::collections::BinaryHeap;
use std::time::{Duration, Instant};

#[derive(Debug)]
struct ScheduledTask {
    run_at: Instant,
    sequence: u64,
    name: String,
}

impl PartialEq for ScheduledTask {
    fn eq(&self, other: &Self) -> bool {
        self.run_at == other.run_at && self.sequence == other.sequence
    }
}

impl Eq for ScheduledTask {}

impl Ord for ScheduledTask {
    fn cmp(&self, other: &Self) -> Ordering {
        other
            .run_at
            .cmp(&self.run_at)
            .then_with(|| other.sequence.cmp(&self.sequence))
    }
}

impl PartialOrd for ScheduledTask {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

struct DelayQueue {
    heap: BinaryHeap<ScheduledTask>,
    sequence: u64,
}

impl DelayQueue {
    fn new() -> Self {
        Self {
            heap: BinaryHeap::new(),
            sequence: 0,
        }
    }

    fn schedule(&mut self, name: &str, delay: Duration) {
        self.sequence += 1;
        self.heap.push(ScheduledTask {
            run_at: Instant::now() + delay,
            sequence: self.sequence,
            name: name.to_string(),
        });
    }

    fn pop_ready(&mut self, now: Instant) -> Vec<String> {
        let mut ready = Vec::new();
        loop {
            let due = match self.heap.peek() {
                Some(task) => task.run_at <= now,
                None => false,
            };
            if !due {
                break;
            }
            if let Some(task) = self.heap.pop() {
                ready.push(task.name);
            }
        }
        ready
    }

    fn next_deadline(&self) -> Option<Instant> {
        self.heap.peek().map(|task| task.run_at)
    }
}

fn main() {
    let start = Instant::now();
    let mut queue = DelayQueue::new();
    queue.schedule("warm-cache", Duration::from_millis(50));
    queue.schedule("send-email", Duration::from_millis(150));
    queue.schedule("refresh-token", Duration::from_millis(100));

    println!("ready at start: {:?}", queue.pop_ready(start));
    println!("has deadline: {}", queue.next_deadline().is_some());
    println!(
        "ready at 120ms: {:?}",
        queue.pop_ready(start + Duration::from_millis(120))
    );
    println!(
        "ready at 200ms: {:?}",
        queue.pop_ready(start + Duration::from_millis(200))
    );
}