消息队列与异步任务

注册后发欢迎邮件、下单后通知仓库、每晚生成报表……这些工作如果都放在请求里同步完成，用户就得等它们全部做完；其中任何一个慢了或失败了，整个请求都会受影响。异步任务就是把它们移出请求，交给后台处理。

但“移出去”之后，问题就变成了：怎么保证它一定会完成，而且只产生一次效果。真实项目里常见的事故是：

• 消费者处理订单消息时，报“订单不存在”，可数据库里明明有；
• 用户只下了一单，却被扣了两次钱；
• 一条格式错误的消息反复失败，后面的消息全部堵住；
• 队列积压了几十万条，没人发现。

读完这篇你会

• 判断哪些工作应该异步化
• 理解“不丢、重试、幂等”三个可靠性保证
• 修复事务时序、重复执行、毒消息、积压和乱序等问题

1. 原理：餐厅的出单夹

餐厅前台接到订单后，把单子夹到厨房的出单夹上，就转身去接待下一位客人了，不需要站在厨房门口等菜做好。厨师按顺序取单、做菜，做完把单子划掉。

生产者（前台）：在请求里创建任务，放进队列，立即返回
      ↓
队列（出单夹）：暂存待处理的任务
      ↓
消费者（厨师）：后台进程取出任务、执行、完成后确认（划单）

这套流程要可靠，有三个要求：

1. 单子不能丢：前台夹上去的单子，厨房一定能看到；厨师做到一半去了趟洗手间（进程重启），单子也还在；
2. 做砸了可以重做：一道菜失败了，可以重新做，但不能无限重做；
3. 同一张单不能做两份菜：如果厨师做完了却忘了划单，另一个厨师可能又做一遍。

第三点特别关键：大多数消息队列提供的是至少一次（at-least-once）投递。为了不丢消息，它宁可多投递一次：消费者处理完、还没来得及确认就崩溃了，消息会被重新投递。所以重复执行不是意外，而是一定会发生的正常情况，消费者必须能处理它。

2. 什么应该异步化

判断标准：用户是否需要立刻知道这个操作的结果？

同步完成	异步处理
创建订单、扣减库存	发送确认邮件、短信、推送
权限校验、支付下单	生成报表、导出文件、处理图片
用户必须马上看到的数据	同步数据到其他系统、统计分析

用户正在等待结果的操作（比如支付是否成功），强行异步反而让体验变差：你得让前端轮询或推送结果，复杂度更高。

// 请求里：核心写入同步完成，通知异步处理
async function register(req, res) {
  const user = await userService.create(req.body)
  await emailQueue.add('welcome', { userId: user.id }) // 放进队列，不等待发送
  res.status(201).json({ id: user.id })                // 用户立即得到响应
}

// 后台消费者
emailQueue.process('welcome', async (job) => {
  await sendWelcomeEmail(job.data.userId)
})

3. 坑一：消费者说“订单不存在”，可数据库里明明有

现象：消费者处理“订单已创建”消息时，偶尔报错“找不到订单”。可人工去数据库里查，订单确实存在。

await db.transaction(async (tx) => {
  const order = await orderRepo.create(tx, input)
  await queue.add('order.created', { orderId: order.id }) // ❌ 事务还没提交，消息已经发出
  await couponRepo.use(tx, input.couponId)
})

原因：消息在事务提交之前就已经发出了。消费者处理得很快，去数据库查订单时，事务还没提交，自然查不到。更糟的情况是，事务最后回滚了，消息却已经发出，消费者处理了一个根本不存在的订单。

修复：

• 简单场景：事务提交之后再发消息；
• 需要“数据提交和消息发送同时成功”时，使用发件箱模式：在同一个事务里把消息写进发件箱表，提交后由后台进程把它投递到队列，详见 服务边界与业务编排。

4. 坑二：重试导致重复扣款

现象：用户只下了一单，银行卡被扣了两次。日志显示扣款任务执行了两次，两次都“成功”了。

原因：第一次扣款成功后，消费者在确认消息之前崩溃了。队列认为这条消息没有被处理，重新投递，扣款又执行了一次。这正是第 1 节说的“至少一次投递”。

修复：让执行逻辑幂等：同一个任务执行多少次，效果都和执行一次一样。

paymentQueue.process('charge', async (job) => {
  const { orderNo, amount } = job.data

  // 1. 用唯一约束登记这笔支付：同一个订单号只会有一条记录（不要“先查再插”，并发时会同时通过检查）
  await db.query(
    'INSERT IGNORE INTO payment_attempts (order_no, status) VALUES (?, ?)',
    [orderNo, 'processing']
  )
  const [attempt] = await db.query('SELECT status FROM payment_attempts WHERE order_no = ?', [orderNo])
  if (attempt.status === 'done') return // 已经成功处理过，直接跳过

  // 2. 调用支付渠道时传入商户订单号：渠道对同一个订单号只会扣款一次
  await paymentGateway.charge({ outTradeNo: orderNo, amount })

  await db.query('UPDATE payment_attempts SET status = ? WHERE order_no = ?', ['done', orderNo])
})

