RocketMQ 如何保证消息的顺序性

创作时间:

作者:

@小白创作中心

RocketMQ 如何保证消息的顺序性

引用

CSDN

https://blog.csdn.net/weixin_43281498/article/details/141340201

RocketMQ 与 Kafka 类似，仅支持同一队列内的消息顺序性。为了确保消息的顺序性，需要从生产者发送、Broker 存储和消费者消费三个方面进行保证。

如何保证生产者发送消息的顺序性？

Apache RocketMQ 通过生产者和服务端的协议保障单个生产者串行地发送消息，并按序存储和持久化。要保证消息生产的顺序性，必须满足以下条件：

单一生产者：消息生产的顺序性仅支持单一生产者。不同生产者分布在不同的系统中，即使设置相同的消息组，不同生产者之间产生的消息也无法判定其先后顺序。
串行发送（同步发送）：Apache RocketMQ 生产者客户端支持多线程安全访问，但如果生产者使用多线程并行发送，则不同线程间产生的消息将无法判定其先后顺序。

如何保证队列中数据的顺序性？

当作为 MQ 的生产者需要发送顺序消息时，需要在 send 方法中传入一个 MessageQueueSelector。MessageQueueSelector 中需要实现一个 select 方法，这个方法用来定义要把消息发送到哪个 MessageQueue。通常可以使用取模法进行路由：

SendResult sendResult = producer.send(msg, new MessageQueueSelector(){
    @Override
    //mqs:该Topic下所有可选的MessageQueue
    //msg:待发送的消息
    //arg:发送消息时传递的参数
    public MessageQueue select(List<MessageQueue>mgs,Message msg,Object arg){
        Integer id=(Integer)arg;
        //根据参数，计算出一个要接收消息的MessageQueue的下标
        int index= id % mgs.size();
        //返回这个MessageQueue
        return mgs.get(index);
    }
}, orderId);

通过以上形式就可以将需要有序的消息发送到同一个队列中。需要注意的是，这里需要使用同步发送的方式。

将顺序消息发送至 Apache RocketMQ 后，会保证设置了同一消息组的消息，按照发送顺序存储在同一队列中。服务端顺序存储逻辑如下：

相同消息组的消息按照先后顺序被存储在同一个队列。
不同消息组的消息可以混合在同一个队列中，且不保证连续。

如上图所示，消息组 1 和消息组 4 的消息混合存储在队列 1 中，Apache RocketMQ 保证消息组 1 中的消息 G1-M1、G1-M2、G1-M3 是按发送顺序存储，且消息组 4 的消息 G4-M1、G4-M2 也是按顺序存储，但消息组 1 和消息组 4 中的消息不涉及顺序关系。

如何保证消费者消费的顺序性？

RocketMQ 的 MessageListener 回调函数提供了两种消费模式：

有序消费模式 MessagelistenerOrderly
并发消费模式 MessageListenerConcurrently

想要实现顺序消费，需要使用 MessageListenerOrderly 模式接收消息：

consumer.registerMessageListener(new MessageListenerOrderly(){
    Override
    public Consume0rderlystatus consumeMessage(List<MessageExt>msgs ,ConsumeOrderlyContext context) {
        System.out.printf("Receive order msg:"+ new string(msgs.get(@).getBody()));
        return ConsumeOrderlystatus.SUCCESS;
    }
});

当我们用以上方式注册一个消费之后，为了保证同一个队列中的有序消息可以被顺序消费，就要保证 RocketMQ 的 Broker 只会把消息发送到同一个消费者上，这时候就需要加锁了。

在实现中，ConsumeMessageOrderyService 初始化的时候，会启动一个定时任务会尝试向 Broker 为当前消费者客户端申请分布式锁。如果获取成功，那么后续消息将会只发给这个 Consumer。（第一把锁）

接下来在消息拉取的过程中，消费者会一次性拉取多条消息的，并且会将拉取到的消息放入 ProcessQueue，同时将消息提交到消费线程池进行执行。

那么拉取之后的消费过程，怎么保证顺序消费呢？这里就需要更多的锁了。

RocketMQ 在消费的过程中，需要申请 MessageQueue 锁，确保在同一时间，一个队列中只有一个线程能处理队列中的消息。（第二把锁）

获取到 MessageQueue 的锁后，就可以从 ProcessQueue 中依次拉取一批消息处理了，但是这个过程中，为了保证消息不会出现重复消费，还需要对 ProcessQueue 进行加锁。（第三把锁）