任务队列设计：p-queue 限速 + 重试策略

本文面向：想了解如何用任务队列控制并发和失败重试的开发者。 预计阅读时间：10 分钟 最终效果：掌握 p-queue 限速配置、指数退避重试、TaskTracker 状态机、SSE 实时推送的完整设计。

为什么需要任务队列

ChatCrystal 的核心功能之一是将 AI 对话交给 LLM 生成结构化摘要。这个过程看似简单，实际操作中却面临一个关键约束：LLM API 有速率限制。

当你批量导入 100 条对话并点击「全部摘要」时，如果同时发出 100 个请求，API 会立即返回 429 Too Many Requests。即使你用的是本地 Ollama，GPU 显存有限，并发推理也会导致超时或 OOM。

我们需要一个机制来：

这就是任务队列存在的意义。

p-queue 简介

ChatCrystal 选择了 p-queue 作为队列引擎。它是 Sindre Sorhus 编写的轻量级库，核心特点：

• Promise-based – 不依赖消息中间件（Redis、RabbitMQ），纯内存队列
• TypeScript 原生 – 类型完整
• 配置灵活 – 支持并发数、速率限制、超时等参数
• API 简洁 – queue.add(fn) 即可入队

对于单进程 Node.js 应用来说，p-queue 是最合适的方案。它不需要额外基础设施，零运维成本。

限速配置

ChatCrystal 的队列定义非常精炼：

import PQueue from 'p-queue';

export const summarizeQueue = new PQueue({
concurrency: 1,
intervalCap: 1,
interval: 1000,
carryoverConcurrencyCount: true,
});

四个参数的含义：

最终效果：每秒最多执行 1 个任务，严格串行。

carryoverConcurrencyCount: true 是一个容易忽略的细节。如果设为 false，当队列空闲一段时间后突然涌入一批任务，会因为累积了大量未使用的配额而瞬间爆发并发。设为 true 可以保证速率始终平滑。

重试策略：指数退避

即使有了限速，LLM API 仍然可能返回 429。原因很多：共享 API Key 的全局额度、同机其他进程也在调用、API 端的突发保护等。ChatCrystal 的做法是只对 429 错误做指数退避重试：

async function enqueueWithRetry<T>(
taskId: string,
taskTitle: string,
fn: () => Promise<T>,
maxRetries = 3,
): Promise<T> {
taskTracker.add(taskId, taskTitle);

const result = await summarizeQueue.add(async () => {
taskTracker.start(taskId);
let lastError: Error | undefined;
for (let attempt = 0; attempt <= maxRetries; attempt++) {
try {
const res = await fn();
taskTracker.complete(taskId);
return res;
} catch (err) {
lastError = err instanceof Error ? err : new Error(String(err));
if (lastError.message.includes('429') && attempt < maxRetries) {
const delay = 1000 * 2 ** attempt;
await new Promise(r => setTimeout(r, delay));
continue;
}
taskTracker.fail(taskId, lastError.message);
throw lastError;
}
}
taskTracker.fail(taskId, lastError?.message || 'Unknown error');
throw lastError;
});
return result as T;
}

退避延迟的计算方式是 1000 * 2^attempt：

这个设计有几个值得注意的点：

任务跟踪器：状态机

队列负责调度，但前端还需要知道每个任务的实时状态。ChatCrystal 设计了一个 TaskTracker 类来管理任务的生命周期：

queued → processing → completed
→ failed

每个任务是一个 TaskEntry：

interface TaskEntry {
id: string; // 任务唯一标识
title: string; // 人类可读的标题，如 "项目A / 对话摘要"
status: 'queued' | 'processing' | 'completed' | 'failed';
error?: string; // 失败时的错误信息
addedAt: number; // 入队时间戳
startedAt?: number; // 开始执行时间戳
finishedAt?: number; // 完成/失败时间戳
}

状态转换由 enqueueWithRetry 在不同阶段调用 taskTracker 的方法驱动：

• taskTracker.add() – 任务入队，状态设为 queued
• taskTracker.start() – 任务开始执行，状态变为 processing
• taskTracker.complete() – 任务成功，状态变为 completed
• taskTracker.fail() – 任务失败，状态变为 failed，记录错误信息

有一个设计细节值得注意：TaskTracker 内部有一个自动重置机制。当所有任务都完成后（没有 queued 或 processing 的任务），它会启动一个 5 秒的定时器，之后清空所有任务记录。这样做的好处是队列状态页面不会显示一堆已经完成的历史任务，保持界面干净。

