《TypeScript编程实战》9.2 重试、幂等与死信

await emailQueue.add('welcome', payload, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 1000 },
  removeOnComplete: { count: 1000 },
  removeOnFail: false, // 失败的要留着，死信处理要用
});

const withJitter = (delay: number) => delay * (0.5 + Math.random());

await queue.add(name, data, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 1000 },
  // 自定义策略需要注册在 Worker 上，见下
});

// worker 侧注册自定义退避策略
new Worker('tasks', processor, {
  connection,
  settings: {
    backoffStrategy: (attemptsMade, type, err) => {
      if (type !== 'jitter') return -1; // 不处理则交回内置策略
      return withJitter(1000 * 2 ** (attemptsMade - 1));
    },
  },
});

import { UnrecoverableError, type Job } from 'bullmq';
import { ZodError } from 'zod';

async function processEmail(job: Job<EmailPayload, EmailResult>) {
  const parsed = EmailPayloadSchema.safeParse(job.data);
  if (!parsed.success) {
    // 数据错了，重试无意义 —— 直接判死
    throw new UnrecoverableError(`invalid payload: ${parsed.error.message}`);
  }

  const res = await mailer.send(parsed.data);
  if (res.status === 429) {
    throw new Error('rate limited'); // 可重试
  }
  return { messageId: res.id };
}

worker 取出 job → 执行 processor → 写回 completed
                                  ↑
                          进程在这里崩溃

await enqueue('order:charge', { orderId, idempotencyKey: `charge:${orderId}` });

import { redis } from './redis';

export async function once<T>(key: string, ttlSec: number, fn: () => Promise<T>): Promise<T | null> {
  const acquired = await redis.set(key, '1', 'EX', ttlSec, 'NX');
  if (!acquired) {
    return null; // 已经执行过，直接跳过
  }
  try {
    return await fn();
  } catch (err) {
    await redis.del(key); // 执行失败要释放，否则重试会被误判为重复
    throw err;
  }
}

new Worker('tasks', async (job) => {
  const { orderId } = job.data;
  return once(`charge:${orderId}`, 86400, async () => {
    return payments.charge(orderId);
  });
});

await prisma.idempotencyRecord.create({
  data: { key: `charge:${orderId}`, jobId: job.id! },
});

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

const tasks = new Queue('tasks', { connection });
const dlq = new Queue('tasks:dlq', { connection });

// 监听失败事件，把耗尽重试的 job 搬进死信队列
const events = new QueueEvents('tasks', { connection });

events.on('failed', async ({ jobId, failedReason }) => {
  const job = await tasks.getJob(jobId);
  if (!job) return;

  const maxAttempts = job.opts.attempts ?? 1;
  if (job.attemptsMade < maxAttempts) return; // 还会重试，先不搬

  await dlq.add(job.name, {
    originalJobId: jobId,
    payload: job.data,
    reason: failedReason,
    failedAt: new Date().toISOString(),
    attemptsMade: job.attemptsMade,
  });
});

// 死信积压告警
setInterval(async () => {
  const waiting = await dlq.getWaitingCount();
  if (waiting > 100) {
    logger.error({ waiting }, 'DLQ backlog too high');
    await alerting.notify('#oncall', `死信队列积压 ${waiting} 条`);
  }
}, 60_000);

// 重放脚本：修好数据后把死信重新投回原队列
import { Queue } from 'bullmq';

const dlq = new Queue('tasks:dlq', { connection });
const tasks = new Queue('tasks', { connection });

const jobs = await dlq.getJobs(['waiting', 'delayed'], 0, 100);
for (const job of jobs) {
  const { payload } = job.data as { payload: unknown };
  await tasks.add(job.name, payload, { attempts: 5 });
  await job.remove();
}
console.log(`replayed ${jobs.length} jobs`);

$ node scripts/replay-dlq.ts
replayed 37 jobs