RocketMQ 顺序消息实战：全局与分区机制详解

消费者：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class GlobalOrderConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("GlobalOrderConsumerGroup");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("GlobalOrderTopic", "*");
        consumer.setMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String messageBody = new String(msg.getBody(), "UTF-8");
                        System.out.println("消费消息：" + messageBody + ", QueueId: " + msg.getQueueId());
                        Thread.sleep(500);
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("全局顺序消费者启动成功");
    }
}

局限性：性能低、风险高、扩展难，仅适用于极小规模场景。

分区顺序消息的实现

更实用。核心思想：相同业务键消息发往同一 Queue。

例如电商系统，用订单 ID 做业务键。同一订单操作（创建、支付、发货）进入同一队列，不同订单并行。

Java 代码示例

生产者：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import java.util.List;

public class PartitionOrderProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("PartitionOrderProducerGroup");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        String[] orderIds = {"ORDER_001", "ORDER_002", "ORDER_003"};
        try {
            for (String orderId : orderIds) {
                String[] operations = {"CREATE", "PAY", "SHIP", "COMPLETE"};
                for (String operation : operations) {
                    String messageBody = orderId + " - " + operation;
                    Message msg = new Message("PartitionOrderTopic", "OrderTag", messageBody.getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            String id = (String) arg;
                            int index = Math.abs(id.hashCode()) % mqs.size();
                            return mqs.get(index);
                        }
                    }, orderId);
                    System.out.println("发送消息：" + messageBody + ", QueueId: " + sendResult.getMessageQueue().getQueueId());
                    Thread.sleep(100);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.shutdown();
        }
    }
}

消费者：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class PartitionOrderConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("PartitionOrderConsumerGroup");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("PartitionOrderTopic", "*");
        consumer.setMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String messageBody = new String(msg.getBody(), "UTF-8");
                        System.out.println("消费消息：" + messageBody + ", QueueId: " + msg.getQueueId());
                        Thread.sleep(1000);
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("分区顺序消费者启动成功");
    }
}

关键点：

1. MessageQueueSelector：控制路由，确保同键同队列。
2. 业务键选择：唯一标识实体（订单 ID、用户 ID）。
3. Queue 数量：影响并发度，需权衡负载。

顺序消息的消费机制

ConsumeOrderlyStatus 枚举

• SUCCESS：消费成功，继续下一条。
• SUSPEND_CURRENT_QUEUE_A_MOMENT：消费失败，暂停当前队列，稍后重试。

注意：顺序消息不支持并发消费，即使配置多线程，RocketMQ 也会强制单线程处理每个 Queue。

消费失败处理

失败消息会重回队列头部并暂停消费（默认 1 秒）。这防止了跳过消息，但也可能导致阻塞。

应对策略：

• 设置最大重试次数 consumer.setMaxReconsumeTimes()。
• 死信队列处理超时消息。
• 区分业务异常与系统异常，避免无效重试。

@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
    for (MessageExt msg : msgs) {
        try {
            processMessage(msg);
        } catch (BusinessException e) {
            log.error("业务异常，跳过消息：{}", msg.getMsgId(), e);
            return ConsumeOrderlyStatus.SUCCESS;
        } catch (Exception e) {
            log.error("系统异常，重试消息：{}", msg.getMsgId(), e);
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }
    return ConsumeOrderlyStatus.SUCCESS;
}

实际应用场景

1. 电商订单流转：创建 → 支付 → 发货。用订单 ID 做键，防止未付先发货。
2. 用户行为轨迹：浏览 → 加购 → 下单。按用户 ID 排序分析路径。
3. 数据库 binlog 同步：表名 + 主键做键，确保记录变更顺序一致。
4. 物联网指令下发：设备 ID 做键，保证开机 → 设置 → 工作 → 关机顺序。

性能优化与最佳实践

合理设计业务键

• 粒度适中：过大退化为全局顺序，过小失去意义。
• 分布均匀：Hash 值尽量分散，避免热点队列。

优化 Queue 数量

• 消息量大、键多：增加 Queue 数提升并发。
• 建议设置为 Broker 数量的整数倍。

监控与告警

关注消费延迟、失败率、重试次数。结合 Prometheus 和 Grafana 可视化。

常见问题

问题 1：消息顺序错乱

• 原因：Selector 未生效、使用了并发消费模式、业务键不当。
• 解决：检查 Selector 逻辑，确认使用 MessageListenerOrderly。

问题 2：消费性能低下

• 原因：Queue 过少、处理逻辑复杂、频繁重试。
• 解决：增加 Queue 数，优化逻辑，合理处理失败。

问题 3：消费阻塞

• 原因：某条消息持续失败阻塞后续消息。
• 解决：完善异常处理，对不可恢复异常返回 SUCCESS 跳过，设置最大重试次数。

与其他中间件对比

RocketMQ vs Kafka

• Kafka：Partition 级别顺序，天然支持顺序读写。
• RocketMQ：API 更友好，支持自定义路由策略。

RocketMQ vs RabbitMQ

• RabbitMQ：原生不支持顺序，需单队列或客户端排序，效率较低。
• RocketMQ：分区顺序方案更高效优雅。

高级特性

事务消息与顺序消息结合

金融转账等场景需同时保证顺序和事务。可使用 TransactionMQProducer 配合 MessageQueueSelector。

延迟消息与顺序消息

订单超时取消可用延迟消息。注意：延迟期间不保序，到期后才开始顺序消费。

总结

RocketMQ 的分区顺序机制在分布式系统中提供了优雅的解决方案。通过业务键分组，既保证了逻辑正确性，又获得了良好性能。关键在于合理设计路由策略、正确处理消费失败以及做好监控。在实际项目中，根据需求选择全局或分区顺序，并注意性能调优。