17.08-死信队列

要点

重试耗尽的消息进入死信队列（DLQ）——不是丢掉，是保留现场等待排查
死信队列有三个用途：排查失败原因、手动重试、统计分析
Cloudflare Queue 原生支持 DLQ——配置 dead_letter_queue 即可
DLQ 消费者可以自动重试可恢复的错误，不可恢复的报警
死信消息需要设置过期时间——避免无限堆积

1. 为什么需要死信队列

消息重试了 3 次还是失败，怎么办？两个选择：

直接丢掉——简单粗暴，但失败原因丢失了，无法排查
放入死信队列——保留消息内容和错误信息，后续排查或手动重试

死信队列（Dead Letter Queue, DLQ）是消息系统的标准实践。它不是「垃圾堆」，而是「问题现场」——保留失败消息的完整上下文，方便事后分析。

典型场景：

外部 API 持续故障——所有消息都失败，堆积在 DLQ。API 恢复后，批量手动重试
数据格式错误——消息内容有问题，需要修复后重试
业务逻辑 bug——代码有问题，修复后重新处理 DLQ 里的消息
临时限流——API 限流导致失败，等限流窗口过去后重试

2. Cloudflare Queue 的 DLQ 配置

// wrangler.jsonc
{
  "queues": {
    "consumers": [{
      "queue": "tasks",
      "max_batch_size": 5,
      "max_retries": 3,
      "dead_letter_queue": "tasks-dlq"
    }]
  }
}

消息重试 3 次后，Cloudflare 自动把它转到 tasks-dlq。DLQ 也是一个普通的 Queue，需要配置消费者：

// 同一个 wrangler.jsonc
{
  "queues": {
    "producers": [{
      "queue": "tasks-dlq",
      "binding": "DLQ"
    }],
    "consumers": [
      {
        "queue": "tasks",
        "max_retries": 3,
        "dead_letter_queue": "tasks-dlq"
      },
      {
        "queue": "tasks-dlq",
        "max_batch_size": 10,
        "max_retries": 0  // DLQ 不重试，处理失败就丢掉
      }
    ]
  }
}

DLQ 消费者 max_retries: 0——DLQ 里的消息不再重试，避免无限循环。

3. DLQ 消费者

DLQ 消费者的职责：记录失败信息、尝试自动恢复、报警通知。

// src/consumers/dlq-consumer.ts
import type { MessageBatch, Queue } from '@cloudflare/workers-types'
 
interface DeadLetterMessage {
  taskId: string
  userId: string
  payload: unknown
  originalQueue: string
  errorMessage: string
  failedAt: string
  retryCount: number
}
 
export async function dlqConsumer(
  batch: MessageBatch<DeadLetterMessage>,
  env: {
    DB: D1Database
    DLQ: Queue<DeadLetterMessage>
  }
): Promise<void> {
  for (const msg of batch.messages) {
    const deadMsg = msg.body
 
    // 1. 记录到 D1
    await env.DB.prepare(`
      INSERT INTO dead_letters (
        id, task_id, user_id, original_queue, payload,
        error_message, failed_at, retry_count, created_at
      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    `).bind(
      crypto.randomUUID(),
      deadMsg.taskId,
      deadMsg.userId,
      deadMsg.originalQueue ?? 'tasks',
      JSON.stringify(deadMsg.payload),
      deadMsg.errorMessage,
      deadMsg.failedAt,
      deadMsg.retryCount,
      new Date().toISOString(),
    ).run()
 
    // 2. 更新任务状态
    await env.DB.prepare(`
      UPDATE tasks SET status = 'failed', error_message = ?, updated_at = ?
      WHERE id = ?
    `).bind(deadMsg.errorMessage, new Date().toISOString(), deadMsg.taskId).run()
 
    // 3. 尝试自动恢复
    const recoverable = await tryAutoRecover(deadMsg, env)
 
    if (!recoverable) {
      // 4. 不可恢复，报警
      await sendAlert(deadMsg, env)
    }
 
    msg.ack()
  }
}

自动恢复

某些错误可以自动恢复：

