rocketmq 怎么保证消息消费的顺序 代码实现

RocketMQ可以通过设置顺序消息来保证消息的消费顺序。即按照消息的发送顺序来消费消息，这种方式只能保证顺序消息的有序消费，普通消息的消费仍然是无序的。

在RocketMQ中，顺序消息是指在同一个消息队列中，按顺序发送的消息。如果我们需要保证消息的顺序消费，需要做以下几个步骤：

1. 创建顺序消息生产者

DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("localhost:9876");
producer.start();

2. 设置顺序消息

Message message = new Message("topic", "tag", "key", "body".getBytes());
// 设置顺序消息的业务ID，用来保证消息发送的顺序
message.setKeys("order_id");
// 发送顺序消息
SendResult sendResult = producer.send(message, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
        // 获取订单ID
        String orderId = (String) o;
        // 计算订单ID的hash值
        int index = Math.abs(orderId.hashCode()) % list.size();
        // 选择消息队列
        return list.get(index);
    }
}, "order_id");

3. 创建顺序消息消费者

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("topic", "tag");
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext context) {
        // 消费消息
        for (MessageExt messageExt : list) {
            System.out.println(new String(messageExt.getBody()));
        }
        // 返回消费状态
        return ConsumeOrderlyStatus.SUCCESS;
    }
});
consumer.start();

在顺序消息的发送端，我们需要设置顺序消息的业务ID，并使用`MessageQueueSelector`选择消息队列。在顺序消息的消费端，我们需要注册`MessageListenerOrderly`监听器，按照顺序消费消息，最后返回消费状态。

需要注意的是，为了保证顺序消息的顺序性，我们需要保证同一个业务ID的消息被发送到同一个消息队列中，这样才能保证消息的消费顺序。