消息队列在 IM 中的作用：削峰、异步落库与扇出
消息队列（message queue，MQ）在 IM 里不是"为了解耦而解耦"的装饰，它解决三个具体问题：削峰（上行峰值 5 万/s，落库与扇出可能只处理得动 1 万/s）、异步落库（除正文外的次要写：索引、计数、搜索）、扇出（一条群消息要给 N 个成员写收件箱、给 M 台设备发推送）。代价是引入"至少一次"投递、重复消费、顺序破坏与积压，本章讲清怎么付这个代价。
先落库还是先投递
这是最容易搞反的顺序，三种方案摆在一起看：
| 方案 | 流程 | 优点 | 致命问题 |
|---|---|---|---|
| A. 先投递后落库 | 入队 → 回 ACK → 消费端落库 | ACK 快、延迟低 | 消费端崩溃或队列丢消息时"已 ACK 但查不到"，接收方拉历史为空；发送方重发又被去重挡住，形成永久空洞 |
| B. 先落库后投递 | 落库成功 → 回 ACK → 投递 MQ 驱动扇出与推送 | ACK 语义准确；队列全丢只是变慢，接收方上线仍能拉到 | 落库进入同步路径，写入延迟直接计入端到端延迟 |
| C. 本地消息表（Outbox） | 同一事务写业务表 + outbox → relay 读 outbox 投递 → 消费成功标记 | 落库与投递最终一致，投递可重试，不依赖 MQ 事务消息 | 多一张表和一个 relay 进程，需要清理 |
推荐 B，规模上来或要求"投递绝不丢"时升级为 C。 判断依据只有一条：ACK 之后的消息必须能查到、能补拉，凡是不满足这条的方案都是错的。
Outbox 表与 relay
CREATE TABLE mq_outbox (
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
topic VARCHAR(64) NOT NULL COMMENT '如 im.fanout.msg',
shard_key BIGINT UNSIGNED NOT NULL COMMENT 'conv_id 或 uid，用于顺序分区',
payload VARBINARY(8192) NOT NULL COMMENT '事件体：msg_id/conv_id/conv_seq',
status TINYINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '0=待投递 1=已投递 2=已确认',
retry INT UNSIGNED NOT NULL DEFAULT 0,
created_at DATETIME(3) NOT NULL,
PRIMARY KEY (id),
KEY idx_status_id (status, id) -- relay 按此索引顺序扫描，天然保序
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
relay 两个要点：按 (status, id) 顺序批量扫描，保证同一 shard_key 的相对顺序；投递成功后先标 status=1，等消费端回报再标 2，避免"投出去了但消费端没收到"时无从判断。
幂等消费
MQ 只能承诺至少一次（at-least-once），幂等（idempotent）必须由消费端保证：
| 做法 | 实现 | 适用 |
|---|---|---|
| 唯一键去重表 | 消费时插入去重表，冲突即跳过 | 单分片、要求强一致 |
| Redis SETNX | SET im:mq:dedup:{msgID}:{uid} 1 NX EX 86400 | 高吞吐、可接受极小概率误判 |
| 状态机幂等 | 动作本身可重复（GREATEST 更新、ZSet 覆盖写） | 无副作用的更新操作 |
CREATE TABLE mq_dedup (
msg_id BIGINT UNSIGNED NOT NULL,
uid BIGINT UNSIGNED NOT NULL COMMENT '收件人；群聊里同一 msg_id 会有多行',
created_at DATETIME(3) NOT NULL,
PRIMARY KEY (msg_id, uid)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='消费去重表，按 created_at 定期清理';
关键提醒：去重键必须带接收方维度。群聊一条消息要扇出给 500 个成员，消费端可能拆成 500 条处理，只用 msg_id 去重会导致只处理第一个成员、其余全被跳过。
顺序消费的取舍
"全局有序"在 IM 里既不必要也不现实，真正要保的是会话内有序：
| 粒度 | 实现 | 代价 | 是否推荐 |
|---|---|---|---|
| 全局单分区 | 单 topic 单分区单消费者 | 吞吐上限就是单线程 | 否 |
| 按 conv_id 分区 | hash(conv_id) % N 决定分区 | 热点大群固定在单分区 | 推荐（默认） |
| 按 uid 分区 | 每个用户的收件箱写入各自有序 | 会话内可能乱序 | 仅当只保用户维度时 |
| 不保序 + 客户端排序 | 消费端并发，客户端按 conv_seq 排 | 客户端要处理乱序与断洞 | 所有方案都建议保留此兜底 |
工程上的组合拳是**"按 conv_id 分区保绝大部分顺序 + 客户端按 conv_seq 兜底"**：前者让 99% 的消息按序到达，后者让热点会话拆分区或多个消费者时依然正确——客户端本来就按 seq 排序渲染，发现 10 而本地最大是 7 就触发一次会话级补拉。
队列选型
| 选型 | 定位 | 顺序 | 注意 |
|---|---|---|---|
| Kafka | 高吞吐事件流、扇出、索引与审计 | 分区内有序 | 运维成本高，不适合海量小 topic |
| RocketMQ | 业务消息，事务消息与延迟消息成熟 | 队列内有序 | 支持事务消息，适合方案 C |
| RabbitMQ | 中小规模、路由灵活（direct/topic/fanout） | 队列内有序 | 大积压时内存压力大，需要惰性队列 |
| Redis Stream | 进程内小规模可靠投递、单机 fan-out | 全局有序 | 数据在内存，需要 MAXLEN 裁剪 |
常见组合：上行与扇出走 Kafka/RocketMQ，接入机之间的下行唤醒走 Pub/Sub（14 章），进程内小批量任务走 Redis Stream。不要指望一个中间件同时满足"海量堆积 + 低延迟 + 强顺序 + 事务"。
削峰与积压治理
| 场景 | 现象 | 处置 |
|---|---|---|
| 上行洪峰（活动、节日） | 队列 lag 上涨 | 消费者水平扩容；分区数必须提前开够（分区数 ≥ 消费者数） |
| 扇出消费慢 | lag 持续增长 | 分层：先写收件箱索引，推送与搜索索引降级到二级队列 |
| 消费者异常重启 | 重复消费一批 | 幂等去重 + 从已提交位点续读，不要从头读 |
| 死信堆积 | 个别消息反复失败 | 重试上限（建议 5 次）后转死信队列，人工或定时任务处理，绝不无限重试阻塞主队列 |
| 队列整体不可用 | 上行无法异步化 | 降级：落库后只做在线下行，扇出与推送延后由对账任务补齐 |
Lag 监控要落到"最老未处理消息的年龄"，而不是只看队列总条数：总条数 100 万但都是 1 秒前的，与总条数 1 万但最老是 10 分钟前的，严重程度完全不同。
对账与补偿
队列与存储之间的不一致必须有人兜：
- 正向对账：扫描
msg_content中"已落库但收件箱缺失"的消息（按 created_at 区间比对 msg_id），补写收件箱； - 反向对账：扫描收件箱中"有索引但正文缺失"的行，补拉正文；
- 推送对账：离线超过 N 分钟仍未被任何设备拉取的消息，转交厂商推送（21 章）。
对账任务必须能安全重跑：全部基于 msg_id 幂等写入，跑第二遍的代价是几次写冲突，而不是脏数据。
工程注意点
- ACK 只承诺落库：不要因为"已投进 MQ"就提前回 ACK，MQ 不是持久化承诺；
- 消息体只放定位信息：
{msg_id, conv_id, conv_seq, sender_uid}，正文让消费端回源，别把队列当存储； - 消费端要有超时与并发上限：每分区一个消费者协程、单条处理超时（如 3s）、单批条数上限，防止一次拉太多把内存打满；
- 顺序与吞吐必须显式取舍：在架构文档里写清"哪些队列保序、按什么键分区"，否则后续有人加消费者就把顺序搞坏；
- 压测重点是 lag 恢复能力：峰值只是一瞬间，积压恢复慢会持续影响所有用户。
小结：MQ 在 IM 里解决削峰、异步落库与扇出三件事，前提是顺序不能错——必须先落库再投递（或走本地消息表 Outbox 保证最终一致），ACK 只承诺落库；消费端用 (msg_id, uid) 维度做幂等去重；顺序按 conv_id 分区保会话内有序，并保留客户端按 conv_seq 排序作为兜底；积压治理看"最老未处理消息年龄"，重试有上限并转死信，最后用正反向对账把队列与存储之间的不一致补齐。