RocketMQ 实现顺序消息（Ordered Message） 的核心思想是：保证同一业务标识（如订单 ID）的消息，在发送、存储和消费三个阶段都严格按照发送顺序被处理。这是通过 “局部有序”（Partition/Queue 级别有序）来实现的，而非全局有序（全局有序性能极差，一般不推荐）。

下面从原理、实现方式、代码示例、注意事项等方面详细说明 RocketMQ 如何实现顺序消息。

顺序消息的类型

RocketMQ 支持两种顺序消息：

类型 说明 适用场景
全局顺序消息 所有消息严格按发送顺序消费 性能极低，仅用于极特殊场景（如金融对账）
分区顺序消息（局部顺序） 同一业务 Key（如订单 ID）的消息有序，不同 Key 之间可并行 电商订单、状态机变更等主流场景

生产环境几乎只用“分区顺序消息”，因为全局顺序要求 Topic 只有一个 Queue，吞吐量严重受限。

实现原理

核心总结：同一业务 Key → 同一 Queue → 单线程消费 = 顺序保证

各阶段如何保证顺序

  1. 发送阶段：消息路由到同一个 Queue

    生产者通过 MessageQueueSelector 接口，根据业务 Key（如 orderId）选择固定的 MessageQueue。

    例如：所有 orderId=1001 的消息都发送到 Queue 0，orderId=1002 的消息都发送到 Queue 1。

    这样保证同一个业务实体的消息落在同一个 Queue 中。

  2. 存储阶段：Queue 内部 FIFO

    RocketMQ 的每个 Queue 本质是一个顺序写入的日志文件（CommitLog 分片）。

    同一 Queue 中的消息天然按写入顺序存储，保证存储有序。

  3. 消费阶段：单线程顺序消费

    消费者对每个 Queue 使用单线程拉取消息并处理。

    RocketMQ 的 PushConsumer 默认为每个 Queue 分配一个消费线程（可通过配置调整，但顺序消费必须单线程）。

    如果消费失败，会暂停该 Queue 的消费，直到成功或跳过，避免乱序。

实现的关键机制

  1. Queue 锁机制（消费端）

    RocketMQ 为每个 Queue 在消费端加锁（分布式锁，基于 Broker）。保证同一时间，一个 Queue 只能被一个 Consumer 实例消费，确保单线程处理。如果 Consumer 宕机，Broker 会在 20 秒后释放锁，由其他实例接管。

  2. 消息重试策略

    顺序消息不进入重试队列。失败时会暂停当前 Queue 的消费，持续重试，直到成功或人工干预。

    可通过 context.setSuspendCurrentQueueTimeMillis(1000) 自定义暂停时间（最大 30 秒）。

  3. 性能与扩展性

    顺序性是以牺牲并行度为代价的。

    建议：合理设计业务 Key，避免热点（如所有消息都用同一个 Key，导致所有流量压到一个 Queue）。

    可通过增加 Queue 数量提升吞吐（如 16 个 Queue，16 个订单 ID 均匀分布）。

Demo 示例

demo代码Java实现

生产者

使用 MessageQueueSelector

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
DefaultMQProducer producer = new DefaultMQProducer("OrderProducerGroup");
producer.setNamesrvAddr("localhost:9876");
producer.start();

String[] orderIds = {"1001", "1002", "1001", "1002"};

for (int i = 0; i < orderIds.length; i++) {
    String orderId = orderIds[i];
    Message msg = new Message(
        "OrderTopic",
        "OrderTag",
        ("订单操作-" + i + ", orderId=" + orderId).getBytes(RemotingHelper.DEFAULT_CHARSET)
    );

    // 关键：根据 orderId 选择 Queue
    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
        @Override
        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
            String orderId = (String) arg;
            int index = orderId.hashCode() % mqs.size(); // 础但有效
            return mqs.get(index);
        }
    }, orderId); // 将 orderId 作为 arg 传入 selector

    System.out.printf("发送结果: %s, QueueId: %d%n", sendResult.getSendStatus(), sendResult.getMessageQueue().getQueueId());
}

producer.shutdown();

消费者

使用顺序消费监听器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("OrderConsumerGroup");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("OrderTopic", "OrderTag");

// 注册顺序消息监听器
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
            String orderId = getOrderIdFromMsg(msg); // 从消息中解析 orderId
            System.out.println("消费消息: " + new String(msg.getBody()) + ", QueueId: " + msg.getQueueId());
            
            // 模拟业务处理
            try {
                // 处理订单状态变更逻辑
                processOrder(orderId, msg);
            } catch (Exception e) {
                // 消费失败：返回 SUSPEND_CURRENT_QUEUE_A_MOMENT，稍后重试
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        }
        // 消费成功
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

consumer.start();

典型应用场景

  1. 电商订单状态流转
    创建 → 支付 → 发货 → 完成，必须严格有序。

  2. 数据库 Binlog 同步
    同一行数据的多次变更必须按序应用，否则数据错乱。

  3. 玩家游戏操作日志
    移动 → 攻击 → 使用技能，顺序影响游戏逻辑。

  4. 金融交易流水
    同一账户的多笔交易需按时间顺序处理。