单 Queue 保顺序 · 多 Queue 换性能 · 消费端兜状态 · Join 模式保严格顺序

导读：短信状态流转用单 Queue 保顺序，TPS 顶在单 Queue 瓶颈；改为 1024 Queue + 消费端幂等状态机 后吞吐上去，再用 uniqueId 分区 + Orderly 消费 降低锁竞争。若业务要求 A→B→C 严格串行 或 多步到齐再执行，才需要 任务组 Join + 到达日志。本文按四种顺序语义选型，代码以 RocketMQ 4.x 为例（5.x 见文末说明）。

---

阅读导航

章节	内容
背景与痛点	现象、根因：为什么单 Queue 会成为瓶颈
初步改造	1024 Queue + 并发消费 + 幂等状态机
进一步优化	分区键 + Orderly 顺序消费
严格顺序 Join	A→B→C 严格串行 / 多步到齐再执行
三种语义选型	乐观 / 终态正确 / 严格 FIFO / Join
结语与踩坑	全链路破坏顺序的因素

---

背景与痛点

现象

某 短信平台 向运营商发送短信后，需异步记录并更新发送状态。改造前出现：

现象	说明
消费堆积	高峰时段 Topic 消费 lag 持续升高，补跑需 数十分钟
TPS 上不去	仅 1 个 Queue，全集群同一时刻只有 一个消费线程 在处理该 Queue
扩容无效	Broker 从 2 扩到 6 台，该 Topic 吞吐几乎不变

业务场景

状态按固定链路流转，每条变更通过 RocketMQ 异步通知，下游消费端落库：

已提交运营商  →  已发送  →  用户已读

本文短信案例走的是「业务终态正确」路线（选型第二档）：依赖 至少一次投递 + 消费端幂等状态机，并非分布式理论里狭义的「副本最终一致」。未使用 Join 到达日志；Join 仅适用于后文「严格串行 / 多步到齐」类需求。

根因：常见顺序消息实现

讨论 MQ 顺序消息时，大家通常按这条链路理解：

1. 生产端：创建消息，发送到 MQ Server
2. Broker：消息写入 Topic 的 某一个 Queue
3. 消费端：从该 Queue 顺序消费（4.x 为 MessageListenerOrderly）

在「一个 Queue + 顺序消费」的前提下，顺序性最好保证，但 集群能力基本用不上。

现状与问题

项	说明
顺序保证方式	Topic 只开 1 个 Queue，消费端顺序消费
优点	状态流转不会乱序，逻辑简单
问题	TPS 卡在 单 Broker 单 Queue 上限，集群扩展无效
改造目标	提高业务 TPS，同时保证状态 最终正确

项	说明
架构	1 Topic × 1 Queue；Orderly 消费，集群内该 Queue 同时仅一线程处理
顺序	✓ MQ 层严格顺序
性能	✗ 单 Queue 瓶颈
集群扩展	✗ Broker 扩容无效

---

初步改造：多 Queue + 消费端状态控制

核心思路

把「顺序消息」改成「普通消息 + 消费端幂等状态机」，用 业务层纠偏 换 吞吐。

不再依赖 MQ 层面的全局顺序，而在消费端根据 当前库表状态 + 消息携带状态 决定是否处理、如何更新。RocketMQ 仍可能 至少一次 投递，靠状态机与幂等保证终态正确。

这是真实改造的第一刀：先证明「多 Queue + 状态机 + 并发消费」可行，生产者暂用默认轮询——分区键与顺序消费放在下一节。

改造步骤

1. 调大 Topic 读写 Queue 数（示例 1024，按 Broker 数与目标并行度配置）
   mqadmin updateTopic -n <nameserver> -t <topic> -r 1024 -w 1024
2. 生产者暂不做自定义路由，默认轮询投递到各 Queue
3. 消费者使用 MessageListenerConcurrently 多线程并发消费
4. 处理前对 uniqueId 加锁（或 DB 乐观锁），再按状态机更新或跳过

效果数据

压测条件：测试环境 3 Broker，消费者批量拉取，灌入 1w 条模拟状态消息（不含运营商回调间隔）。

指标	改造前（1 Queue + Orderly）	改造一（1024 Queue + 并发 + 状态机）	变化
消费 TPS	数百 msg/s 量级	数千 msg/s 量级	约一个数量级提升
单条消费 P99	明显偏高	降至可接受范围	消费链路纯 DB 更新
高峰消费 lag	持续堆积	分钟级可追平	生产高峰观测

消费端状态规则（MySQL 可重复读场景）

业务前提（上线前与产品确认）：

