技术小册

主题

RocketMQ 消息类型与使用场景 ​

一句话：RocketMQ 用五种发送语义覆盖「普通异步、分区有序、定时、事务最终一致、批量吞吐」，每种语义都对应一条明确的存储与重试路径，选错类型的代价通常出现在故障时。

适用版本：4.9 / 5.3 / 5.5。延迟消息在 4.x 与 5.x 是两套实现，必须分开看。

为什么需要 ​

业务消息不是只有「发出去、收得到」。订单状态要按订单维度有序；支付结果要等本地事务提交后才可见；超时未支付要 30 分钟后取消；一批埋点要一次发完减少 RPC。把这些都塞进普通消息，调用方就得自己实现顺序、定时和一致性，成本更高且容易出错。

RocketMQ 把这些模式做成 Broker 侧的一等能力，代价是每种模式都有额外约束：顺序消息牺牲并行度，事务消息多一次回查，延迟消息占额外的调度队列。

核心机制 ​

五种类型的边界 ​

类型保证主要代价典型场景
普通至少一次投递不保证顺序通知、异步解耦
顺序同一 Queue 内 FIFO失败会阻塞该 Queue订单状态流转、binlog
延迟到期后才可见4.x 只有固定级别；5.x 有上限超时取消、重试调度
事务本地事务与消息发送最终一致需要回查接口，半消息占存储分布式事务、outbox 替代
批量一次 RPC 发多条同一 Topic、总大小受限日志、埋点批量写入

延迟消息：4.x 与 5.x 是两套实现 ​

4.x：18 个固定级别

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

发送时设置 delayTimeLevel（1 对应 1s，18 对应 2h）。Broker 把消息先写到内部 Topic SCHEDULE_TOPIC_XXXX，queueId = level - 1。ScheduleMessageService 启动时为每个级别各创建一个 DeliverDelayedMessageTimerTask，每个任务维护自己的消费 offset，到期后把消息恢复成原 Topic 再投递；offset 定期写入 config/delayOffset.json。18 个级别各有独立任务与独立 offset，只是共用同一个调度 Timer 线程。

5.x：Timer 时间轮（RIP-43）

5.0 起开源版支持任意时间的延迟消息：发送时给消息设置投递时间戳（gRPC 客户端用 setDeliveryTimestamp），Broker 把它写入 rmq_sys_wheel_timer，由时间轮按秒级精度推进。关键配置：

properties
timerWheelEnable = true          # 5.x 默认 true
timerPrecisionMs = 1000          # 精度 1s
timerMaxDelaySec = 86400         # 最大 86400s（24h），以官方版本为准

超过 timerMaxDelaySec 的消息会被拒绝（MESSAGE_ILLEGAL）。所以「5.x 支持任意时间」的准确说法是：支持任意时间点，但受最大延迟上限约束；4.x 的 18 级限制不再适用。

顺序消息：有序的代价是队头阻塞 ​

Producer 用 MessageQueueSelector 按业务键（订单号）选同一个 Queue，Broker 只保证同一 Queue 内 FIFO，不保证跨 Queue 全局有序。

消费端用 MessageListenerOrderly，同一 Queue 同一时刻只有一个线程消费。消费失败时在本地重试（默认间隔 suspendCurrentQueueTimeMillis=1000ms），重试到 maxReconsumeTimes 上限后进入死信队列，然后才放行后续消息。顺序消费不是「不重试」，而是「不换队列、原地重试」，代价是失败期间整条 Queue 停住。

事务消息：半消息 + 固定周期回查 ​

Broker 侧由 TransactionalMessageCheckService 扫描，参数含义要分清：

参数默认值含义
transactionTimeOut6000ms半消息写入后多久仍无结果就开始回查
transactionCheckInterval60000ms回查线程的固定扫描周期，不是递增间隔
transactionCheckMax15最多回查次数，超过后丢弃并记录

回查没有指数退避：每隔一个扫描周期检查一批超时半消息，直到达到次数上限。回查接口必须幂等、快速返回，否则半消息会长期占用存储。

批量消息：硬限 4 MiB ​

批量发送要求同一 Topic，单次总大小受 maxMessageSize 限制，默认 4 MiB；官方建议单批控制在 1 MiB 以内，避免大包导致发送超时和 Broker 内存抖动。延迟消息、事务消息不能放进批量。

实战场景 ​

一、顺序消息的队头阻塞：一条脏数据停住整条队列 ​

