消息队列与异步任务

异步化与消息队列手册：讲清何时该用 MQ、RabbitMQ/Kafka/Redis Stream 选型、生产与消费实现、延迟任务、幂等重试、可靠性与积压处理，以及从轮询到专业 MQ 的演进路径。

一、什么时候需要消息队列

1.1 三个典型场景

1.2 判断标准

该用 MQ 的信号：

• 请求链路里有明显慢的、非必须同步完成的操作（发短信、生成 PDF）
• 一个动作要触发多个下游处理，且下游可能失败或需要重试
• 存在明显的流量尖峰，下游处理能力有限
• 需要延迟执行（30 分钟未支付自动取消）

不该用 MQ 的信号：

• 只是「代码看起来更优雅」——多一个中间件就多一层运维负担
• 业务要求强一致且能同步完成（如扣库存必须立刻成功）
• 数据量极小、直接同步调用即可

一句话：当「同步做会拖慢用户」或「必须失败重试」时才引入 MQ。

二、选型对比

选型建议

现实建议：先用最简单的方案跑通，等真的成为瓶颈再升级。为了每天 100 条消息上 Kafka，是典型的过度设计。

三、核心概念

两种投递语义：

工程上默认按 At least once 设计——可靠性靠不丢，正确性靠消费端幂等。

四、Redis Stream 实战

4.1 生产者

const Redis = require('ioredis')
const redis = new Redis()

// 发送消息（* 表示由 Redis 生成 ID）
async function publishOrderPaid(orderId, userId) {
  const id = await redis.xadd(
    'stream:order:paid',
    '*',
    'orderId', orderId,
    'userId', userId,
    'ts', Date.now()
  )
  return id
}

4.2 消费组创建与消费

const STREAM = 'stream:order:paid'
const GROUP = 'worker-group'
const CONSUMER = `worker-${process.pid}`

// 启动时创建消费组（从最新消息开始；'0' 表示从头消费全部历史）
try {
  await redis.xgroup('CREATE', STREAM, GROUP, '$', 'MKSTREAM')
} catch (e) {
  if (!e.message.includes('BUSYGROUP')) throw e    // 已存在则忽略
}

// 消费循环
async function consume() {
  while (true) {
    // '>' 表示从未投递过的消息
    const res = await redis.xreadgroup(
      'GROUP', GROUP, CONSUMER,
      'COUNT', 10,
      'BLOCK', 5000,          // 最多阻塞 5 秒
      'STREAMS', STREAM, '>'
    )

    if (!res) continue

    for (const [, messages] of res) {
      for (const [id, fields] of messages) {
        try {
          await handleMessage(fields)
          await redis.xack(STREAM, GROUP, id)        // ACK
        } catch (err) {
          console.error('处理失败', id, err)
          // 不 ACK：消息会留在 PEL 中，需由补偿逻辑重试
        }
      }
    }
  }
}

4.3 处理挂起消息（消费者崩溃后的补偿）

// 查看长时间未 ACK 的消息（空闲超过 60 秒）
async function reclaimStale() {
  const pending = await redis.xpending(STREAM, GROUP, '-', '+', 100)
  for (const [id, consumer, idle] of pending) {
    if (idle > 60000) {
      // 认领并重试
      const msg = await redis.xclaim(STREAM, GROUP, CONSUMER, 60000, id)
      if (msg) {
        try {
          await handleMessage(msg[0][1])
          await redis.xack(STREAM, GROUP, id)
        } catch (e) {
          await moveToDLQ(id, msg)
        }
      }
    }
  }
}

4.4 修剪历史消息（防无限增长）

// 保留最近 10000 条（近似修剪，性能好）
await redis.xtrim(STREAM, 'MAXLEN', '~', 10000)

必须做修剪：Stream 不会自动删除已消费的消息，不修剪会持续占用内存直到 OOM。

五、RabbitMQ 要点

5.1 交换机类型

5.2 Node 示例

const amqp = require('amqplib')

// 生产者
async function publish(routingKey, payload) {
  const conn = await amqp.connect(process.env.AMQP_URL)
  const ch = await conn.createChannel()

  // 声明持久化交换机与队列
  await ch.assertExchange('app.events', 'topic', { durable: true })
  await ch.assertQueue('order.queue', { durable: true })
  await ch.bindQueue('order.queue', 'app.events', 'order.*')

  ch.publish('app.events', routingKey, Buffer.from(JSON.stringify(payload)), {
    persistent: true,               // 消息持久化
    messageId: payload.messageId,   // 便于去重
  })

  await ch.close()
  await conn.close()
}

// 消费者
async function consume() {
  const conn = await amqp.connect(process.env.AMQP_URL)
  const ch = await conn.createChannel()
  await ch.assertQueue('order.queue', { durable: true })
  await ch.prefetch(10)             // 每次最多取 10 条，防单实例吃光

  ch.consume('order.queue', async (msg) => {
    if (!msg) return
    try {
      await handle(JSON.parse(msg.content.toString()))
      ch.ack(msg)
    } catch (e) {
      // requeue=false → 进入死信队列，避免无限重试打爆队列
      ch.nack(msg, false, false)
    }
  })
}

5.3 死信队列配置

await ch.assertQueue('order.dlq', { durable: true })

await ch.assertQueue('order.queue', {
  durable: true,
  arguments: {
    'x-dead-letter-exchange': '',
    'x-dead-letter-routing-key': 'order.dlq',
    'x-message-ttl': 60000,
  },
})

死信来源：消费被 reject 且不重回队列、消息 TTL 过期、队列超长。

六、延迟任务

6.1 四种实现方式

6.2 Redis ZSet 实现（推荐）

