p-queue size 已降到阈值以下,onSizeLessThan 仍在等
旧版 p-queue 的背压等待只在 next 事件重查条件。启动暂停队列或提高并发时,出队没有立即唤醒等待者,需核对正式版本和完成语义。
暂停队列里放了五项任务,调用 start() 后三项立即开始,size 已经从5变成2。此前等待的 onSizeLessThan(3) 却要到第一项任务完成才恢复。这样的延迟要检查等待者监听的事件,不必先怀疑 size 读取错误。
在正式 p-queue 9.0.1 中,onSizeLessThan 先检查一次条件,不满足后只等待 next 事件。正常由任务完成推动的出队会唤醒它;启动暂停队列、提高 concurrency 或速率窗口放行造成的出队,则可能没有及时触发这条重查路径。
背压等待的是待执行数量
size 只统计还没开始的任务,pending 是已经运行的任务数。size 降到2并不表示整个队列只剩两项工作,更不代表此前三项已经完成。
因此,生产者等待 onSizeLessThan 后继续加入数据,和等待所有任务结果,是不同的用途。不要改成 onIdle 来代替背压,否则生产者可能每次都等到整条队列完全空闲,改变吞吐与调度方式。
正式 9.3.2 已将该等待改为同时监听 next 和 active,覆盖启动新任务造成的出队;9.3.3 也保留此处理。可以沿兼容范围采用包含该实现的正式包,不据版本抽样推定最早修复版本。
保留任务结果和生产者边界
下面展示单个生产者的结构,queue 是现有实例,items 与 handleItem 来自业务:
const pendingResults = [];
for (const item of items) {
await queue.onSizeLessThan(100);
const result = queue.add(() => handleItem(item));
result.catch(reportTaskFailure);
pendingResults.push(result);
}
const results = await Promise.allSettled(pendingResults);
reportTaskFailure 是应用自己的错误记录函数,不能抛出新的未处理异常。allSettled 保留每项成功或失败,调用方要检查 results,不能将它的完成当成所有任务都成功。
100是示例等待数量,需按任务占用设置。这个片段针对数量有限的输入,会保留全部结果;无限或极大的数据流需要增量保存结果,不能让 pendingResults 一直增长。
多个并行生产者还可能在同一次阈值释放后一起加入。onSizeLessThan 不是一个原子的容量预留接口;需要严格控制总容量时,应让入队入口协调生产者,而不是每个生产者各自检查一次就认为获得了独占名额。
回归时分别覆盖暂停后 start、运行中提高并发、速率窗口放行,以及普通任务完成的路径。判断等待者何时恢复,也要继续等待实际任务结果。不能为了唤醒它手工发送内部 next 事件,那可能让其他监听器对一次未发生的完成作出反应。


