背景任務編排器 是做什麼的？
專精於設計與實作生產級的背景任務系統，能夠處理長時間運行的任務，而不阻塞 API 回應。
使用時機
✅ 適用於：
- 長時間任務（發送電子郵件、生成報告、圖片處理）
- 批次操作（大量匯入、匯出、資料遷移）
- 排程任務（每日摘要、清理任務、定期報告）
- 需要重試邏輯的任務（外部 API 呼叫、不穩定操作）
- 基於優先級的處理（高級用戶優先、關鍵警報）
- 受速率限制的操作（API 配額、第三方服務限制）
❌ 不適用於：
- 即時雙向通訊（請使用 WebSocket）
- 次毫秒延遲需求（請使用記憶體快取）
- 簡單延遲（低於 5 秒可使用 setTimeout）
- 同步 API 回應（請將邏輯保留在請求處理程式中）
快速決策樹
這個任務是否：
├── 需要超過 5 秒？ → 背景任務
├── 失敗時需要重試？ → 背景任務
├── 依照排程執行？ → 背景任務（cron 模式）
├── 會阻擋用戶互動？ → 背景任務
├── 需要批次處理？ → 背景任務
└── 需要立即回應？ → 保持同步
技術選擇
Node.js：BullMQ（2024 年推薦）
使用時機：
- TypeScript 專案
- 已使用 Redis
- 需要進階功能（速率限制、優先級、可重複任務）
為何選擇 BullMQ 而非 Bull：
- Bull (v3) → BullMQ (v4+)：使用 TypeScript 完全重寫
- 更好的 Redis 連線處理
- 改進的並發性和效能
- 積極維護（Bull 處於維護模式）
Python：Celery
使用時機：
- Python/Django 專案
- 需要分散式任務執行
- 複雜工作流程（chains、groups、chords）
替代方案：
- 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 分鐘） | ✅ 是 - 永遠使用背景任務 |
| 驗證表單輸入 | ❌ 否 - 同步驗證 |
| 每日 Cron 任務 | ✅ 是 - 使用可重複任務 |
| 即時聊天訊息 | ❌ 否 - 使用 WebSocket |
技術比較
| 功能 | BullMQ | Celery | AWS SQS |
|---|---|---|---|
| 語言 | Node.js | Python | 任意 (HTTP API) |
| 後端 | Redis | Redis/RabbitMQ/SQS | 託管 |
| 優先級 | ✅ | ✅ | ✅ |
| 速率限制 | ✅ | ❌ | ✅ (透過屬性) |
| 重複/Cron | ✅ | ✅ (celery-beat) | ❌ (使用 EventBridge) |
| UI 儀表板 | Bull Board | Flower | CloudWatch |
| 工作流程 | ❌ | ✅ (chains, groups) | ❌ |
| 學習曲線 | 中等 | 中等 | 低 |
| 成本 | Redis 託管費 | Redis 託管費 | $0.40/百萬請求 |
參考資料
/references/bullmq-patterns.md- 進階 BullMQ 模式與範例/references/celery-workflows.md- Celery chains、groups 和 chords/references/job-observability.md- 監控、警報與除錯
腳本
scripts/setup_bullmq.sh- 使用 Redis 初始化 BullMQscripts/queue_health_check.ts- 佇列指標儀表板scripts/retry_failed_jobs.ts- 大量重試失敗任務
此技能引導：背景任務實作 | 佇列架構 | 重試策略 | 工作器擴展 | 任務可觀測性