background-jobs

Background Jobs Skill — 后台任务系统

何时使用

• 把慢操作（发邮件 / 生成 PDF / 转码视频 / 调用第三方 API）从请求路径剥离
• 定时任务（每天清理 / 每小时同步报表）
• 工作流编排（订单 → 库存 → 支付 → 发货 多步事务）
• 重试 / 失败处理 / 死信队列
• 选择 Sidekiq vs Bull vs Celery vs Temporal

一、基本架构

[ HTTP 请求 ]
       │  enqueue (毫秒级返回)
       ▼
[ Queue ]  ← Redis / RabbitMQ / SQS / Kafka / Postgres
       │
       ▼
[ Worker(s) ]  ← 多实例并行消费
       │
       ▼
[ DB / Email / 3rd-party API ]

核心收益：

• 解耦：请求快返回（用户体验）
• 削峰：突发流量入队，worker 按能力消费
• 重试：失败自动重新尝试
• 调度：未来某时执行

二、何时不要用 background job

• ❌ 简单同步够快（< 100ms）
• ❌ 用户期望立即看到结果（用 SSE / WS 反馈进度）
• ❌ 无幂等设计（队列默认 at-least-once，会重复）
• ❌ 只跑一次的小脚本（用 cron 直接跑）

三、跨语言库选型

语言	库	后端	卖点
Ruby	Sidekiq	Redis	业界标杆 / 简单 / 高性能
Ruby	GoodJob / SolidQueue	Postgres	Rails 8 默认（无 Redis 依赖）
Python	Celery	Redis / RabbitMQ	老牌 / 配置复杂
Python	RQ	Redis	简单替代 Celery
Python	Dramatiq / arq	Redis	现代 / async
Node	BullMQ	Redis	业界标准 / TS 友好
Node	Agenda	MongoDB	时间调度强
Go	asynq	Redis	Sidekiq 风格
Go	River	Postgres	无 Redis / 事务级一致
Java	Quartz	DB / 内存	老牌 / 调度强
多语言	Temporal / Cadence	自家集群	工作流 / Saga / 长事务
云	AWS SQS + Lambda	/	全托管 / 无 worker 维护
云	GCP Cloud Tasks / Pub/Sub	/	全托管

默认推荐：

• 简单任务：Sidekiq / BullMQ / asynq
• 复杂工作流（多步骤 / 长时间 / 补偿）：Temporal
• 不想运维 Redis：River（Go）/ SolidQueue（Rails）

四、BullMQ 标准模式（Node.js）

import { Queue, Worker, QueueEvents } from 'bullmq'

const connection = { host: 'localhost', port: 6379 }

// 入队
const emailQueue = new Queue('email', { connection })
await emailQueue.add('welcome', {
  userId: '123',
  template: 'signup',
}, {
  jobId: `welcome-${userId}`,         // 幂等 key（重复 add 同 ID 不入队）
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
  removeOnComplete: { age: 3600, count: 1000 },
  removeOnFail: { age: 24 * 3600 },
})

// 延迟任务
await emailQueue.add('reminder', data, { delay: 60_000 })

// 定时任务（cron）
await emailQueue.add('daily-report', {}, {
  repeat: { pattern: '0 9 * * *', tz: 'Asia/Tokyo' },
})

// 优先级
await emailQueue.add('urgent', data, { priority: 1 })  // 越小越优先

// Worker
const worker = new Worker('email', async (job) => {
  switch (job.name) {
    case 'welcome': return sendWelcomeEmail(job.data)
    case 'reminder': return sendReminderEmail(job.data)
    case 'daily-report': return generateDailyReport()
  }
}, {
  connection,
  concurrency: 10,                    // 单 worker 并发数
  limiter: { max: 100, duration: 60_000 },  // 每分钟 100 个任务
})

worker.on('failed', (job, err) => {
  logger.error({ jobId: job?.id, err }, 'job failed')
})

// 优雅停机
process.on('SIGTERM', async () => {
  await worker.close()                // 等当前 job 处理完
  await emailQueue.close()
  process.exit(0)
})