background-job-orchestrator

后台任务编排器

擅长设计和实现生产级后台任务系统，处理长时间运行的任务，同时不阻塞 API 响应。

适用场景

✅ 适用于：

• 长时间运行的任务（发送邮件、生成报告、处理图像）
• 批量操作（批量导入、导出、数据迁移）
• 定时任务（每日摘要、清理任务、定期报告）
• 需要重试逻辑的任务（调用外部 API、不稳定的操作）
• 基于优先级的处理（优先处理高级用户的任务、关键警报）
• 受速率限制的操作（API 配额、第三方服务限制）

❌ 不适用于：

• 实时双向通信（使用 WebSockets）
• 延迟要求低于 1 秒的场景（使用内存缓存）
• 简单的延迟操作（少于 5 秒时，使用 setTimeout 即可）
• 需要同步 API 响应的场景（在请求处理程序中保留逻辑）

快速决策树

Does this task:
├── Take >5 seconds? → Background job
├── Need to retry on failure? → Background job
├── Run on a schedule? → Background job (cron pattern)
├── Block user interaction? → Background job
├── Process in batches? → Background job
└── Return immediately? → Keep synchronous

---

技术选择

Node.js：BullMQ（推荐用于 2024 年及之后）

适用场景：

• TypeScript 项目
• 技术栈中已使用 Redis
• 需要高级功能（速率限制、优先级、可重复任务）

BullMQ 相较于 Bull 的优势：

• Bull（v3）→ BullMQ（v4+）：使用 TypeScript 完全重写
• 更好的 Redis 连接处理
• 提升了并发能力和性能
• 积极维护（Bull 处于维护模式）

Python：Celery

适用场景：

• Python/Django 项目
• 需要分布式任务执行
• 复杂工作流（链、组、弦）

替代方案：

• RQ（Redis 队列）：更简单，功能较少
• Dramatiq：较新，生态系统较小
• Huey：轻量，适用于小型项目

云原生：AWS SQS、Google Cloud Tasks

适用场景：

• 无服务器架构
• 不想管理 Redis/RabbitMQ
• 需要保证消息投递和死信队列

---

常见反模式

反模式 1：没有死信队列

新手想法：“重试 3 次，然后静默失败”

问题：失败的任务会消失，既无法查看，也没有恢复途径。

正确做法：

// BullMQ with dead letter queue
const queue = new Queue('email-queue', {
  connection: redis,
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000
    },
    removeOnComplete: 100, // Keep last 100 successful
    removeOnFail: false     // Keep all failed for inspection
  }
});

// Monitor failed jobs
const failedJobs = await queue.getFailed();

时间线：

• 2020 年以前：重试后不再处理
• 2020 年起：死信队列成为标准
• 2024 年起：必须对任务失败进行可观测性监控

---

反模式 2：同步处理任务

症状：API 端点等待任务完成

问题：

// ❌ WRONG - Blocks API response
app.post('/send-email', async (req, res) => {
  await sendEmail(req.body.to, req.body.subject);
  res.json({ success: true });
});

错误原因：超时、用户体验差、浪费服务器资源

正确做法：

// ✅ RIGHT - Queue and return immediately
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'
  });
});

// Separate worker processes the job
worker.process('send', async (job) => {
  await sendEmail(job.data.to, job.data.subject);
});

---

反模式 3：没有幂等性

问题：任务运行两次 → 重复扣款、重复发送邮件

原因：

• Redis 连接在处理中断开
• Worker 在任务完成前崩溃
• 任务超时触发重试，但原任务仍在运行

正确做法：

// ✅ Idempotent job with deduplication key
await queue.add('charge-payment', {
  userId: 123,
  amount: 50.00
}, {
  jobId: `payment-${orderId}`, // Prevents duplicates
  attempts: 3
});

// In worker: Check if already processed
worker.process('charge-payment', async (job) => {
  const { userId, amount } = job.data;

  // Check idempotency
  const existing = await db.payments.findOne({
    jobId: job.id
  });
  if (existing) {
    return existing; // Already processed
  }

  // Process payment
  const result = await stripe.charges.create({...});

  // Store idempotency record
  await db.payments.create({
    jobId: job.id,
    result
  });

  return result;
});

---

反模式 4：没有速率限制

问题：让第三方 API 不堪重负，或耗尽配额

症状：Sendgrid、Stripe 等服务返回“超出速率限制”错误

正确做法：

// BullMQ rate limiting
const queue = new Queue('api-calls', {
  limiter: {
    max: 100,        // Max 100 jobs
    duration: 60000  // Per 60 seconds
  }
});

// Or: Priority-based rate limits
await queue.add('send-email', data, {
  priority: user.isPremium ? 1 : 10,
  rateLimiter: {
    max: user.isPremium ? 1000 : 100,
    duration: 3600000 // Per hour
  }
});