现象：订单状态 Topic 用顺序消费（MessageListenerOrderly），8 个 Queue。某天 queueId=3 的积压从 0 涨到 12 万条，其余队列正常。日志里反复出现同一条消息的处理异常，消费间隔停在默认的 1000ms 本地重试。前端表现为“一部分订单状态一直不更新”，且这部分恰好是按订单号哈希落在同一队列的那批商家。

排查：顺序消费不是“不重试”，而是“不换队列、原地重试”，代价是失败期间整条 Queue 停住。

定位层取证据判定
积压是否单队列控制台按 queueId 看积压分布单队列暴涨即队头阻塞，不是整体消费能力问题
重试次数与状态消费端日志 ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT、reconsumeTimes原地重试且未达上限 → 后续消息不放行
失败消息内容取出该消息 body，看是否触发同一个异常同一条重复失败即脏数据
死信阈值maxReconsumeTimes 配置与 %DLQ%<group> 是否有数据阈值设得过大则长期阻塞
并行度Topic 的 Queue 数与消费者实例数Queue 数就是顺序消费的最大并行度
是否误用并发监听器同一 Queue 是否被多线程消费用 Concurrently 会破坏顺序保证

对症做法：

java
// 1. 明确区分“可重试失败”和“脏数据”：脏数据不能挡住队列
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext ctx) {
    for (MessageExt m : msgs) {
        try {
            handler.handle(m);
        } catch (RetryableException e) {
            ctx.setSuspendCurrentQueueTimeMillis(2000);       // 短暂挂起，等待下游恢复
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        } catch (BadMessageException e) {
            quarantine.save(m, e);                             // 隔离区落库，人工/自动修复
            metrics.quarantine(m.getTopic(), m.getMsgId());
            // 继续返回 SUCCESS 放行后续消息，顺序性由“该消息已被记录”保证不丢
        }
    }
    return ConsumeOrderlyStatus.SUCCESS;
}
properties
# 2. 顺序消费的重试上限要小，配合死信监控
maxReconsumeTimes=3
suspendCurrentQueueTimeMillis=2000
text
3. 前置校验把脏数据挡在发送侧：状态机非法跳转、字段缺失一律不发。
4. 热点商家/大 key 造成单队列倾斜时，用“同 key 串行 + 跨 key 并行”的二级分发：
   MQ 只保证同 key 有序，业务侧再按 key 哈希到固定线程。
5. 死信监控必须存在：%DLQ%<consumerGroup> 有新增即告警，
   并提供重新投递工具，否则死信只是“把问题搬到别处”。

复盘要点：判据是“单队列积压不得显著高于其他队列”和“隔离区消息量可解释”。顺序性是业务需求而不是默认目标：只要求“同一订单有序”就不要用全局有序；一个 Topic 只建 1 个 Queue 会把吞吐退化成单机串行。死信与重试的完整口径见 死信队列与消费重试。

二、把 delayTimeLevel=15 当成“45 分钟后取消” ​

现象：订单超时取消最初用 4.x 的固定级别，业务要求“30 分钟未支付则取消”，开发选了 delayTimeLevel=15（对应 20m）凑近；后来规则改成 45 分钟，代码里改成 18（2h），结果所有订单都被延后 1 小时 15 分钟取消，客服投诉“已付款仍被取消”。升级 5.x 后又出现一批消息直接被拒 MESSAGE_ILLEGAL。

排查：4.x 的 18 级是硬编码，5.x 的 Timer 有上限，两套实现不能混为一谈。

定位层取证据判定
集群版本与实现Broker 版本；是否写 SCHEDULE_TOPIC_XXXX 或 rmq_sys_wheel_timer4.x 级别制、5.x 时间轮是两套存储
请求的延迟值消息的 delayTimeLevel 或投递时间戳只能落在 18 个级别之一，否则被就近/拒绝
5.x 上限timerMaxDelaySec（默认 86400，即 24h）超过即 MESSAGE_ILLEGAL，不是发送失败重试问题
精度timerPrecisionMs=1000秒级精度，对“准点”要求高的场景要确认
业务口径取消时间与支付截止时间差级别不匹配会整批延后或提前
4.x 可选级别：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
             （level 1..18，SCHEDULE_TOPIC_XXXX 的 queueId = level - 1）

对症做法：

