BullMQ 后台任务队列实战

app.post('/api/export', async (req, res) => {
  await buildBigReport(req.body.userId);   // 可能要跑 2 分钟
  res.json({ ok: true });                 // 客户端等到超时
});

bull:mail:wait       待执行任务列表
bull:mail:active     正在执行的任务
bull:mail:delayed    延迟/定时任务
bull:mail:failed     失败任务
bull:mail:completed  已完成任务
bull:mail:repeat     重复任务定义

npm i bullmq ioredis

import { Queue, Worker, QueueEvents } from 'bullmq';
import { Redis } from 'ioredis';

const connection = new Redis({ host: '127.0.0.1', port: 6379, maxRetriesPerRequest: null });

Job 名称     → 决定由哪个处理器消费
Job 数据     → payload，序列化成 JSON
Job 选项     → 优先级、延迟、重试、超时

const queue = new Queue('mail', { connection });

// 最简单的入队
await queue.add('send-welcome', { userId: 123, template: 'welcome' });

await queue.add('send-batch', { fileId: 'abc' }, {
  priority: 5,            // 数字越小越先执行
  delay: 60 * 1000,       // 延迟 1 分钟执行
  attempts: 3,            // 失败重试 3 次
  removeOnComplete: true, // 完成后自动清理
  removeOnFail: 100,      // 失败最多保留 100 条
});

const worker = new Worker('mail', async (job) => {
  const { userId, template } = job.data;
  await sendMail(userId, template);
  return { sent: userId };   // 返回值写入 job.returnvalue
}, { connection, concurrency: 5 });

worker.on('completed', (job) => console.log(`完成 ${job.id}`));
worker.on('failed', (job, err) => console.error(`失败 ${job.id}: ${err.message}`));

await queue.add('remind', { orderId: 7 }, { delay: 30 * 60 * 1000 });

const repeatable = await queue.add('daily-report', { type: 'sales' }, {
  repeat: {
    pattern: '0 8 * * *',       // 每天 8 点，cron 表达式
    tz: 'Asia/Shanghai',
  },
  jobId: 'daily-report',        // 固定 jobId 保证幂等
});

// 停止重复
await queue.removeRepeatable('daily-report', {
  pattern: '0 8 * * *', tz: 'Asia/Shanghai',
});

需要「精确到秒、动态跳过」 → 自定义循环任务（setTimeout + 重新入队）
固定周期执行               → repeat cron 足够

await queue.add('convert-video', { src: 'a.mp4' }, {
  attempts: 5,              // 含首次，共尝试 5 次
  backoff: {
    type: 'exponential',    // 指数退避
    delay: 1000,            // 首次等待 1 秒，之后 2 倍递增
  },
  jobId: 'video-a',         // 幂等：同 ID 不重复入队
});

class PermanentError extends Error {
  constructor(message) { super(message); this.permanent = true; }
}

const worker = new Worker('convert-video', async (job) => {
  const src = job.data.src;
  if (!await fileExists(src)) throw new PermanentError('源文件不存在');
  await convert(src);
});

worker.on('failed', async (job, err) => {
  if (err.permanent) await job.remove();   // 永久错误，移除不再重试
});

const worker = new Worker('mail', handler, {
  stalledInterval: 30 * 1000,   // 每 30 秒检查停滞
  lockDuration: 60 * 1000,      // 任务锁时长，超时判定停滞
});

// 单 Worker 内并发
const worker = new Worker('image', handler, { concurrency: 10 });

// 横向扩展：多 Worker 进程共享同一队列
// 每个进程 new Worker(...)，Redis 负责分发

// 处理器文件 processor.ts
export default async function (job) {
  if (job.data.malicious) process.exit(1);   // 只会杀沙箱，不影响主进程
  return heavyWork(job.data);
}

import { Job } from 'bullmq';
const worker = new Worker('sandboxed', new Job.SandboxedJobProcessor('processor.ts'), {
  useWorkerThreads: true,   // 用 worker_threads 隔离
});

const events = new QueueEvents('mail', { connection });

events.on('completed', ({ jobId, returnvalue }) => {
  console.log(`job ${jobId} 完成，结果：${JSON.stringify(returnvalue)}`);
});
events.on('failed', ({ jobId, failedReason }) => {
  console.error(`job ${jobId} 失败：${failedReason}`);
});
events.on('progress', ({ jobId, data }) => {
  console.log(`job ${jobId} 进度 ${data}%`);
});

await job.updateProgress(50);   // 触发 progress 事件

const counts = await queue.getJobCounts('wait', 'active', 'delayed', 'failed');
console.log(counts);   // { wait: 12, active: 3, delayed: 1, failed: 2 }

await queue.obliterate({ force: true });   // 清空队列（含正在执行的）
await queue.clean(3600 * 1000, 'completed'); // 清理 1 小时前的完成记录