---

反模式 5：忘记扩展 Worker

问题：单个 Worker 无法及时处理队列中的任务

症状：队列积压，任务延迟数小时或数天

正确做法：

// Horizontal scaling with multiple workers
const worker = new Worker('email-queue', async (job) => {
  await processEmail(job.data);
}, {
  connection: redis,
  concurrency: 5  // Process 5 jobs concurrently per worker
});

// Run multiple worker processes (PM2, Kubernetes, etc.)
// Each worker processes concurrency * num_workers jobs

监控：

// Set up alerts for queue depth
setInterval(async () => {
  const waiting = await queue.getWaitingCount();
  if (waiting > 1000) {
    alert('Queue depth exceeds 1000, scale workers!');
  }
}, 60000);

---

实现模式

模式 1：邮件营销活动

// Queue setup
const emailQueue = new Queue('email-campaign', { connection: redis });

// Enqueue batch
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);
}

// Worker with retry logic
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; // Retry
    }
    // Invalid email, don't retry
    console.error(`Invalid email for user ${userId}`);
  }
}, {
  connection: redis,
  concurrency: 10
});

模式 2：定时报告

// Daily report at 9 AM
await queue.add('daily-report', {
  type: 'sales',
  recipients: ['[email protected]']
}, {
  repeat: {
    pattern: '0 9 * * *', // Cron syntax
    tz: 'America/New_York'
  }
});

// Worker generates and emails report
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: `Daily ${type} Report`,
    attachments: [{ filename: 'report.pdf', content: pdf }]
  });
});

模式 3：视频转码流水线

// Multi-stage job with progress tracking
await videoQueue.add('transcode', {
  videoId: 123,
  formats: ['720p', '1080p', '4k']
}, {
  attempts: 2,
  timeout: 3600000 // 1 hour timeout
});

worker.process('transcode', async (job) => {
  const { videoId, formats } = job.data;

  for (let i = 0; i < formats.length; i++) {
    const format = formats[i];

    // Update progress
    await job.updateProgress((i / formats.length) * 100);

    // Transcode
    await ffmpeg.transcode(videoId, format);
  }

  await job.updateProgress(100);
});

// Client polls for progress
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
  });
});

---

监控与可观测性

必备指标

// Queue health dashboard
async function getQueueMetrics() {
  const [waiting, active, completed, failed, delayed] = await Promise.all([
    queue.getWaitingCount(),
    queue.getActiveCount(),
    queue.getCompletedCount(),
    queue.getFailedCount(),
    queue.getDelayedCount()
  ]);

  return {
    waiting,    // Jobs waiting to be processed
    active,     // Jobs currently processing
    completed,  // Successfully completed
    failed,     // Failed after retries
    delayed,    // Scheduled for future
    health: waiting < 1000 && failed < 100 ? 'healthy' : 'degraded'
  };
}

BullMQ Board（界面）

// Development: Monitor jobs visually
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());
// Visit http://localhost:3000/admin/queues

---

生产环境检查清单

□ Dead letter queue configured
□ Retry strategy with exponential backoff
□ Job timeout limits set
□ Rate limiting for third-party APIs
□ Idempotency keys for critical operations
□ Worker concurrency tuned (CPU cores * 2)
□ Horizontal scaling configured (multiple workers)
□ Queue depth monitoring with alerts
□ Failed job inspection workflow
□ Job data doesn't contain PII in logs
□ Redis persistence enabled (AOF or RDB)
□ Graceful shutdown handling (SIGTERM)

---

适用与不适用场景

场景	是否使用后台任务？
用户注册时发送欢迎邮件	✅ 是 - 可能需要 2-5 秒
扣取信用卡费用	⚠️ 视情况而定 - 取决于支付服务提供商的延迟
生成 PDF 报告（30 秒）	✅ 是 - 应在后台处理
从数据库获取用户资料	❌ 否 - 耗时仅几毫秒，应保持同步
处理视频上传（5 分钟）	✅ 是 - 始终在后台处理
验证表单输入	❌ 否 - 同步验证
每日 cron 任务	✅ 是 - 使用可重复任务
实时聊天消息	❌ 否 - 使用 WebSockets

---

技术对比

特性	BullMQ	Celery	AWS SQS
语言	Node.js	Python	任意（HTTP API）
后端	Redis	Redis/RabbitMQ/SQS	托管式
优先级	✅	✅	✅
速率限制	✅	❌	✅（通过属性）
重复执行/定时任务	✅	✅（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 初始化 BullMQ
• scripts/queue_health_check.ts - 队列指标仪表板
• scripts/retry_failed_jobs.ts - 批量重试失败的任务

---

此技能适用于：后台任务实现 | 队列架构 | 重试策略 | Worker 扩缩容 | 任务可观测性