17.07-任务重试机制

要点

分布式系统里失败是常态——网络抖动、API 限流、服务临时不可用
不是所有错误都值得重试——格式错误重试 100 次也没用，网络超时值得重试
重试间隔不能固定——固定间隔会导致雪崩，指数退避更合理
最大重试次数要设上限——无限重试会浪费资源，3-5 次通常足够
Cloudflare Queue 内置重试支持——msg.retry({ delaySeconds })
重试要配合幂等性——同一条消息处理两次，结果应该一样

1. 为什么要重试

异步任务调外部 API 是常态——LLM API、OCR API、向量数据库。这些外部服务不可控：

网络超时——请求发出去了，但没收到响应
服务限流——429 Too Many Requests
临时故障——500 Internal Server Error，但几分钟后恢复
连接中断——处理到一半，连接断了

这些错误通常是暂时的。如果立即失败，用户需要手动重试。自动重试可以在故障恢复后自动完成。

但重试也有风险：

无限重试——浪费资源，任务永远完不成
立即重试——外部服务还没恢复，重试也是失败
所有任务同时重试——雪崩效应，打垮外部服务

2. 哪些错误值得重试

重试前先判断错误类型。不是所有错误都值得重试。

type RetryDecision =
  | { retry: true; delaySeconds: number }
  | { retry: false; reason: string }
 
function decideRetry(error: unknown, attemptCount: number): RetryDecision {
  const maxRetries = 3
  if (attemptCount >= maxRetries) {
    return { retry: false, reason: `超过最大重试次数 (${maxRetries})` }
  }
 
  const message = error instanceof Error ? error.message : String(error)
 
  // 可重试的错误
  if (/timeout|ETIMEDOUT|ECONNRESET/i.test(message)) {
    return { retry: true, delaySeconds: 30 * (attemptCount + 1) }
  }
  if (/429|rate limit/i.test(message)) {
    return { retry: true, delaySeconds: 60 * (attemptCount + 1) }
  }
  if (/500|502|503|504|temporarily unavailable/i.test(message)) {
    return { retry: true, delaySeconds: 30 * (attemptCount + 1) }
  }
 
  // 不可重试的错误
  if (/400|401|403|404|invalid format|unsupported/i.test(message)) {
    return { retry: false, reason: `客户端错误: ${message}` }
  }
 
  // 默认重试
  return { retry: true, delaySeconds: 30 * (attemptCount + 1) }
}

判断规则：

网络错误（timeout、connection reset）——重试
限流（429、rate limit）——重试，间隔更长
服务端错误（5xx）——重试
客户端错误（4xx、格式错误、权限不足）——不重试
未知错误——默认重试

3. 指数退避

重试间隔不能固定。假设外部服务故障了，100 个任务同时失败，如果都 1 秒后重试：

第 1 秒：100 个任务重试
外部服务还没恢复，100 个全部失败
第 2 秒：100 个任务又重试
外部服务刚恢复，被 100 个请求打垮，又故障

这就是「重试雪崩」。指数退避避免这个问题——每次重试间隔翻倍。

// 指数退避 + 抖动
function calculateDelay(attempt: number, baseDelay = 30): number {
  // 指数增长：30s, 60s, 120s, 240s...
  const exponential = baseDelay * Math.pow(2, attempt)
 
  // 加随机抖动（±25%），避免所有任务同时重试
  const jitter = exponential * 0.25 * (Math.random() * 2 - 1)
 
  return Math.round(exponential + jitter)
}
 
// 示例
console.log(calculateDelay(0))  // ~30 秒
console.log(calculateDelay(1))  // ~60 秒
console.log(calculateDelay(2))  // ~120 秒

抖动（jitter）的作用：如果 100 个任务都在第 30 秒重试，外部服务会被打垮。加 ±25% 的抖动后，重试时间分散在 22.5-37.5 秒之间，压力更均匀。

4. Cloudflare Queue 的重试

Cloudflare Queue 内置重试支持。消费者处理失败时，调 msg.retry() 或抛异常。