这里有两层保护：

• 第一层，已经完成的任务直接跳过，避免重复劳动；
• 第二层才是真正防止重复扣款的关键：支付渠道按商户订单号去重。即使两个消费者同时处理同一条消息，或者上次在扣款成功后、更新状态前崩溃了，重试时再次调用也不会多扣一次钱。

如果第三方接口不支持幂等参数，就要在调用前先向它查询这笔业务的状态（查单），确认没有处理过再调用。

怎么避免再犯：写每一个消费者时都假设“这条消息会被处理两次”，然后问自己：两次之后结果对不对？发邮件可以用“这封邮件发过没有”的记录去重；创建资源用业务唯一键去重；调用第三方时尽量传入它支持的幂等参数。

5. 坑三：一条坏消息堵住了整个队列

现象：某条消息的数据格式有问题，消费者每次处理都报错。它被不断重试，后面成千上万条正常消息全都排在它后面无法处理。

原因：没有重试上限，失败的消息被无限次重新放回队列头部。这种无论重试多少次都不会成功的消息叫毒消息。

修复：重试要有上限和退避（每次重试间隔逐渐变长），超过上限的消息移入死信队列，等待人工排查，不影响其他消息：

处理失败 → 1 秒后重试 → 4 秒后重试 → 16 秒后重试 → 仍失败 → 进入死信队列 + 告警

同时区分错误类型：网络超时这类暂时性错误值得重试；数据格式错误、业务规则不满足这类永久性错误，重试一万次也没用，应该直接进入死信队列。

6. 坑四：队列积压越来越多，没人发现

现象：用户反馈“注册邮件一直没收到”，查看才发现邮件队列里积压了几十万条，已经持续了好几天。

修复：异步任务看不见，所以必须有监控：

指标	说明什么
队列深度（待处理数量）	持续增长说明消费速度跟不上
处理延迟（入队到完成的时间）	突然变长说明依赖变慢或消费者有问题
失败率和重试次数	持续失败说明代码或环境有问题
死信队列数量	有新增就需要有人处理
长时间“处理中”的任务	消费者可能卡死了

发现积压时，不要第一反应就加消费者：先弄清楚原因。如果是某个任务卡住、或者下游服务变慢，加消费者只会让下游压力更大。

7. 坑五：消息乱序，旧状态覆盖了新状态

现象：订单先“已支付”再“已发货”，但下游系统里最终显示“已支付”。

原因：两条消息被不同的消费者并行处理，“已发货”先处理完，“已支付”后处理完，覆盖了新状态。

修复：

• 需要按顺序处理的消息，按业务键（如订单号）分配到同一个分区或同一个消费者；
• 或者让消息带上版本号或时间戳，消费者只接受比当前更新的状态：

UPDATE orders SET status = ?, status_version = ?
WHERE id = ? AND status_version < ?  -- 旧消息不会覆盖新状态

8. 并发和优先级

• 并发数不是越大越好：下游服务（数据库、第三方接口）有承受上限。第三方邮件服务限制 10 并发，消费者就设 8 左右，留出余量；
• 区分优先级：密码重置邮件比营销邮件紧急，可以放进不同的队列，分别配置消费能力。

排查思路

1. 消息处理失败：先看死信队列里的消息内容和错误信息，判断是数据问题、代码问题还是依赖问题。
2. “数据不存在”类报错：检查消息是在事务提交前还是提交后发出的。
3. 重复执行：在消费者里打印消息 ID 和业务键，看同一业务键是否被处理了多次，确认幂等措施是否生效。
4. 积压：对比生产速度和消费速度；查看单个任务的处理耗时是否变长；检查是否有任务长时间卡在处理中。

检查清单

• [ ] 只有用户不需要立即看到结果的工作才异步化
• [ ] 队列是持久化的，消费者处理完成后才确认
• [ ] 消息在事务提交后发出，或者使用发件箱模式
• [ ] 每个消费者都是幂等的，使用唯一约束或第三方幂等键去重
• [ ] 重试有上限和退避，失败的消息进入死信队列并告警
• [ ] 监控队列深度、处理延迟、失败率和死信数量
• [ ] 有顺序要求的消息按业务键分区或使用版本号

延伸阅读

• 服务边界与业务编排：发件箱模式
• 错误处理与请求追踪：重试的退避与抖动
• 事务、幂等与补偿：业务流程中的幂等与补偿