带饥饿老化的优先任务队列

功能说明

该示例把异步任务按优先级调度，同时限制并发数。每个任务的得分等于基础优先级加上等待时间除以 agingMs 的整数部分，因此低优先级任务等待越久得分越高，最终会被选中，避免被源源不断的高优先级任务饿死。任务使用唯一 id 注册，返回 Promise 与执行结果或错误。适用于后台导出、图片处理、批处理等混合优先级队列。

关键代码

add 校验 id、优先级与执行函数，并把 id 加入 knownIds 防止重复入队。pick 每次线性扫描任务数组，计算老化后的得分，选择最高分；分数相同时保留更早入队的任务。pump 在并发未满时不断取任务执行，任务完成后减少 active、移除 id、通知空闲等待者并继续调度。onIdle 在没有排队与执行中任务时立即解决，否则把 resolve 放入等待列表。

使用注意

老化会让分数随时间无限增长，长期运行后所有任务分数都会很大，但相对差值仍由优先级与等待时间决定，不影响正确性。优先级必须是有限数值，支持小数。该实现没有取消或暂停接口，已入队任务不能撤回。任务失败只拒绝自己的 Promise，不会停止队列。id 在任务完成后释放，可再次使用。

验证方式

用假时钟：先加入低优先级任务，再加入多个高优先级任务，推进时间越过老化阈值后断言低优先级任务先执行；检查并发数不超过配置值；任务失败时其 Promise 拒绝而其他任务继续；重复 id 抛错；onIdle 在所有任务完成后解决；统计执行顺序与插入顺序在相同分数下保持一致。

代码示例

export interface PriorityQueueOptions {
  concurrency?: number;
  agingMs?: number;
  now?: () => number;
}

interface InternalTask {
  id: string;
  priority: number;
  enqueuedAt: number;
  run: () => Promise<unknown>;
  resolve: (value: unknown) => void;
  reject: (error: unknown) => void;
}

export class PriorityTaskQueue {
  private readonly tasks: InternalTask[] = [];
  private readonly knownIds = new Set<string>();
  private readonly idleWaiters: Array<() => void> = [];
  private readonly concurrency: number;
  private readonly agingMs: number;
  private readonly now: () => number;
  private active = 0;

  constructor(options: PriorityQueueOptions = {}) {
    const concurrency = options.concurrency ?? 4;
    if (!Number.isInteger(concurrency) || concurrency < 1) {
      throw new RangeError("concurrency must be a positive integer");
    }
    const agingMs = options.agingMs ?? 5_000;
    if (!Number.isFinite(agingMs) || agingMs <= 0) {
      throw new RangeError("agingMs must be a positive finite number");
    }
    this.concurrency = concurrency;
    this.agingMs = agingMs;
    this.now = options.now ?? Date.now;
  }

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

  get activeCount(): number {
    return this.active;
  }

  add<T>(id: string, priority: number, run: () => Promise<T>): Promise<T> {
    if (typeof id !== "string" || id.length === 0) {
      throw new TypeError("id must be a non-empty string");
    }
    if (!Number.isFinite(priority)) throw new TypeError("priority must be finite");
    if (typeof run !== "function") throw new TypeError("run must be a function");
    if (this.knownIds.has(id)) throw new Error("duplicate task id: " + id);
    this.knownIds.add(id);

    return new Promise<T>((resolve, reject) => {
      this.tasks.push({
        id,
        priority,
        enqueuedAt: this.now(),
        run,
        resolve: (value) => resolve(value as T),
        reject,
      });
      this.pump();
    });
  }

  onIdle(): Promise<void> {
    if (this.active === 0 && this.tasks.length === 0) return Promise.resolve();
    return new Promise<void>((resolve) => {
      this.idleWaiters.push(resolve);
    });
  }

  private pick(): InternalTask | undefined {
    if (this.tasks.length === 0) return undefined;
    const now = this.now();
    let bestIndex = 0;
    let bestScore = Number.NEGATIVE_INFINITY;
    for (let index = 0; index < this.tasks.length; index += 1) {
      const task = this.tasks[index];
      const waited = Math.max(0, now - task.enqueuedAt);
      const aging = Math.floor(waited / this.agingMs);
      const score = task.priority + aging;
      if (score > bestScore) {
        bestScore = score;
        bestIndex = index;
      }
    }
    return this.tasks.splice(bestIndex, 1)[0];
  }

  private pump(): void {
    while (this.active < this.concurrency) {
      const task = this.pick();
      if (!task) {
        this.maybeIdle();
        return;
      }
      this.active += 1;
      Promise.resolve()
        .then(task.run)
        .then(task.resolve, task.reject)
        .finally(() => {
          this.active -= 1;
          this.knownIds.delete(task.id);
          this.maybeIdle();
          this.pump();
        });
    }
  }

  private maybeIdle(): void {
    if (this.active !== 0 || this.tasks.length !== 0) return;
    const waiters = this.idleWaiters.splice(0);
    for (const resolve of waiters) resolve();
  }
}