PriorityQueue

泛型优先级队列，支持 ID 去重、外部排序策略、变更订阅与批量操作

概述

PriorityQueue<T> 是一个通用优先级队列，核心能力：

ID 去重 — 相同 id 的 item 只会存一份，重复入队会被静默忽略
外部排序策略 — 排序逻辑由调用方通过 compare 函数注入，队列不对 T 的形状做任何假设
变更订阅 — 每次写操作后触发 subscribe 回调，传入当前完整有序快照，天然适配 Jotai / Zustand
惰性缓存 — 内部维护一个 sorted 数组，只在写操作时重建，读操作零开销

快速上手

import { PriorityQueue } from '@skyroc/utils';

type Task = {
  taskId: string;
  priority: number;
  createdAt: number;
};

const queue = new PriorityQueue<Task>({
  // 从 item 里提取唯一 ID
  getId: t => t.taskId,
  // 排序规则：priority 小的在前（高优先级），priority 相同时 createdAt 早的在前
  compare: (a, b) => a.priority - b.priority || a.createdAt - b.createdAt
});

queue.enqueue({ taskId: '1', priority: 2, createdAt: 1000 });
queue.enqueue({ taskId: '2', priority: 1, createdAt: 2000 });
queue.enqueue({ taskId: '1', priority: 2, createdAt: 1000 }); // 重复，被忽略

console.log(queue.size); // 2
console.log(queue.peek()?.taskId); // '2'（priority 1，优先级更高）

const top = queue.dequeue(); // 取出 taskId='2'
console.log(queue.size); // 1

核心概念

ID 去重

每个 item 通过 getId 函数提取唯一标识，内部用 Map<string, T> 存储。相同 id 的 item 无论入队多少次，队列中只保留一份：

queue.enqueue({ taskId: 'a', priority: 1, createdAt: Date.now() }); // 入队，返回 true
queue.enqueue({ taskId: 'a', priority: 1, createdAt: Date.now() }); // 忽略，返回 false

排序策略

compare(a, b) 遵循 Array.prototype.sort 惯例：

返回负数 → a 排在 b 前面（a 优先级更高）
返回正数 → b 排在 a 前面
返回 0   → 保持原序

示例：数字越小越优先，相同优先级按入队时间排序：

compare: (a, b) => a.priority - b.priority || a.createdAt - b.createdAt;

变更订阅

每次写操作（enqueue / dequeue / remove / clear）完成后，会同步触发所有 subscribe 回调，参数为当前完整有序快照（readonly T[]）：

const unsub = queue.subscribe(sorted => {
  // 同步推送给 Jotai atom 或 Zustand store
  store.set(taskQueueAtom, [...sorted]);
});

// 不再需要时取消
unsub();

API

构造函数

new PriorityQueue<T>(config: QueueConfig<T>)

interface QueueConfig<T> {
  /** 从 item 中提取唯一标识，用于去重和定向移除 */
  getId: (item: T) => string;
  /** 排序比较器，遵循 Array.prototype.sort 惯例 */
  compare: (a: T, b: T) => number;
  /** 队列容量上限，超出时丢弃排序后最末尾的那些。不传表示不限 */
  capacity?: number;
}

只需两个纯函数即可驱动整个队列，不对 T 的形状做任何假设。

裁剪放在队列里做而不是交给调用方：它必须发生在排序之后（丢的是优先级最低的），外面做要多一轮排序和一次额外通知。

// 消息中心只保留优先级最高的 50 条
const queue = new PriorityQueue<Notice>({
  getId: n => n.id,
  compare: (a, b) => b.weight - a.weight,
  capacity: 50
});

写操作

`enqueue(item)` → `boolean`

单条入队。id 已存在则跳过（幂等），返回是否实际入队。

queue.enqueue({ taskId: '1', priority: 0, createdAt: Date.now() }); // true
queue.enqueue({ taskId: '1', priority: 0, createdAt: Date.now() }); // false（已存在）

`enqueueMany(items)` → `number`

批量入队，跳过已存在的 id，仅触发一次排序和订阅通知。返回实际入队数量。

const added = queue.enqueueMany([task1, task2, task3]); // 返回实际新增数

`dequeue()` → `T | undefined`

移除并返回当前优先级最高的 item（队首）。队列为空时返回 undefined。

const task = queue.dequeue();

`remove(id)` → `boolean`

按 id 移除指定 item，返回是否找到并移除。

queue.remove('task-id-1'); // true / false

`removeBy(predicate)` → `number`

按条件批量移除，仅在有移除时触发一次排序和通知。返回实际移除数量。

// 移除所有过期任务
const removed = queue.removeBy(item => item.expiredAt < Date.now());

`update(id, updater)` → `boolean`

按 id 就地更新一条，updater 拿到旧值返回新值。更新后重新排序并通知一次。返回是否找到该 id。

// 把某条任务提权
queue.update('task-1', prev => ({ ...prev, priority: 0 }));

`updateBy(predicate, updater)` → `number`

按条件批量更新，仅在有命中时触发一次排序和通知。返回实际更新数量。

// 把所有超时未处理的任务降权
const updated = queue.updateBy(
  item => Date.now() - item.createdAt > 60_000,
  item => ({ ...item, priority: item.priority + 10 })
);

`setCapacity(capacity)`

运行时调整容量上限。调小后会立即裁掉超出的部分并通知订阅者。

queue.setCapacity(20);

`clear()`

清空队列，触发一次订阅通知。

queue.clear();

读操作

`peek()` → `T | undefined`

查看队首（最高优先级）但不移除。

const top = queue.peek();

`has(id)` → `boolean`

检查指定 id 是否在队列中。

queue.has('task-id-1'); // true / false

`get(id)` → `T | undefined`

按 id 获取 item，不影响队列顺序。

const task = queue.get('task-id-1');

`toArray()` → `readonly T[]`

返回完整有序队列的不可变快照。返回的是内部缓存的引用，不会每次创建新数组。需要修改时先展开：

const sorted = queue.toArray(); // readonly 引用
const mutable = [...queue.toArray()]; // 可变副本

`size` → `number`

队列中 item 数量。

`isEmpty` → `boolean`

队列是否为空。

遍历

支持 for...of，按优先级顺序遍历：

for (const task of queue) {
  console.log(task.taskId, task.priority);
}

在 React / Jotai 中使用

将 PriorityQueue 与 Jotai 结合，实现响应式的优先级队列：

// bannerQueue.ts
import { atom } from 'jotai';
import { PriorityQueue } from '@skyroc/utils';

type Banner = { id: string; priority: number; message: string };

// 内部队列实例（不放入 atom，避免 Jotai 对其做 snapshot）
const queue = new PriorityQueue<Banner>({
  getId: b => b.id,
  compare: (a, b) => a.priority - b.priority
});

// 暴露给 React 的只读 atom
export const bannerListAtom = atom<readonly Banner[]>([]);

// 每次队列变更时同步 atom
export function initBannerQueue(store: ReturnType<typeof import('jotai').createStore>) {
  return queue.subscribe(sorted => {
    store.set(bannerListAtom, sorted);
  });
}

// 操作 API
export const bannerQueue = {
  push: (b: Banner) => queue.enqueue(b),
  dismiss: (id: string) => queue.remove(id),
  clear: () => queue.clear()
};

Last updated on