private scheduleResetIfIdle() {
if (this.hasActiveTasks()) return;
this.cancelReset();
this.resetTimer = setTimeout(() => {
this.tasks.clear();
this.resetTimer = null;
}, 5000);
}

取消队列

用户可能在批量摘要进行到一半时想取消剩余任务。ChatCrystal 提供了 cancelQueue 函数：

export function cancelQueue() {
summarizeQueue.clear(); // 从 p-queue 中移除待执行的任务
const cancelled = taskTracker.cancelQueued(); // 将这些任务标记为 failed
return { cancelled: cancelled.length, queue: getQueueStatus() };
}

两步操作缺一不可：

• summarizeQueue.clear() 只清除 p-queue 内部的待执行队列，不影响 TaskTracker 中的状态记录
• taskTracker.cancelQueued() 遍历所有状态为 queued 的任务，将其标记为 failed，错误信息设为「已取消」

注意：已经在执行中的任务（processing 状态）不会被取消。强行中断一个正在进行的 LLM 调用既不安全也没有必要，让它自然结束即可。

SSE 实时状态推送

队列状态需要实时推送到前端。ChatCrystal 使用 Server-Sent Events（SSE）实现：

app.get('/api/queue/stream', async (_req, reply) => {
reply.raw.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
});

const interval = setInterval(() => {
const snapshot = getQueueStatus();
reply.raw.write(`event: status\\ndata: ${JSON.stringify(snapshot)}\\n\\n`);
}, 1000);

// 客户端断开时清理定时器
_req.raw.on('close', () => clearInterval(interval));
});

选择 SSE 而不是 WebSocket 的原因：

• SSE 是单向通信（服务器到客户端），队列状态推送恰好只需要这个方向
• SSE 基于 HTTP，不需要额外的协议升级
• 浏览器原生支持 EventSource API，自动重连
• 对于这种低频（每秒一次）的状态推送场景，SSE 比 WebSocket 更轻量

前端通过 EventSource 监听：

const source = new EventSource('/api/queue/stream');
source.addEventListener('status', (event) => {
const data = JSON.parse(event.data);
// data.tasks 包含所有任务的状态列表
// data.total / data.completed / data.failed 提供汇总统计
});

使用场景

ChatCrystal 中有三个地方使用了任务队列：

1. 摘要生成

最常见的场景。用户导入对话后，调用 LLM 生成结构化摘要（标题、总结、关键结论、代码片段、标签）。

enqueueWithRetry(id, title, () => triggerSummarize(id))

2. Embedding 生成

摘要生成完成后，自动将笔记文本切块并生成向量嵌入，存入 vectra 索引用于语义搜索。

enqueueWithRetry(`embed-${noteId}`, `Embedding #${noteId}`, () => generateEmbeddings(noteId))

3. 关系发现

使用 LLM 分析笔记之间的语义关系，构建知识图谱。

enqueueWithRetry(`relations-${noteId}`, `Relations: ${title}`, () => discoverRelations(noteId))

三种任务都涉及 LLM 调用，都面临速率限制问题，因此共享同一个队列。

设计决策：为什么 concurrency = 1

你可能会问：为什么不用更高的并发数？比如 concurrency = 3，速度不是更快吗？

原因有三：

如果未来需要支持更高吞吐，只需要调整 concurrency 和 intervalCap 两个参数。p-queue 的设计使得这种调整不会影响其他代码。

总结

ChatCrystal 的任务队列设计遵循了几个原则：

• 最小依赖 – p-queue 是纯内存队列，不需要 Redis 或其他外部服务
• 精确限速 – concurrency + intervalCap 双重控制，防止 API 过载
• 智能重试 – 只对 429 做指数退避，其他错误快速失败
• 状态透明 – TaskTracker + SSE 让前端实时掌握任务进展
• 用户可控 – 支持取消队列，不会让用户陷入无法中断的等待

对于学者来说，这个设计是一个很好的学习案例。它展示了如何用最少的代码解决「并发控制 + 失败恢复 + 状态同步」这三个经典问题。当你在自己的项目中遇到类似的批量任务处理需求时，可以参考这个模式。

项目地址：github.com/ZengLiangYi/ChatCrystal

如有疑问欢迎在 GitHub Issues 或私信交流，很乐意解答。