后台任务编排器
精通设计和实现生产级后台作业系统，能处理长时间运行的任务而不阻塞 API 响应。
使用时机
✅ 适用于：
- 长时间运行的任务（邮件发送、报告生成、图片处理）
- 批量操作（批量导入、导出、数据迁移）
- 定时任务（每日摘要、清理作业、定期报告）
- 需要重试逻辑的任务（外部 API 调用、不稳定的操作）
- 基于优先级的处理（高级用户优先、关键告警）
- 限制速率的操作（API 配额、第三方服务限制）
❌ 不适用于：
- 实时双向通信（使用 WebSockets）
- 亚秒级延迟要求（使用内存缓存）
- 简单的延时（小于 5 秒可使用 setTimeout）
- 同步 API 响应（将逻辑保留在请求处理器中）
快速决策树
此任务：
├── 耗时超过 5 秒？→ 后台作业
├── 失败时需要重试？→ 后台作业
├── 需要按计划执行？→ 后台作业（cron 模式）
├── 会阻塞用户交互？→ 后台作业
├── 需要批量处理？→ 后台作业
└── 需要立即返回？→ 保持同步
技术选型
Node.js：BullMQ（2024 年及以后推荐）
使用场景：
- TypeScript 项目
- 已在技术栈中使用 Redis
- 需要高级功能（速率限制、优先级、可重复作业）
为什么选择 BullMQ 而非 Bull：
- Bull (v3) → BullMQ (v4+)：完全用 TypeScript 重写
- 更好的 Redis 连接处理
- 改进的并发性和性能
- 持续维护（Bull 已进入维护模式）
Python：Celery
使用场景：
- Python/Django 项目
- 需要分布式任务执行
- 复杂工作流（链、组、和弦）
替代方案：
- RQ（Redis Queue）：更简单，功能更少
- Dramatiq：现代，生态系统较小
- Huey：轻量级，适合小型项目
云原生：AWS SQS、Google Cloud Tasks
使用场景：
- 无服务器架构
- 不想管理 Redis/RabbitMQ
- 需要确保送达和死信队列
常见反模式
反模式 1：没有死信队列
新手思维：“重试 3 次，然后静默失败”
问题：失败的作业消失了，没有可见性或恢复路径。
正确做法：
// BullMQ 带死信队列
const queue = new Queue('email-queue', {
connection: redis,
defaultJobOptions: {
attempts: 3,
backoff: {
type: 'exponential',
delay: 2000
},
removeOnComplete: 100, // 保留最近 100 个成功的
removeOnFail: false // 保留所有失败的以供检查
}
});
// 监控失败的作业
const failedJobs = await queue.getFailed();
时间线：
- 2020 年前：重试并遗忘
- 2020 年起：死信队列成为标准
- 2024 年起：作业失败的可观测性成为必需
反模式 2：同步处理作业
症状：API 端点等待作业完成
问题：
// ❌ 错误 - 阻塞 API 响应
app.post('/send-email', async (req, res) => {
await sendEmail(req.body.to, req.body.subject);
res.json({ success: true });
});
为什么错误：超时、用户体验差、浪费服务器资源
正确做法：
// ✅ 正确 - 入队后立即返回
app.post('/send-email', async (req, res) => {
const job = await emailQueue.add('send', {
to: req.body.to,
subject: req.body.subject
});
res.json({
success: true,
jobId: job.id,
status: 'queued'
});
});
// 独立的工作进程处理作业
worker.process('send', async (job) => {
await sendEmail(job.data.to, job.data.subject);
});
反模式 3：没有幂等性
问题：作业运行两次 → 重复扣费、重复发送邮件
原因：
- Redis 连接在处理过程中断开
- 工作进程在作业完成前崩溃
- 作业超时触发重试，但原作业仍在运行
正确做法：
// ✅ 带去重键的幂等作业
await queue.add('charge-payment', {
userId: 123,
amount: 50.00
}, {
jobId: `payment-${orderId}`, // 防止重复
attempts: 3
});
// 在工作进程中：检查是否已处理
worker.process('charge-payment', async (job) => {
const { userId, amount } = job.data;
// 检查幂等性
const existing = await db.payments.findOne({
jobId: job.id
});
if (existing) {
return existing; // 已处理
}
// 处理支付
const result = await stripe.charges.create({...});
// 存储幂等记录
await db.payments.create({
jobId: job.id,
result
});
return result;
});
反模式 4：没有速率限制
问题：压垮第三方 API 或耗尽配额
症状：来自 Sendgrid、Stripe 等的“速率限制超出”错误
正确做法：
// BullMQ 速率限制
const queue = new Queue('api-calls', {
limiter: {
max: 100, // 最多 100 个作业
duration: 60000 // 每 60 秒
}
});
// 或者：基于优先级的速率限制
await queue.add('send-email', data, {
priority: user.isPremium ? 1 : 10,
rateLimiter: {
max: user.isPremium ? 1000 : 100,
duration: 3600000 // 每小时
}
});
反模式 5：忘记工作进程扩展
问题：单个工作进程无法跟上队列深度
症状：队列积压，作业延迟数小时/天
正确做法：
// 水平扩展，多个工作进程
const worker = new Worker('email-queue', async (job) => {
await processEmail(job.data);
}, {
connection: redis,
concurrency: 5 // 每个工作进程并发处理 5 个作业
});
// 运行多个工作进程（PM2、Kubernetes 等）
// 每个工作进程处理 concurrency * num_workers 个作业
监控：
// 设置队列深度告警
setInterval(async () => {
const waiting = await queue.getWaitingCount();
if (waiting > 1000) {
alert('队列深度超过 1000，请扩展工作进程！');
}
}, 60000);
实现模式
模式 1：邮件营销
// 队列设置
const emailQueue = new Queue('email-campaign', { connection: redis });
// 批量入队
async function sendCampaign(userIds: number[], template: string) {
const jobs = userIds.map(userId => ({
name: 'send',
data: { userId, template },
opts: {
attempts: 3,
backoff: { type: 'exponential', delay: 5000 }
}
}));
await emailQueue.addBulk(jobs);
}
// 带重试逻辑的工作进程
const worker = new Worker('email-campaign', async (job) => {
const { userId, template } = job.data;
const user = await db.users.findById(userId);
const email = renderTemplate(template, user);
try {
await sendgrid.send({
to: user.email,
subject: email.subject,
html: email.body
});
} catch (error) {
if (error.code === 'ECONNREFUSED') {
throw error; // 重试
}
// 无效邮箱，不重试
console.error(`用户 ${userId} 的邮箱无效`);
}
}, {
connection: redis,
concurrency: 10
});
模式 2：定时报表
// 每天上午 9 点的日报
await queue.add('daily-report', {
type: 'sales',
recipients: ['[email protected]']
}, {
repeat: {
pattern: '0 9 * * *', // Cron 语法
tz: 'America/New_York'
}
});
// 工作进程生成并通过邮件发送报表
worker.process('daily-report', async (job) => {
const { type, recipients } = job.data;
const data = await generateReport(type);
const pdf = await createPDF(data);
await emailQueue.add('send', {
to: recipients,
subject: `每日 ${type} 报表`,
attachments: [{ filename: 'report.pdf', content: pdf }]
});
});
模式 3：视频转码流水线
// 多阶段作业，带进度跟踪
await videoQueue.add('transcode', {
videoId: 123,
formats: ['720p', '1080p', '4k']
}, {
attempts: 2,
timeout: 3600000 // 1 小时超时
});
worker.process('transcode', async (job) => {
const { videoId, formats } = job.data;
for (let i = 0; i < formats.length; i++) {
const format = formats[i];
// 更新进度
await job.updateProgress((i / formats.length) * 100);
// 转码
await ffmpeg.transcode(videoId, format);
}
await job.updateProgress(100);
});
// 客户端轮询进度
app.get('/videos/:id/status', async (req, res) => {
const job = await queue.getJob(req.params.jobId);
res.json({
state: await job.getState(),
progress: job.progress
});
});
监控与可观测性
关键指标
// 队列健康仪表盘
async function getQueueMetrics() {
const [waiting, active, completed, failed, delayed] = await Promise.all([
queue.getWaitingCount(),
queue.getActiveCount(),
queue.getCompletedCount(),
queue.getFailedCount(),
queue.getDelayedCount()
]);
return {
waiting, // 等待处理的作业
active, // 正在处理的作业
completed, // 成功完成的
failed, // 重试后失败的
delayed, // 计划在将来执行的
health: waiting < 1000 && failed < 100 ? 'healthy' : 'degraded'
};
}
BullMQ Board (UI)
// 开发环境：可视化监控作业
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';
const serverAdapter = new ExpressAdapter();
createBullBoard({
queues: [
new BullMQAdapter(emailQueue),
new BullMQAdapter(videoQueue)
],
serverAdapter
});
app.use('/admin/queues', serverAdapter.getRouter());
// 访问 http://localhost:3000/admin/queues
生产清单
□ 已配置死信队列
□ 带指数退避的重试策略
□ 设置作业超时限制
□ 第三方 API 的速率限制
□ 关键操作的幂等键
□ 工作进程并发数已调优（CPU 核心数 * 2）
□ 已配置水平扩展（多个工作进程）
□ 带告警的队列深度监控
□ 已失败的作业检查工作流
□ 作业数据不包含日志中的 PII
□ Redis 持久化已启用（AOF 或 RDB）
□ 优雅关闭处理（SIGTERM）
使用 vs 避免 场景
| 场景 | 使用后台作业？ |
|---|---|
| 注册时发送欢迎邮件 | ✅ 是 - 可能需要 2-5 秒 |
| 信用卡扣款 | ⚠️ 可能 - 取决于支付提供商延迟 |
| 生成 PDF 报告（30 秒） | ✅ 是 - 肯定是后台 |
| 从数据库获取用户资料 | ❌ 否 - 毫秒级，保持同步 |
| 处理视频上传（5 分钟） | ✅ 是 - 始终后台 |
| 验证表单输入 | ❌ 否 - 同步验证 |
| 每日定时任务 | ✅ 是 - 使用可重复作业 |
| 实时聊天消息 | ❌ 否 - 使用 WebSockets |
技术对比
| 特性 | BullMQ | Celery | AWS SQS |
|---|---|---|---|
| 语言 | Node.js | Python | 任意（HTTP API） |
| 后端 | Redis | Redis/RabbitMQ/SQS | 托管 |
| 优先级 | ✅ | ✅ | ✅ |
| 速率限制 | ✅ | ❌ | ✅（通过属性） |
| 重复/Cron | ✅ | ✅ (celery-beat) | ❌（使用 EventBridge） |
| UI 仪表盘 | Bull Board | Flower | CloudWatch |
| 工作流 | ❌ | ✅（链、组） | ❌ |
| 学习曲线 | 中等 | 中等 | 低 |
| 成本 | Redis 托管费用 | Redis 托管费用 | $0.40/百万请求 |
参考资料
/references/bullmq-patterns.md- 高级 BullMQ 模式与示例/references/celery-workflows.md- Celery 链、组与和弦/references/job-observability.md- 监控、告警与调试
脚本
scripts/setup_bullmq.sh- 使用 Redis 初始化 BullMQscripts/queue_health_check.ts- 队列指标仪表盘scripts/retry_failed_jobs.ts- 批量重试失败的作业
本技能指导：后台作业实现 | 队列架构 | 重试策略 | 工作进程扩展 | 作业可观测性