async function tryAutoRecover(
  msg: DeadLetterMessage,
  env: Env
): Promise<boolean> {
  const error = msg.errorMessage.toLowerCase()
 
  // 限流导致的失败——等一段时间后重试
  if (error.includes('rate limit') || error.includes('429')) {
    // 延迟 5 分钟后重新入队
    await env.TASK_QUEUE.send(
      { taskId: msg.taskId, ...msg.payload },
      { delaySeconds: 300 }
    )
    return true
  }
 
  // 临时故障——立即重试
  if (error.includes('503') || error.includes('temporarily unavailable')) {
    await env.TASK_QUEUE.send({ taskId: msg.taskId, ...msg.payload })
    return true
  }
 
  // 其他错误——不可自动恢复
  return false
}

报警通知

不可自动恢复的死信消息需要报警：

async function sendAlert(msg: DeadLetterMessage, env: Env): Promise<void> {
  // 方案一：写日志
  console.error('Dead letter alert:', {
    taskId: msg.taskId,
    error: msg.errorMessage,
    retryCount: msg.retryCount,
  })
 
  // 方案二：发通知（邮件、Slack、企业微信）
  await env.KV.put(
    `alert:dead-letter:${msg.taskId}`,
    JSON.stringify({
      taskId: msg.taskId,
      userId: msg.userId,
      error: msg.errorMessage,
      failedAt: msg.failedAt,
    }),
    { expirationTtl: 86400 * 7 }  // 保留 7 天
  )
 
  // 方案三：调 Webhook
  // await fetch(env.ALERT_WEBHOOK_URL, {
  //   method: 'POST',
  //   body: JSON.stringify({
  //     type: 'dead_letter',
  //     taskId: msg.taskId,
  //     error: msg.errorMessage,
  //   }),
  // })
}

4. 查询死信

给运维或管理员提供查询接口：

// 查询死信列表
app.get('/api/dead-letters', async (c) => {
  const page = parseInt(c.req.query('page') ?? '1')
  const pageSize = 20
  const offset = (page - 1) * pageSize
 
  const [deadLetters, countResult] = await Promise.all([
    c.env.DB.prepare(`
      SELECT id, task_id, user_id, original_queue, error_message,
             failed_at, retry_count, created_at
      FROM dead_letters
      ORDER BY created_at DESC
      LIMIT ? OFFSET ?
    `).bind(pageSize, offset).all(),
    c.env.DB.prepare(`
      SELECT COUNT(*) as total FROM dead_letters
    `).first(),
  ])
 
  return c.json({
    deadLetters: deadLetters.results,
    total: countResult?.total ?? 0,
    page,
    pageSize,
  })
})
 
// 查询单个死信详情
app.get('/api/dead-letters/:id', async (c) => {
  const deadLetter = await c.env.DB.prepare(`
    SELECT * FROM dead_letters WHERE id = ?
  `).bind(c.req.param('id')).first()
 
  if (!deadLetter) return c.json({ error: 'Not found' }, 404)
 
  return c.json({
    ...deadLetter,
    payload: JSON.parse(deadLetter.payload),
  })
})

5. 手动重试死信

死信排查清楚、问题修复后，可以手动重试：

// 手动重试单个死信
app.post('/api/dead-letters/:id/retry', async (c) => {
  const id = c.req.param('id')
 
  const deadLetter = await c.env.DB.prepare(`
    SELECT * FROM dead_letters WHERE id = ?
  `).bind(id).first()
 
  if (!deadLetter) return c.json({ error: 'Not found' }, 404)
 
  // 重新入队
  const payload = JSON.parse(deadLetter.payload)
  await c.env.TASK_QUEUE.send({
    taskId: deadLetter.task_id,
    ...payload,
  })
 
  // 更新任务状态
  await c.env.DB.prepare(`
    UPDATE tasks SET status = 'pending', retry_count = 0, error_message = NULL, updated_at = ?
    WHERE id = ?
  `).bind(new Date().toISOString(), deadLetter.task_id).run()
 
  // 从死信表删除
  await c.env.DB.prepare('DELETE FROM dead_letters WHERE id = ?').bind(id).run()
 
  return c.json({ status: 'retrying' })
})
 