• 允许跨级写入：无记录时可直接插入「用户已读」等终态（仅展示最终结果，不要求库中经历每一中间态）
• 不允许回退：已处于更高状态时，忽略更低状态消息
• 表结构：unique_id 建 唯一索引；更新时用 UPDATE ... WHERE unique_id = ? AND status = ? 作乐观锁，避免并发写覆盖

假设表中已有状态，MQ 消息携带「目标状态」。原则：只允许沿状态 DAG 前进；无法处理的乱序一律跳过并打 metrics。

库表当前状态	消息状态	处理
无记录	已提交运营商 / 已发送 / 用户已读	插入为消息中的状态（允许跨级）
已提交运营商	已发送 / 用户已读	更新（前进，允许跨级到终态）
已发送	用户已读	更新为「用户已读」
已发送	已提交运营商	跳过（不允许回退）
用户已读	任意	跳过（终态，幂等成功）
已提交运营商	已提交运营商	跳过（重复消息）
任意更高态	任意更低态	跳过 + metric: status_rollback_blocked

并发双插：INSERT 命中唯一索引冲突时，改为 SELECT 后按上表走更新/跳过逻辑。

消费端伪代码（改造一：并发消费）

注册 并发 监听器（改造一专用，与改造二的 Orderly 不同）：

consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt ext : msgs) {
        ConsumeConcurrentlyStatus status = handleStatusMessage(parse(ext));
        if (status != ConsumeConcurrentlyStatus.CONSUME_SUCCESS) {
            return status;
        }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

核心处理逻辑（改造一、改造二共用）：

private ConsumeConcurrentlyStatus handleStatusMessage(SmsStatusMessage msg) {
    String lockKey = "sms:status:" + msg.getUniqueId();
    if (!tryLock(lockKey, 3, TimeUnit.SECONDS)) {
        // 拿不到锁：触发重投，禁止静默 ACK
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
    try {
        SmsRecord record = smsMapper.selectByUniqueId(msg.getUniqueId());
        if (record == null) {
            try {
                smsMapper.insert(buildFrom(msg));
            } catch (DuplicateKeyException ex) {
                record = smsMapper.selectByUniqueId(msg.getUniqueId());
                applyStateTransition(record, msg);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        applyStateTransition(record, msg);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    } catch (ConcurrentUpdateException ex) {
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    } finally {
        unlock(lockKey);
    }
}

private void applyStateTransition(SmsRecord record, SmsStatusMessage msg) {
    if (shouldSkip(record.getStatus(), msg.getStatus())) {
        metricCounter("status_skip", msg.getStatus());
        return;
    }
    if (shouldUpdate(record.getStatus(), msg.getStatus())) {
        int rows = smsMapper.updateStatusIfMatch(msg.getUniqueId(), record.getStatus(), msg.getStatus());
        if (rows == 0) {
            throw new ConcurrentUpdateException(); // 触发 RECONSUME_LATER
        }
    }
}

也可去掉分布式锁，仅用 唯一索引 + 乐观锁 UPDATE 实现幂等，锁竞争更低。

小结

维度	改造前	改造后
Queue 数	1	1024
消费方式	Orderly（单 Queue 单线程）	Concurrently 并发
顺序保证	MQ 层严格顺序	业务层幂等状态机
性能	单 Queue 瓶颈	集群并行，TPS 约一个数量级 提升

---

进一步优化：按消息 ID 分区

问题（改造一的真实痛点）

改造一里生产者 轮询 投递，同一 uniqueId 的多条状态消息可能进 不同 Queue，由多个消费线程并行处理，分布式锁 / 乐观锁竞争明显——这是上分区键并切换 Orderly 的直接动机。

做法（RocketMQ 4.x）

以 uniqueId 作为分区键，同一 ID 的消息始终进入 同一 Queue。开源 4.x 核心是 MessageQueueSelector；__SHARDINGKEY 为阿里云 SDK 可选属性，便于控制台识别，非 Apache 开源必配。

Message msg = new Message(topic, tags, uniqueId, body.getBytes(StandardCharsets.UTF_8));
// 可选：阿里云控制台识别分区
msg.putUserProperty("__SHARDINGKEY", uniqueId);

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        String key = (String) arg;
        int index = positiveHash(key) % mqs.size();
        return mqs.get(index);
    }
}, uniqueId);

// 避免 hashCode 为 Integer.MIN_VALUE 时 Math.abs 仍为负
private static int positiveHash(String key) {
    int hash = key.hashCode();
    return hash == Integer.MIN_VALUE ? 0 : Math.abs(hash);
}

注意：Topic 读写 Queue 数变更 或 Rebalance 后，同一 key 可能映射到新 Queue，短期内仍可能乱序——状态机继续兜底。

消费端同一 Queue 内改用 顺序消费，保证分区内 FIFO；业务仍保留上一节的状态机，作为 Rebalance / 重试窗口的兜底：

consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    for (MessageExt ext : msgs) {
        // 复用 handleStatusMessage；失败时挂起当前 Queue 稍后重试
        ConsumeConcurrentlyStatus status = handleStatusMessage(parse(ext));
        if (status != ConsumeConcurrentlyStatus.CONSUME_SUCCESS) {
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }
    return ConsumeOrderlyStatus.SUCCESS;
});
consumer.start();