java
// 5.x：按投递时间点发送，并显式校验上限，超范围在业务层拒绝
long deliverAt = order.createdAt() + cancelWindow;              // 精确到分钟
if (deliverAt - System.currentTimeMillis() > timerMaxDelayMs) {
    scheduler.register(order.id(), new Date(deliverAt));        // 超长定时交给调度系统
    return;
}
message.setDelayTimeSec((int) ((deliverAt - System.currentTimeMillis()) / 1000));
properties
# 需要非标准延迟时，5.x 侧调上限并确认存储与清理
timerWheelEnable=true
timerPrecisionMs=1000
timerMaxDelaySec=86400
text
1. 4.x 且延迟值不在 18 级里时，不要“就近选一级”：
   改为外部调度（定时任务扫表）或升级到 5.x，两者都比错级别可靠。
2. 延迟语义要与业务时间口径绑定：取消动作判断“是否已支付”要再查一次状态，
   不能只依赖延迟消息到达即执行，避免提前/延后造成误取消。
3. 超过 24h 的定时属于调度任务，不属于 MQ：见 ElasticJob/XXL-JOB 的调度语义。

复盘要点：delayTimeLevel 这种“级别枚举”最容易写出静默错误的业务规则，判据应该是“代码里不出现裸的 level 数字”——统一用 DelayLevel.of(Duration) 之类的换算工具，找不到完全匹配级别就抛异常而不是就近取整。消息被 MESSAGE_ILLEGAL 拒绝要有独立指标，它和发送超时是两类问题。

三、回查接口做了写操作，事务消息被丢 ​

现象：支付成功后要发“扣减库存”事务消息。上线后约 3% 的消息既没投递也没重试记录，Broker 日志出现 transaction check max ... discard。查回查实现，发现 checkLocalTransaction() 里除了查订单状态，还会“补发一次扣减”并写了一张标记表；同时该查询走了未加索引的从库，单次耗时 4 秒。

排查：回查是固定周期、有次数上限的，超限即丢弃。

定位层取证据判定
回查次数与结果Broker 日志该半消息的回查记录、返回的 LocalTransactionState反复 UNKNOWN 直到 transactionCheckMax=15 被丢弃
回查耗时回查接口 P99（本例 4s）慢回查让扫描线程积压，超过间隔仍未决
回查是否幂等回查里是否有写操作/副作用有写就可能重复扣减，且把回查变成业务路径
半消息存储RMQ_SYS_TRANS_HALF_TOPIC / OP 主题中的残留长期残留说明既没 COMMIT 也没 ROLLBACK
参数关系transactionTimeOut=6000ms、transactionCheckInterval=60000ms6s 后进入待回查，实际最快也要等下一个 60s 周期
Producer 存活客户端是否在提交前崩溃/被 kill崩溃后只能靠回查决定，此时回查实现就是最后防线

回查没有指数退避：每隔一个扫描周期检查一批超时半消息，直到达到次数上限后丢弃并记录。

对症做法：

java
// 1. 回查只做“状态判断”，绝不产生副作用
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    String bizId = msg.getUserProperty("payOrderId");
    PayRecord r = payRepo.findByBizId(bizId);          // 只读，走索引
    if (r == null) return LocalTransactionState.ROLLBACK_MESSAGE;
    return r.isSuccess() ? LocalTransactionState.COMMIT_MESSAGE
                         : LocalTransactionState.ROLLBACK_MESSAGE;
}
text
2. 返回 UNKNOWN 必须有前提：能明确判定的场景一律返回 COMMIT/ROLLBACK，
   只把“确实暂时查不到”的情况留给下一轮。
3. 回查依赖的查询必须走索引并在毫秒级返回（本例补 pay_record(biz_id) 唯一索引）。
4. 客户端提交前崩溃是常态：半消息机制的价值就在这里，
   所以本地事务状态表要能仅凭消息里的业务键判定成败（bizId 唯一索引 + 状态列）。
5. 超过 transactionCheckMax 的丢弃是不可逆的：
   要么把上限调大（存储会占更久），要么业务侧加对账把漏发补回来。

复盘要点：判据是“半消息存活时间 P99 应接近一个 transactionCheckInterval 周期”，出现“存活到第 15 次”就是回查逻辑在拖延。事务消息保证的是“本地事务与消息发送最终一致”，不保证消费端恰好一次，消费侧仍需幂等；对账兜底方案见 可靠事件投递：本地消息表与幂等。

最小上手 ​

4.9 客户端，四类消息各一段（省略 try/catch 与 producer.shutdown()）：

java
DefaultMQProducer producer = new DefaultMQProducer("pg-demo");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();

// 1) 普通消息：同步发送，失败默认重试 2 次
producer.send(new Message("t-normal", "A", "hello".getBytes()));

// 2) 顺序消息：按订单号选同一个 Queue
Message order = new Message("t-order", "PAID", "order-1".getBytes());
producer.send(order, (queues, msg, arg) ->
    queues.get(Math.floorMod(arg.hashCode(), queues.size())), "order-1");