// 批量重试
app.post('/api/dead-letters/batch-retry', async (c) => {
  const { ids } = await c.req.json<{ ids: string[] }>()
 
  const deadLetters = await c.env.DB.prepare(`
    SELECT * FROM dead_letters WHERE id IN (${ids.map(() => '?').join(',')})
  `).bind(...ids).all()
 
  for (const deadLetter of deadLetters.results) {
    const payload = JSON.parse(deadLetter.payload)
    await c.env.TASK_QUEUE.send({
      taskId: deadLetter.task_id,
      ...payload,
    })
 
    await c.env.DB.prepare(`
      UPDATE tasks SET status = 'pending', retry_count = 0, updated_at = ?
      WHERE id = ?
    `).bind(new Date().toISOString(), deadLetter.task_id).run()
  }
 
  await c.env.DB.prepare(`
    DELETE FROM dead_letters WHERE id IN (${ids.map(() => '?').join(',')})
  `).bind(...ids).run()
 
  return c.json({ retried: deadLetters.results.length })
})

6. 清理死信

死信不能无限堆积。设置清理策略：

// Cron 定时清理（每天凌晨）
async function cleanupDeadLetters(env: Env): Promise<void> {
  // 删除 30 天前的死信
  const cutoff = new Date()
  cutoff.setDate(cutoff.getDate() - 30)
 
  await env.DB.prepare(`
    DELETE FROM dead_letters WHERE created_at < ?
  `).bind(cutoff.toISOString()).run()
 
  // 或者：把老死信归档到 R2
  const oldDeadLetters = await env.DB.prepare(`
    SELECT * FROM dead_letters WHERE created_at < ?
  `).bind(cutoff.toISOString()).all()
 
  if (oldDeadLetters.results.length > 0) {
    const archiveKey = `dead-letters-archive/${new Date().toISOString().slice(0, 10)}.json`
    await env.BUCKET.put(archiveKey, JSON.stringify(oldDeadLetters.results, null, 2))
 
    await env.DB.prepare(`
      DELETE FROM dead_letters WHERE created_at < ?
    `).bind(cutoff.toISOString()).run()
  }
}

7. 死信统计

监控死信数量和趋势：

async function getDeadLetterStats(env: Env): Promise<{
  total: number
  last24h: number
  byError: Record<string, number>
}> {
  const [totalResult, recentResult, errorResult] = await Promise.all([
    env.DB.prepare('SELECT COUNT(*) as count FROM dead_letters').first(),
    env.DB.prepare(`
      SELECT COUNT(*) as count FROM dead_letters
      WHERE created_at > datetime('now', '-1 day')
    `).first(),
    env.DB.prepare(`
      SELECT error_message, COUNT(*) as count
      FROM dead_letters
      GROUP BY error_message
      ORDER BY count DESC
      LIMIT 10
    `).all(),
  ])
 
  const byError: Record<string, number> = {}
  for (const row of errorResult.results) {
    byError[row.error_message] = row.count
  }
 
  return {
    total: totalResult?.count ?? 0,
    last24h: recentResult?.count ?? 0,
    byError,
  }
}

死信数量突然升高说明系统有问题——外部 API 故障、数据质量问题、代码 bug。

8. D1 表设计

CREATE TABLE dead_letters (
  id TEXT PRIMARY KEY,
  task_id TEXT NOT NULL,
  user_id TEXT NOT NULL,
  original_queue TEXT NOT NULL DEFAULT 'tasks',
  payload TEXT NOT NULL,           -- 原始消息内容（JSON）
  error_message TEXT NOT NULL,     -- 最后一次失败的错误信息
  failed_at TEXT NOT NULL,         -- 失败时间
  retry_count INTEGER NOT NULL,    -- 重试次数
  created_at TEXT NOT NULL         -- 进入 DLQ 的时间
);
 
CREATE INDEX idx_dead_letters_created ON dead_letters(created_at DESC);
CREATE INDEX idx_dead_letters_task ON dead_letters(task_id);

总结

回顾这一节的要点：

死信队列保留重试耗尽的消息——不是丢掉，是保留现场等待排查
Cloudflare Queue 配置 dead_letter_queue 即可启用
DLQ 消费者记录失败信息、尝试自动恢复、不可恢复的报警
限流、临时故障可以自动恢复——延迟后重新入队
数据错误、权限问题不可自动恢复——需要人工排查
提供查询和手动重试接口——修复后可以重新处理
定期清理死信——30 天归档或删除，避免无限堆积
监控死信数量和趋势——突然升高说明系统有问题

下一篇讲长任务进度查询——前端怎么知道任务跑到哪了。