// 投递延迟任务：30 分钟后执行
async function schedule(taskId, payload, delayMs) {
  const executeAt = Date.now() + delayMs
  await redis.zadd('delay:queue', executeAt, JSON.stringify({ taskId, payload }))
}

// 轮询取到期任务（每秒一次）
async function pollDue() {
  const now = Date.now()
  const due = await redis.zrangebyscore('delay:queue', 0, now, 'LIMIT', 0, 100)

  for (const item of due) {
    // 用 ZREM 原子移除，成功者才执行（防多实例重复）
    const removed = await redis.zrem('delay:queue', item)
    if (removed === 1) {
      const { taskId, payload } = JSON.parse(item)
      await handleTask(taskId, payload)
    }
  }
}

关键点：用 ZREM 返回 1/0 来保证多实例下的原子抢占，只有一个实例会执行。

6.3 常见延迟任务

七、幂等与重试

7.1 为什么必须幂等

At least once 语义 + 网络重试 + 消费者重启，都会导致同一消息被处理多次。

7.2 幂等的三种实现

方式一：业务唯一键（最可靠）

-- 消息带唯一 ID，落库时靠唯一索引去重
INSERT INTO points_log (message_id, user_id, amount) VALUES (?, ?, ?);
-- 唯一索引冲突 → 说明已处理，直接 ACK

方式二：Redis 去重标记

const dedupKey = `processed:${messageId}`

// SET NX 原子占位，只有第一次能成功
const first = await redis.set(dedupKey, '1', 'EX', 86400, 'NX')
if (!first) {
  logger.info({ event: 'msg.duplicate', messageId })
  return   // 已处理过，直接返回（随后 ACK）
}

try {
  await handle(msg)
} catch (e) {
  await redis.del(dedupKey)   // 失败要释放，允许重试
  throw e
}

方式三：状态机 + 条件更新

UPDATE orders SET status = 'paid'
WHERE id = ? AND status = 'pending';
-- 影响 0 行 → 已被处理，视为成功

7.3 重试策略

// 指数退避：1s → 2s → 4s → 8s → 16s，最多 5 次
function backoff(attempt) {
  return Math.min(1000 * Math.pow(2, attempt), 60000)
}

超过最大重试次数 → 进入死信队列 + 告警，不要静默丢弃。

八、消息可靠性

8.1 三个环节都不能丢

8.2 最常见的错误

// ❌ 先 ACK 后处理：处理失败消息已丢
ch.consume(queue, async (msg) => {
  ch.ack(msg)
  await handle(msg)      // 这里抛异常，消息已经没了
})

// ✅ 先处理成功再 ACK
ch.consume(queue, async (msg) => {
  try {
    await handle(msg)
    ch.ack(msg)
  } catch (e) {
    ch.nack(msg, false, true)   // 重回队列（注意配合重试上限）
  }
})

8.3 本地消息表（跨库强一致场景）

跨服务操作时，用「本地消息表」保证「业务操作」与「消息发送」的一致性：

1. 同一数据库事务内：写业务表 + 写消息表（状态=待发送）
2. 事务提交后，异步扫描消息表发送
3. 发送成功 → 标记已发送
4. 定时任务补偿未发送的记录

九、积压处理

9.1 积压的成因

9.2 应对

// 监控积压量
const len = await redis.xlen('stream:order:paid')
if (len > 10000) {
  logger.warn({ event: 'mq.backlog', length: len })
}

9.3 批量消费提升吞吐

// ❌ 逐条处理：每条一次数据库往返
for (const msg of batch) await db.insert(msg)

// ✅ 批量处理：一次往返
await db.insertMany(batch.map(parse))

十、常见问题

十一、检查清单

选型与设计

• [ ] 确认真的需要 MQ（不是可以同步完成的事）
• [ ] 选型匹配规模（小量用 Redis Stream 或数据库轮询即可）
• [ ] 消息体精简（只传 ID + 必要字段，不放整个对象）
• [ ] 消息带 messageId 与业务唯一键

可靠性

• [ ] 生产者有发送确认与失败重发
• [ ] 队列与消息已持久化
• [ ] 消费者手动 ACK，处理成功后才确认
• [ ] 配置最大重试次数 + 死信队列
• [ ] 有挂起消息（PEL）的补偿机制

幂等

• [ ] 消费端按 At least once 语义设计
• [ ] 有幂等实现（唯一索引 / Redis 去重 / 状态机）
• [ ] 幂等标记的 TTL 与业务窗口匹配

延迟任务

• [ ] 多实例下用原子操作抢占（ZREM 返回值）
• [ ] 有到期任务遗漏的补偿扫描

运维

• [ ] 队列深度有监控与告警
• [ ] Stream 已配置 MAXLEN 修剪
• [ ] 死信队列有监控与处理流程
• [ ] 消费失败日志含 messageId 便于排查

十二、总结

异步与消息队列的落地，可以按这个顺序推进：

1. 先判断是否真的需要——同步能完成、量又小，就别引入中间件。
2. 从最轻的方案起步——Redis Stream 或「数据库表 + 定时轮询」足以支撑中小项目；真成瓶颈再换 RabbitMQ/Kafka。
3. 可靠性靠三件事：生产者确认、消息持久化、消费者处理成功后再 ACK。
4. 正确性靠幂等——按 At least once 设计，消费端用唯一键或 Redis 去重兜底。
5. 别忘修剪与监控——Stream 不修剪会 OOM，队列积压不监控会静默失效。

最容易踩的坑，往往不是选错 MQ，而是：先 ACK 后处理导致丢消息、没有幂等导致重复扣款、死循环重投把下游打爆。这三点都有成熟的解法，落到代码里的成本也不高——难的是记得去做。

完