// 3) 延迟消息（4.x）：level 3 = 10s
Message delay = new Message("t-delay", "T", "later".getBytes());
delay.setDelayTimeLevel(3);
producer.send(delay);

// 4) 批量消息：同一 Topic，总大小 < maxMessageSize
producer.send(List.of(new Message("t-batch", "B", "m1".getBytes()),
                      new Message("t-batch", "B", "m2".getBytes())));

事务消息需要单独的 TransactionMQProducer：

java
TransactionMQProducer tx = new TransactionMQProducer("pg-tx");
tx.setNamesrvAddr("127.0.0.1:9876");
tx.setTransactionListener(new TransactionListener() {
    public LocalTransactionState executeLocalTransaction(Message m, Object a) {
        return LocalTransactionState.COMMIT_MESSAGE;      // 本地事务成功
    }
    public LocalTransactionState checkLocalTransaction(MessageExt m) {
        return LocalTransactionState.COMMIT_MESSAGE;      // 必须幂等
    }
});
tx.start();
tx.sendMessageInTransaction(new Message("t-tx", "T", "pay".getBytes()), null);

5.x 任意时间延迟（gRPC 客户端 ≥ 5.0）：

java
Message message = provider.newMessageBuilder()
    .setTopic("t-delay").setBody("later".getBytes())
    .setDeliveryTimestamp(System.currentTimeMillis() + 35_000)  // 35 秒后
    .build();
provider.send(message);

验证：

bash
# 看队列 min/max offset 是否增长
sh bin/mqadmin topicStatus -n 127.0.0.1:9876 -t t-order
# 看消费进度；diff 为 0 表示追上
sh bin/mqadmin consumerProgress -n 127.0.0.1:9876 -g cg-demo
# 按业务 key 反查消息（发送时需 setKeys）
sh bin/mqadmin queryMsgByKey -n 127.0.0.1:9876 -t t-order -k order-1

生产动作 ​

  • 延迟消息先确认版本与上限：4.x 只能选 18 级；5.x 配 timerMaxDelaySec，Timer 与 4.x 的 SCHEDULE_TOPIC_XXXX 不是同一套存储。
  • 顺序消息的 Queue 数就是最大并行度，失败会阻塞整条 Queue。必须设置合理的 maxReconsumeTimes 并监控 %DLQ%。
  • 事务回查接口做成只读状态查询，不要在执行回查时做写操作；调大 transactionCheckInterval 只会让不确定状态持续更久。
  • 批量大小按 1 MiB 控制，并给 sendMsgTimeout（默认 3000ms）留余量；批量不是提高单条可靠性的手段。
  • 顺序消息不要用并发监听器，否则同 Queue 会被多线程并发消费，顺序保证失效。

边界 ​

  • 全局有序不现实：一个 Topic 只建 1 个 Queue 会让吞吐退化成单机串行，只在极低流量场景考虑。
  • 4.x 的 18 级是硬编码，不能通过配置增加；需要任意时间只能升级到 5.x 或自己做外部调度。
  • 事务消息只保证「本地事务 + 消息发送」的最终一致，不保证消费端恰好一次；消费端仍需幂等。
  • 延迟消息在 5.x 也有上限（默认 24h，以官方文档为准），超长定时任务应该用调度系统而不是 MQ；批量消息不能跨 Topic，也不能混用延迟/事务。

讲清楚 ​

RocketMQ 的消息类型，本质是把几种常见的发送语义固化到 Broker。普通消息最简单，至少一次投递，不保证顺序。顺序消息靠 Producer 把同一业务键路由到同一个 Queue，消费端加锁串行；失败不是不重试，而是原地重试到上限后进死信，这期间整条 Queue 会停住，所以它是用并行度换顺序。

延迟消息要分版本说。4.x 只有 18 个固定级别，Broker 为每个级别各起一个投递任务、各自维护 offset；5.x 用时间轮支持任意时间点，但有最大延迟上限，默认 24 小时。事务消息是半消息加回查：本地事务结果没送达时，Broker 按固定周期扫描超时半消息并回调生产者，没有指数退避，所以回查接口必须幂等且快。批量消息则是同一 Topic 下合并发送，硬上限是 maxMessageSize，默认 4 MiB。

选型的判断顺序是：先问要不要有序，再问要不要定时，再问要不要和本地事务一致，最后才看批量。每加一个语义，都会多一份约束和故障面。

把一门技术讲透到既懂也会用

技术小册 · 持续更新