--- title: p-queue 任务队列：限速、重试与取消各自保证什么 description: 对照 ChatCrystal 的队列与状态跟踪，区分任务限速和请求限速，并用离线练习验证清空队列不会中断在途任务。 image: "https://zengliangyi.com/og/blog/pqueue-task-queues.png" --- 队列中的一个任务可能调用摘要模型、多个 Embedding 请求和关系模型。限制每秒启动一个任务，不等于限制每秒只有一个模型请求。本文沿着 ChatCrystal 的队列、重试和状态跟踪阅读，再用一个离线练习验证取消的实际边界。 ## 限制的是队列任务的启动 固定版本的 [queue/index.ts](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/server/src/queue/index.ts)使用 p-queue，[锁文件](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/pnpm-lock.yaml)解析为 9.3.3，配置核心如下。这个片段依赖现有导入，不是独立程序。 ```ts new PQueue ({ concurrency: 1 , intervalCap: 1 , interval: 1000 , carryoverIntervalCount: true , }); ``` concurrency 限制同时执行的队列任务；intervalCap 与 interval 限制时间窗口内启动的任务。carryoverIntervalCount 把跨窗口尚未结束的任务计入下一个窗口，不是累积或清除“未用完的额度”。当前配置也没有启用滚动窗口的 strict 模式。 任务函数内部可以发起多个请求，AI SDK 和应用层还各有重试。因此这份配置不能证明某个供应商的全局配额一定满足，也不会限制另一个进程、未入队入口或共用 API key 的客户端。调整并发前应先列出实际请求路径；多进程写入和存储竞争也不能从 concurrency=1 推导为不存在。 ## 外层重试在槽位内部等待 `enqueueWithRetry` 默认 maxRetries 为 3，循环包含首次尝试，最多执行四次任务函数。仅当错误消息包含 `'429'` 且还有重试机会时，等待 1000、2000、4000 毫秒后继续。等待发生在任务内部，该任务仍占用执行槽位。 这是错误文本匹配，不是结构化 HTTP 状态检查：其他错误文字也可能包含 429，真实限流错误也可能被包装成不含这个数字的消息。代码没有读取 Retry-After，也没有加入随机抖动。应把这些作为当前策略的限制，而不是称其为最可靠的错误识别方式。 外层最多四次任务执行还不等于最多四个 HTTP 请求。任务内部的 AI SDK 调用可能自行重试；多块 Embedding 会产生多次请求。格式、业务和保存错误是否重试，需要按相应入口逐项检查，参见[结构化 Prompt 的失败类型](/blog/structured-json-prompts/)。 ## 清空等待队列不会中断在途任务 在新临时目录安装 `p-queue@9.3.3`，保存以下 `lesson.mjs`，用 Node 24+ 运行。它是**教学实例**，只验证执行槽位和 clear，不使用应用的全局队列、数据库或模型，也没有测试实际每秒限速。 ```sh npm init -y npm install --ignore-scripts --no-audit --no-fund p-queue@9.3.3 node lesson.mjs ``` ```js import assert from 'node:assert/strict' ; import { setTimeout as delay } from 'node:timers/promises' ; import PQueue from 'p-queue' ; const queue = new PQueue ({ concurrency: 1 , autoStart: false }); let release; let entered; let executions = 0 ; const gate = new Promise (( resolve ) => { release = resolve; }); const started = new Promise (( resolve ) => { entered = resolve; }); const active = queue. add ( async () => { executions ++ ; entered (); await gate; return 'finished' ; }); const waiting = queue. add ( async () => { executions ++ ; return 'waiting' ; }); queue. start (); await started; assert. equal (queue.pending, 1 ); assert. equal (queue.size, 1 ); queue. clear (); assert. equal (queue.size, 0 ); assert. equal (queue.pending, 1 ); release (); assert. equal ( await active, 'finished' ); await queue. onIdle (); assert. equal (executions, 1 ); const waitingState = await Promise . race ([ waiting. then ( () => 'resolved' , () => 'rejected' ), delay ( 20 ). then (() => 'unsettled' ), ]); assert. equal (waitingState, 'unsettled' ); console. log ( 'clear removes waiting work; active finishes; removed promise stays pending' ); ``` 把 executions 的预期从 1 改成 2，程序会失败；恢复后通过。`pending` 是正在执行的数量，`size` 是等待数量。onIdle 可以在已清空队列且在途任务结束后完成，但被移除任务的 add Promise 仍不会因此完成。20 毫秒只用于观察这次构造实例，不是性能指标；长期不完成的行为依据该版本 clear 的实现。 若需要取消时结算返回的 Promise，可另外设计 AbortSignal 路径，并让运行中的工作响应信号；这不是当前 cancelQueue 已实现的能力。不能直接 await 被 clear 移除的 Promise 来等待“全部取消完成”。 ## 状态记录与执行队列是两份数据 [tracker.ts](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/server/src/queue/tracker.ts)记录 queued、processing、completed、failed，并保存开始和结束时间。cancelQueue 先清空等待队列，再把跟踪器中 queued 的记录标为 failed、错误为“已取消”，processing 继续运行。 跟踪器的 Map 按 ID 忽略重复 add，但这不自动阻止 p-queue 添加重复任务；入口处是否检查 active ID 要单独确认。没有活动任务时，它安排五秒后清空记录，因此这个快照不是持久化任务日志，重启后也不能恢复等待任务。队列和记录一致性需要同时检查。 ## SSE 传递快照，不能代替持久化 [notes.ts](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/server/src/routes/notes.ts)的 `/api/queue/stream` 每秒发送 status 事件，包含汇总、任务和已完成任务的 duration。total 大于零且没有 queued/processing 时，发送 done 并结束连接；请求断开也清理计时器。 这条路径没有用于恢复历史事件的 event ID 或持久化重放。客户端断开、任务记录五秒后清空、服务重启，都需要与正常结束分开验收。完成快照可帮助界面展示，却不能证明每次外部请求都被记录。 纯内存队列减少了当前单进程调度所需的基础设施，同时需要明确丢失、取消和退出边界。相关用法与 API 说明可参考 [p-queue 官方仓库](https://github.com/sindresorhus/p-queue)；本次练习只建立 clear 的行为证据，限速与真实模型链路仍需各自验证。 分享： ## 相关文章 [## 五种 AI 对话解析器：字段保留与损失边界 从 Claude Code、Codex、Cursor、Trae 和 Copilot 的实际解析器建立验收矩阵，检查工具标记、思考文本、顺序与快照覆盖范围。 曾亮懿 2026年10月7日 阅读约 4 分钟 #ChatCrystal #TypeScript #Node.js](/blog/ai-conversation-parser-boundaries) [## AI SDK Provider 抽象：能力表与协议边界 核对 ChatCrystal 的七项 Provider 注册、AI SDK 7 模型工厂和结构化输出，用无网络 mock 验证 Responses、Chat Completions、Embedding 与 schema 失败。 曾亮懿 2026年10月7日 阅读约 4 分钟 #ChatCrystal #Node.js #TypeScript](/blog/ai-sdk-provider-abstraction) [## SourceAdapter 内部链路：身份、内容哈希与重新导入 对照 ChatCrystal 的注册表、导入器和远端载荷，验证会话身份、内容哈希、元数据捷径与 parserVersion 的实际边界。 曾亮懿 2026年10月7日 阅读约 4 分钟 #ChatCrystal #TypeScript #Node.js](/blog/source-adapter-internals) ## 从文章到代码 到我的 GitHub 看看 ChatCrystal、开发者工具和日常学习记录，欢迎交流实现细节。 [访问我的 GitHub](https://github.com/ZengLiangYi) ```json { "@context": "https://schema.org", "@type": "WebSite", "name": "ZengLiangYi Blog", "alternateName": [ "Zeng Liangyi", "ZengLiangYi", "zengliangyi.com" ], "url": "https://zengliangyi.com", "description": "曾亮懿的个人网站。记录 AI Native 探索、开发者工具与全栈产品实践。" } { "@context": "https://schema.org", "@type": "BlogPosting", "headline": "p-queue 任务队列：限速、重试与取消各自保证什么", "description": "对照 ChatCrystal 的队列与状态跟踪，区分任务限速和请求限速，并用离线练习验证清空队列不会中断在途任务。", "url": "https://zengliangyi.com/blog/pqueue-task-queues/", "image": "https://zengliangyi.com/images/blog/pqueue-task-queues.svg", "datePublished": "2026-10-06T00:00:00.000Z", "dateModified": "2026-10-06T00:00:00.000Z", "author": { "@type": "Person", "name": "曾亮懿", "url": "https://zengliangyi.com/about" }, "publisher": { "@type": "Organization", "name": "ZengLiangYi Blog", "logo": { "@type": "ImageObject", "url": "https://zengliangyi.com/favicon-96x96.png" } }, "mainEntityOfPage": { "@type": "WebPage", "@id": "https://zengliangyi.com/blog/pqueue-task-queues/" } } { "@context": "https://schema.org", "@type": "BreadcrumbList", "itemListElement": [ { "@type": "ListItem", "position": 1, "name": "首页", "item": "https://zengliangyi.com/" }, { "@type": "ListItem", "position": 2, "name": "文章", "item": "https://zengliangyi.com/blog" }, { "@type": "ListItem", "position": 3, "name": "p-queue 任务队列：限速、重试与取消各自保证什么", "item": "https://zengliangyi.com/blog/pqueue-task-queues/" } ] } ```