RocketMQ 5.x：可将上述分区键映射为 MessageGroup（msg.setMessageGroup(uniqueId)），语义等价，API 不同。

效果

能力	说明
局部顺序	同一 uniqueId 进同一 Queue + MessageListenerOrderly，分区内按序
并行度	不同 ID 仍分散在不同 Queue，集群并行保留
与状态机配合	Rebalance、重试窗口内若仍乱序，消费端状态规则兜底

指标	改造一	改造二（+ 分区键 + Orderly）
同 key 锁竞争	较高	明显下降（同 key 不再跨 Queue 并发）
消费 TPS	数千 msg/s 量级	同量级或略升（瓶颈通常在 DB，而非锁）

短信场景里，状态回调本身有时间间隔，分区内顺序 + 状态机兜底 通常够用；若只是「同 key FIFO」，到这一档即可，不必上 Join。

消费端补充

改造二下，每个 Queue 由 Orderly 单线程顺序处理，一般不必再按 key 做线程分组。多 Consumer 实例时，各实例各持一部分 Queue，并行度来自 Queue 数量而非单 Queue 内多线程。

---

严格顺序：任务组 + 步骤到达日志（Join / Barrier 模式）

什么时候不够用？

以下两类需求，改造一/二 + 状态机 都不够：

需求类型	示例	为何状态机不够
严格 FIFO	必须先处理 A，再 B，再 C，不能跳过	乱序到达时「跳过」等于丢步骤
多步到齐（Join）	A、B、C 三路独立到达，都到了 才触发下游	不是「终态一致」，而是 Barrier 汇聚

术语说明
原笔记里曾用「母单 / 子单」称呼 Join 模式。这不是 RocketMQ 官方概念，也不是快递「子母运单」。
本文采用：任务组（Task Group）+ 步骤到达日志，等价于 Join / Barrier，与 Saga 编排、工作流汇聚节点同类。

模型说明

任务组：一次完整的多步骤任务（例如 A→B→C），对应一个 groupId
步骤消息：每一步对应一条 MQ 消息（step=A / B / C）
到达日志表：记录同一任务组下，哪些步骤已到达（Join 屏障）

到达日志表（DDL 示例）

CREATE TABLE mq_step_arrival (
    group_id    VARCHAR(64)  NOT NULL COMMENT '任务组 ID',
    step        VARCHAR(16)  NOT NULL COMMENT '步骤枚举 A/B/C',
    payload_ref VARCHAR(128)          COMMENT '可选：消息体摘要或外部 ID',
    arrived_at  DATETIME     NOT NULL COMMENT '到达时间',
    PRIMARY KEY (group_id, step),
    KEY idx_arrived_at (arrived_at)
) COMMENT 'MQ 步骤到达日志（Join 屏障）';

CREATE TABLE mq_task_group (
    group_id    VARCHAR(64)  NOT NULL COMMENT '任务组 ID',
    status      TINYINT      NOT NULL DEFAULT 0 COMMENT '0待齐套 1已齐套 2已处理',
    processed_at DATETIME             COMMENT '流水线执行完成时间',
    PRIMARY KEY (group_id)
) COMMENT '任务组执行状态（防重复触发）';

齐套判断伪代码（含 防重复执行）：

public void onStepMessage(String groupId, String step) {
    arrivalMapper.insertIgnore(groupId, step); // 幂等：PK (group_id, step)
    if (!arrivalMapper.isComplete(groupId, REQUIRED_STEPS)) {
        return;
    }
    // CAS：仅第一个线程将 0→1 成功者可继续
    if (!taskMapper.markReady(groupId)) {
        return;
    }
    if (!taskLock.tryLock("task:" + groupId)) {
        throw new RetryLaterException();
    }
    try {
        if (taskMapper.isProcessed(groupId)) {
            return;
        }
        runPipelineInOrder(groupId, List.of("A", "B", "C"));
        taskMapper.markProcessed(groupId); // 1→2，幂等
    } finally {
        taskLock.unlock("task:" + groupId);
    }
}

