消息队列与异步任务
异步化与消息队列手册：讲清何时该用 MQ、RabbitMQ/Kafka/Redis Stream 选型、生产与消费实现、延迟任务、幂等重试、可靠性与积压处理，以及从轮询到专业 MQ 的演进路径。
一、什么时候需要消息队列
1.1 三个典型场景
| 场景 | 说明 | 例子 |
|---|---|---|
| 解耦 | 主流程不关心后续动作，只管发出事件 | 下单成功后要发短信、加积分、写报表——都通过事件通知 |
| 异步 | 耗时操作移出请求链路 | 导出报表、视频转码、图片压缩 |
| 削峰 | 瞬时高并发排队处理 | 秒杀、批量导入 |
1.2 判断标准
该用 MQ 的信号：
- 请求链路里有明显慢的、非必须同步完成的操作（发短信、生成 PDF）
- 一个动作要触发多个下游处理，且下游可能失败或需要重试
- 存在明显的流量尖峰，下游处理能力有限
- 需要延迟执行（30 分钟未支付自动取消）
不该用 MQ 的信号：
- 只是「代码看起来更优雅」——多一个中间件就多一层运维负担
- 业务要求强一致且能同步完成（如扣库存必须立刻成功）
- 数据量极小、直接同步调用即可
一句话：当「同步做会拖慢用户」或「必须失败重试」时才引入 MQ。
二、选型对比
| 维度 | Redis Stream | RabbitMQ | Kafka |
|---|---|---|---|
| 部署复杂度 | 最低（已有 Redis 即可） | 中 | 高 |
| 吞吐量 | 中（万级/秒） | 中高（万级/秒） | 极高（十万级+/秒） |
| 消息可靠 | 中（可持久化，但不如专业 MQ） | 高 | 高 |
| 延迟队列 | ❌ 需自行实现 | ✅ 插件支持 | ❌ 需自行实现 |
| 消息回溯 | ✅（保留期内可重读） | ❌（消费即删） | ✅（按 offset 重放） |
| 消费组 | ✅ | ✅ | ✅ |
| 运维成本 | 极低 | 中 | 高 |
| 适用规模 | 中小项目 | 通用业务 | 日志/大数据/高吞吐 |
选型建议
| 你的情况 | 建议 |
|---|---|
| 已有 Redis，需求简单（异步任务、轻量队列） | Redis Stream |
| 需要延迟队列、复杂路由、可靠投递 | RabbitMQ |
| 日志采集、事件流、超高吞吐、需要重放 | Kafka |
| 数据量很小（每天几百条） | 数据库表 + 定时轮询（别上 MQ） |
现实建议：先用最简单的方案跑通，等真的成为瓶颈再升级。为了每天 100 条消息上 Kafka，是典型的过度设计。
三、核心概念
| 概念 | 含义 |
|---|---|
| 生产者（Producer） | 发送消息的一方 |
| 消费者（Consumer） | 处理消息的一方 |
| 队列 / Topic | 消息的存放通道 |
| 消费组（Consumer Group） | 多个消费者协同消费，同组内一条消息只被一个实例处理 |
| ACK | 消费成功后确认，未 ACK 的消息可重投 |
| 死信队列（DLQ） | 重试仍失败的消息转入，供人工处理 |
| 可见性超时 | 消息被取出后，超时未 ACK 则重新可见（防消费者崩溃丢消息） |
两种投递语义：
| 语义 | 含义 | 代价 |
|---|---|---|
| At most once（至多一次） | 可能丢，不重复 | 快，适合日志 |
| At least once（至少一次） | 不丢，可能重复 | 需要消费者幂等 |
| Exactly once（恰好一次） | 不丢不重 | 实现复杂，通常只在特定 MQ 内成立 |
工程上默认按 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 交换机类型
| 类型 | 路由规则 | 用途 |
|---|---|---|
direct |
按 routing key 精确匹配 | 点对点任务 |
topic |
按通配符匹配（order.*） |
多类事件分发 |
fanout |
广播到所有队列 | 通知多个系统 |
headers |
按消息头匹配 | 少用 |
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 四种实现方式
| 方式 | 原理 | 精度 | 适用 |
|---|---|---|---|
| 定时扫描数据库 | 每分钟查「到期待处理」记录 | 分钟级 | 最简单，量小够用 |
| Redis ZSet | score 存执行时间戳，轮询取到期成员 | 秒级 | 推荐，轻量 |
| RabbitMQ 延迟插件 | 消息设 TTL + 死信转发 | 秒级 | 已有 RabbitMQ 时 |
| 时间轮 | 内存分级时间轮 | 毫秒级 | 高精度、大数量 |
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 常见延迟任务
| 任务 | 延迟 |
|---|---|
| 订单未支付自动取消 | 15–30 分钟 |
| 支付后未到账补偿查询 | 5 分钟 |
| 优惠券到期提醒 | 到期前 1 天 |
| 注册未完善资料提醒 | 24 小时 |
七、幂等与重试
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)}| 错误类型 | 是否重试 |
|---|---|
| 网络超时、连接失败 | ✅ 重试 |
| 下游 5xx | ✅ 重试 |
| 参数错误（400） | ❌ 不重试，直接进死信 |
| 业务规则拒绝（库存不足） | ❌ 不重试 |
| 数据库死锁 | ✅ 重试（通常能成功） |
超过最大重试次数 → 进入死信队列 + 告警，不要静默丢弃。
八、消息可靠性
8.1 三个环节都不能丢
| 环节 | 风险 | 保障措施 |
|---|---|---|
| 生产者 → MQ | 发送失败 | 发送确认（确认回调/publisher confirm）、失败重发 |
| MQ 存储 | 宕机丢消息 | 消息持久化 + 队列持久化 + MQ 高可用 |
| MQ → 消费者 | 处理中崩溃 | 手动 ACK，处理成功才确认 |
8.2 最常见的错误
// ❌ 先 ACK 后处理：处理失败消息已丢ch.consume(queue, async (msg) => { ch.ack(msg) await handle(msg) // 这里抛异常，消息已经没了})// ✅ 先处理成功再 ACKch.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 应对
| 情况 | 处理 |
|---|---|
| 短期尖峰 | 临时扩容消费者实例（消费组内加机器） |
| 长期积压 | 优化消费逻辑（批量处理、去掉同步远程调用） |
| 消费逻辑有 bug | 先修复再恢复消费，不要盲目扩容（会把错误数据放大） |
| 死循环 | 检查是否有「处理失败 → nack 重投 → 又失败」的循环，加最大重试次数 |
// 监控积压量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))十、常见问题
| 现象 | 原因 | 处理 |
|---|---|---|
| 消息重复处理 | At least once + 无幂等 | 消费端加幂等（唯一键/Redis 去重） |
| 消息丢失 | 先 ACK 后处理 / 未持久化 | 处理成功后再 ACK；开持久化 |
| Redis Stream 内存暴涨 | 未修剪 | XTRIM MAXLEN ~ N |
| 消费越来越慢 | 单实例消费、串行处理 | 消费组扩容 + 批量处理 |
| 延迟任务不准时 | 轮询间隔粗 | 缩短轮询间隔或用 ZSet |
| 多实例重复执行延迟任务 | 未做原子抢占 | ZREM 返回值判断 |
| 死信堆积无人处理 | 未配置监控 | 死信队列长度告警 + 定期人工处理 |
| 重启后消费卡住 | PEL 中的消息未被认领 | 用 XCLAIM 补偿挂起消息 |
十一、检查清单
选型与设计
- 确认真的需要 MQ（不是可以同步完成的事）
- 选型匹配规模（小量用 Redis Stream 或数据库轮询即可）
- 消息体精简（只传 ID + 必要字段，不放整个对象）
- 消息带
messageId与业务唯一键
可靠性
- 生产者有发送确认与失败重发
- 队列与消息已持久化
- 消费者手动 ACK，处理成功后才确认
- 配置最大重试次数 + 死信队列
- 有挂起消息（PEL）的补偿机制
幂等
- 消费端按 At least once 语义设计
- 有幂等实现（唯一索引 / Redis 去重 / 状态机）
- 幂等标记的 TTL 与业务窗口匹配
延迟任务
- 多实例下用原子操作抢占（ZREM 返回值）
- 有到期任务遗漏的补偿扫描
运维
- 队列深度有监控与告警
- Stream 已配置 MAXLEN 修剪
- 死信队列有监控与处理流程
- 消费失败日志含 messageId 便于排查
十二、总结
异步与消息队列的落地，可以按这个顺序推进：
- 先判断是否真的需要——同步能完成、量又小，就别引入中间件。
- 从最轻的方案起步——Redis Stream 或「数据库表 + 定时轮询」足以支撑中小项目；真成瓶颈再换 RabbitMQ/Kafka。
- 可靠性靠三件事：生产者确认、消息持久化、消费者处理成功后再 ACK。
- 正确性靠幂等——按 At least once 设计，消费端用唯一键或 Redis 去重兜底。
- 别忘修剪与监控——Stream 不修剪会 OOM，队列积压不监控会静默失效。
最容易踩的坑，往往不是选错 MQ，而是：先 ACK 后处理导致丢消息、没有幂等导致重复扣款、死循环重投把下游打爆。这三点都有成熟的解法，落到代码里的成本也不高——难的是记得去做。
完