export async function consumer(
  batch: MessageBatch<TaskMessage>,
  env: Env
): Promise<void> {
  for (const msg of batch.messages) {
    try {
      await processTask(msg.body, env)
      msg.ack()  // 成功，确认
    } catch (err) {
      const decision = decideRetry(err, msg.attempts - 1)
 
      if (decision.retry) {
        // 延迟重试
        msg.retry({ delaySeconds: decision.delaySeconds })
      } else {
        // 不可重试，标记失败
        await updateTaskStatus(env.DB, msg.body.taskId, 'failed', {
          errorMessage: decision.reason,
        })
        msg.ack()  // 确认（不再重试）
      }
    }
  }
}

msg.attempts 是当前尝试次数（从 1 开始）。第一次处理时 attempts = 1，第一次重试时 attempts = 2。

配置最大重试次数

在 wrangler.jsonc 中配置：

{
  "queues": {
    "consumers": [{
      "queue": "tasks",
      "max_retries": 3,           // 最多重试 3 次
      "dead_letter_queue": "tasks-dlq"  // 重试耗尽后进入死信队列
    }]
  }
}

max_retries = 3 表示最多尝试 4 次（1 次初始 + 3 次重试）。超过后消息自动进入 dead_letter_queue。

5. 重试和幂等性

Queue 的 at-least-once 语义意味着消息可能被投递多次。如果消费者处理到一半崩溃，消息会被重新投递。重试时，消费者逻辑必须是幂等的。

幂等性：同一条消息处理两次，结果和处理一次一样。

// 非幂等的例子——会重复扣款
async function processPayment(msg: PaymentMessage, env: Env) {
  // 第一次：扣款成功
  await deductBalance(msg.userId, msg.amount)
  // 崩溃，消息重新投递
  // 第二次：又扣款一次——错误！
}
 
// 幂等的例子——用唯一 ID 防重
async function processPayment(msg: PaymentMessage, env: Env) {
  // 检查是否已处理
  const existing = await env.DB.prepare(
    'SELECT id FROM payments WHERE id = ?'
  ).bind(msg.paymentId).first()
 
  if (existing) return  // 已处理，跳过
 
  // 处理
  await deductBalance(msg.userId, msg.amount)
 
  // 记录已处理
  await env.DB.prepare(
    'INSERT INTO payments (id, user_id, amount, created_at) VALUES (?, ?, ?, ?)'
  ).bind(msg.paymentId, msg.userId, msg.amount, new Date().toISOString()).run()
}

用 KV 做去重

D1 查询慢，简单场景用 KV 去重更快：

async function processTask(msg: TaskMessage, env: Env): Promise<void> {
  const processedKey = `processed:${msg.taskId}`
 
  // 检查是否已处理
  const processed = await env.KV.get(processedKey)
  if (processed) return
 
  // 处理
  await doWork(msg, env)
 
  // 标记已处理（24 小时后过期）
  await env.KV.put(processedKey, '1', { expirationTtl: 86400 })
}

用数据库事务

复杂场景用事务保证原子性：

async function processTask(msg: TaskMessage, env: Env): Promise<void> {
  await env.DB.batch([
    // 检查状态
    env.DB.prepare('SELECT status FROM tasks WHERE id = ?').bind(msg.taskId),
  ]).then(async ([checkResult]) => {
    const task = checkResult.results[0]
 
    if (task?.status === 'completed') return  // 已处理
 
    // 处理 + 更新状态在同一个事务里
    await doWork(msg, env)
 
    await env.DB.prepare(
      "UPDATE tasks SET status = 'completed', completed_at = ? WHERE id = ? AND status != 'completed'"
    ).bind(new Date().toISOString(), msg.taskId).run()
  })
}

WHERE status != 'completed' 防止并发更新——如果另一个消费者已经标记为 completed，这个 UPDATE 不会生效。

6. 重试监控

重试次数过多说明系统有问题。监控重试率，及时报警。