与分区顺序（MessageGroup / ShardingKey）的区别

手段	解决什么问题	典型 API
分区键 + 同 Queue 顺序消费	同一 key 的消息 按发送顺序 逐条消费（严格 FIFO）	4.x MessageQueueSelector + MessageListenerOrderly；5.x MessageGroup
任务组 + 到达日志（Join）	多路步骤 都到了 才触发；齐套后按 A→B→C 回放	业务表 + 兜底 Job，非 MQ 内置能力

若只是「同一条业务线按序处理」，分区顺序即可；若是「等多路步骤到齐再执行」，才需要 Join 日志表。两类需求可叠加：步骤消息仍用 groupId 作分区键，到达后再 Join。

实现要点

1. 创建任务组，生成 groupId，定义 step 枚举
2. 步骤消息到达时 幂等写入 到达日志表
3. Barrier：同一 groupId 下必需步骤是否全部到达
4. 齐套后 按 A→B→C 顺序 执行业务（单线程或组内锁）
5. 容错：超时未到齐 → 告警、重发缺失步骤、标记异常
6. 兜底 Job：扫描「已齐套未处理」「长时间未齐套」任务组

注意

• 权衡 性能、扩展性、严格顺序；Join 比状态机更重
• 并发、事务、失败重试需在设计阶段考虑
• 上线前压测 + 异常演练（缺 B、重复 C、超时、Rebalance 等）

---

四种顺序语义怎么选

把顺序消息需求归纳为 四档（第三档拆成 FIFO 与 Join，避免混为一谈）：

1. 尽可能保证顺序（乐观）

项	说明
适用	乱序概率低、间隔大、偶尔乱序可接受
做法	分区键 → 同 Queue；消费端可并发，不保证绝对顺序
特点	实现最轻，性能最好

2. 业务终态正确（本文短信案例）

项	说明
适用	要的是状态 最终正确，而非「严格先 A 后 B」
做法	多 Queue + Concurrently / Orderly + 消费端状态机 / 幂等 / 防回退
特点	虚假顺序消费——消息可乱序到达，业务层纠偏到正确终态
前提	状态机规则可枚举；允许跨级写入；接受至少一次投递

3. 严格 FIFO（同 key 逐条有序）

项	说明
适用	同一业务键下，必须先 A 后 B 后 C，不能跳过任一步
做法	分区键 + 同 Queue + MessageListenerOrderly（4.x）/ MessageGroup（5.x）
兜底	消费失败挂起当前 Queue，勿并发消费同一 Queue

4. 多步到齐再执行（Join / Barrier）

项	说明
适用	A、B、C 独立到达，全部到齐 才触发流水线
做法	任务组 + 到达日志表 + 齐套后顺序回放 + 兜底 Job
特点	实现最重，性能相对较低，语义最强

语义	典型手段	性能
尽可能顺序	分区键 + 同 Queue（可并发消费）	高
最终一致	多 Queue + 幂等状态机（短信案例）	高
严格 FIFO	分区键 + 顺序消费	中高
Join 到齐	到达日志表 + Barrier + 兜底 Job	相对较低

---

结语：哪些环节会破坏顺序？

全链路：生产端 → MQ Broker → 消费端，都可能出现「非顺序」。

环节	典型原因
生产端	多实例并发发送、重试导致后发的先到
Broker	多 Queue 轮询、Broker 切换、主从复制延迟、Queue 数变更导致 hash 漂移
消费端	误用并发消费同一 Queue、Rebalance、消费失败重试

不要默认「用了顺序消息 API 就万事大吉」，先问清楚业务要的是：

• 尽量有序？
• 终态正确即可？（→ 状态机）
• 同 key 严格 FIFO？（→ 分区 + 顺序消费）
• 多步到齐再执行？（→ Join 日志）

再选对应方案。短信场景我们停在 第二档 + 改造二分区/Orderly；若你遇到「步骤不能跳过」或「多路汇聚」，再考虑第三、四档。

RocketMQ 5.x 说明

4.x	5.x
MessageQueueSelector（分区路由核心）	MessageGroup
__SHARDINGKEY（阿里云可选）	控制台 / SDK 分组展示
MessageListenerOrderly	对应 FIFO 消费语义（以所用 SDK 文档为准）

参考资料

• RocketMQ 4.x - 顺序消息发送
• RocketMQ 官方文档 - 顺序消息（FIFO）
• 阿里云 - RocketMQ 4.x 顺序消息收发示例