延迟队列与定时任务

功能说明

DelayQueue 接收任务值与延迟毫秒数，在到期后调用异步处理器，返回可用于取消的任务标识。同一标识再次调度会覆盖旧任务。pending 按到期时间返回快照，clear 清空全部任务。适合延迟通知、订单超时关闭、重试退避和限流窗口结束后的补偿操作。时钟可注入，便于测试与模拟时间跳变。

关键代码

调度时先取消同名任务，计算到期时间并写入 Map，再创建 setTimeout。定时器回调通过 deliver 重新检查到期时间，若系统时钟被回拨或测试时钟尚未推进，则按剩余时间重新挂定时器。到期后先移除记录再执行处理器，防止重入重复投递。定时器尝试调用 unref，避免仅因延迟任务而阻止 Node 进程退出。

使用注意

任务只保存在内存，进程重启会丢失，关键业务需要持久化队列。处理器抛错会被记录到 lastError，不会中断后续任务，调用方应定期检查或直接在处理器内兜底。取消与到期存在竞态，deliver 会再次检查 Map，因此取消后不会再执行。系统时间大幅向前跳变不会自动触发，需要外部唤醒。clear 不等待正在执行的任务。

验证方式

用假时钟或短延迟测试到期调用、值传递与任务标识。调度后立即取消，断言处理器不执行且 size 归零。同名任务重复调度断言只执行最后一次。测试 pending 排序、快照修改不影响内部、clear 清空。注入早于到期时间的 now 验证重排，并测试负延迟、Infinity、处理器异步拒绝与空处理器场景。

代码示例

export interface DelayedItem<T> {
  id: string;
  value: T;
  dueAt: number;
}

export class DelayQueue<T> {
  private readonly items = new Map<string, DelayedItem<T>>();
  private readonly timers = new Map<string, ReturnType<typeof setTimeout>>();
  private readonly handler: (value: T, id: string) => void | Promise<void>;
  private readonly now: () => number;
  private sequence = 0;
  private lastFailure: unknown;

  constructor(
    handler: (value: T, id: string) => void | Promise<void>,
    now: () => number = Date.now,
  ) {
    this.handler = handler;
    this.now = now;
  }

  schedule(value: T, delayMs: number, id = "job-" + (this.sequence += 1)): string {
    if (!Number.isFinite(delayMs) || delayMs < 0) {
      throw new RangeError("delayMs must be a non-negative finite number");
    }
    this.cancel(id);
    const dueAt = this.now() + delayMs;
    this.items.set(id, { id, value, dueAt });
    const timer = setTimeout(() => {
      void this.deliver(id);
    }, delayMs);
    const handle = timer as unknown as { unref?: () => void };
    handle.unref?.();
    this.timers.set(id, timer);
    return id;
  }

  cancel(id: string): boolean {
    const timer = this.timers.get(id);
    if (timer !== undefined) clearTimeout(timer);
    this.timers.delete(id);
    return this.items.delete(id);
  }

  async deliver(id: string): Promise<void> {
    const item = this.items.get(id);
    if (!item) return;
    const remaining = item.dueAt - this.now();
    if (remaining > 0) {
      const timer = setTimeout(() => {
        void this.deliver(id);
      }, remaining);
      const handle = timer as unknown as { unref?: () => void };
      handle.unref?.();
      this.timers.set(id, timer);
      return;
    }
    this.items.delete(id);
    this.timers.delete(id);
    try {
      await this.handler(item.value, id);
    } catch (error) {
      this.lastFailure = error;
    }
  }

  get size(): number {
    return this.items.size;
  }

  get lastError(): unknown {
    return this.lastFailure;
  }

  pending(): DelayedItem<T>[] {
    return [...this.items.values()]
      .sort((left, right) => left.dueAt - right.dueAt)
      .map((item) => ({ ...item }));
  }

  clear(): void {
    for (const id of [...this.items.keys()]) this.cancel(id);
  }
}