// 查询重试率
async function getRetryRate(db: D1Database): Promise<{
  totalTasks: number
  retriedTasks: number
  retryRate: number
}> {
  const stats = await db.prepare(`
    SELECT
      COUNT(*) as total,
      SUM(CASE WHEN retry_count > 0 THEN 1 ELSE 0 END) as retried
    FROM tasks
    WHERE created_at > datetime('now', '-1 hour')
  `).first()
 
  const total = stats?.total ?? 0
  const retried = stats?.retried ?? 0
 
  return {
    totalTasks: total,
    retriedTasks: retried,
    retryRate: total > 0 ? retried / total : 0,
  }
}
 
// 查询平均重试次数
async function getAvgRetries(db: D1Database): Promise<number> {
  const result = await db.prepare(`
    SELECT AVG(retry_count) as avg_retries
    FROM tasks
    WHERE status = 'completed' AND created_at > datetime('now', '-1 hour')
  `).first()
 
  return result?.avg_retries ?? 0
}

重试率突然升高（比如从 5% 升到 30%）说明外部服务有问题。及时检查：

外部 API 是否限流？
网络是否不稳定？
服务是否过载？

7. 手动重试

自动重试失败的任务进入死信队列。但有些错误修复后可以手动重试。

// 手动重试失败的任务
app.post('/api/tasks/:taskId/retry', async (c) => {
  const taskId = c.req.param('taskId')
 
  const task = await c.env.DB.prepare(`
    SELECT id, status, payload FROM tasks WHERE id = ? AND user_id = ?
  `).bind(taskId, c.get('user').id).first()
 
  if (!task) return c.json({ error: 'Not found' }, 404)
  if (task.status !== 'failed') {
    return c.json({ error: 'Only failed tasks can be retried' }, 400)
  }
 
  // 重置状态
  await c.env.DB.prepare(`
    UPDATE tasks SET status = 'pending', retry_count = 0, error_message = NULL, updated_at = ?
    WHERE id = ?
  `).bind(new Date().toISOString(), taskId).run()
 
  // 重新入队
  const payload = JSON.parse(task.payload)
  await c.env.TASK_QUEUE.send({ taskId, ...payload })
 
  return c.json({ status: 'retrying' })
})

手动重试时重置 retry_count 和 error_message，让任务从头开始。

8. 重试策略配置

不同任务类型可以有不同的重试策略：

const RETRY_POLICIES: Record<string, { maxRetries: number; baseDelay: number }> = {
  'parse_pdf': { maxRetries: 3, baseDelay: 30 },
  'ocr_image': { maxRetries: 5, baseDelay: 10 },  // OCR 容易限流，多试几次
  'generate_embedding': { maxRetries: 3, baseDelay: 60 },  // embedding 慢，间隔长一些
  'send_notification': { maxRetries: 10, baseDelay: 5 },  // 通知很重要，多试
}
 
function getRetryPolicy(taskType: string) {
  return RETRY_POLICIES[taskType] ?? { maxRetries: 3, baseDelay: 30 }
}
 
// 在消费者中使用
const policy = getRetryPolicy(msg.body.type)
if (msg.attempts <= policy.maxRetries) {
  const delay = calculateDelay(msg.attempts - 1, policy.baseDelay)
  msg.retry({ delaySeconds: delay })
} else {
  // 进入死信队列
  msg.ack()
}

总结

回顾这一节的要点：

不是所有错误都值得重试——网络超时、限流、服务端错误值得重试；客户端错误不重试
重试间隔用指数退避——固定间隔会导致雪崩
加随机抖动——避免所有任务同时重试
最大重试次数设上限——3-5 次通常足够
Cloudflare Queue 内置重试——msg.retry({ delaySeconds })，max_retries 配置
重试必须配合幂等性——用去重标记、数据库事务、状态检查
监控重试率——突然升高说明外部服务有问题
手动重试——修复后可以重新处理失败任务
不同任务类型可以有不同的重试策略

下一篇讲死信队列——重试耗尽的任务去哪、怎么排查、怎么手动恢复。