RocketMQ-如何保证顺序消息

我会带着你远行 2023-10-14 09:51 154阅读 0赞

# 1. 简介 #

实际开发中会有以下场景，需要保证一组消息的生产顺序与消费顺序相同，例如

1.  监听数据库表单条数据的的多次修改，需要保证监听者最终得到的消息顺序和数据库表对单条数据的修改顺序一样
2.  网购平台创建订单的过程一般都是异步实现，订单创建和支付流程需要保证最终到服务器的顺序一致

那么在 RocketMQ 中该如何保证消息的顺序呢？

# 2. RocketMQ 中该如何保证消息的顺序呢？ #

首先查看一条消息从生产者到消费者的流程：

![image.png][]

![get_code_YjllYzUzNmY1ZWQyOTc4NWFkNGE3Y2VkODFmOTg1NmYsMTY5MDY4NTI1NDI5Mw][]

*  生产者： 生产者可以有多个 每个生产者可以向 topic 的任意队列写数据
 *  消费者 消费者存在于消费者组当中，消费者组可以存在多个消费者 每一个队列在同一时间只能被同一消费者组内的一个消费者消费，但是同一消费者可以同时消费多个队列

想要保证消息的顺序需要满足以下条件：

1.  生产者生产消息保证顺序一致
2.  生产者生产的消息最终到达 broker 且被处理的顺序一致
3.  broker 保证先后到达消息最终处理的结果顺序一致
4.  消费者可以按照固定顺序去拉取并处理消息

下面我们以一条订单的创建看满足这些条件需要做些什么： 假设一条订单从创建到支付有四个节点，创建订单、提交支付、支付、支付成功

1.  用户在页面操作可以保证四个节点必定是按照固定顺序去生产的四条消息，但是最终到达 broker 的顺序却不一定按照生产者发起消息请求到 broker 的顺序，因为网络交互过程是存在不确定性的，先出发的可能后到，后出发的可能先到。
    
    ![image.png][image.png 1]
2.  所以在生产者发送消息时需要由生产者去保证消息顺序，在创建订单完成后再去发送提交支付请求，在提交支付请求完成后再去发送支付请求
3.  broker 保证消息顺序一致，可以查看大标题 《3. broker 与 消费者如何保证消息顺序》
4.  消费者保证顺序一致性，一个队列只能被一个消费者消费，但是同一个消费者可以消费多个队列，生产的消息如果落到不同的队列上有以下两种情况 这些队列被不同消费者所消费：不同消费者消费互不干涉，互不影响所以会出现消费顺序不一致的情况 这些队列被同一个消费者消费：但是消费者在向 broker 拉取消息时是按照以队列id为单位去不断拉取，在同一个队列id下上一条消息消费完成才会去拉去下一条消息，也就是说如果四个节点的消息落到不同队列，最终也会导致消费顺序不一致。例如，创建订单落在队列0，提交支付落在队列1。但是队列 1的消费速度较快所以提交支付的消息就会被先消费。导致消费顺序不一致。
5.  所以如果要保证消息顺序一致必须要保证消息落在同一个队列上 一个简单的方法，可以把订单号对队列数量取余就可以满足，int queueId = orderId % queueNum;
6.  最终消息按照顺序到达了消费者，可是 RocketMQ 默认的消费者监听为了性能将这些消息用多线程进行了消费，最终必然会导致消费顺序不一致，当然 RokectMQ 提供了两种可选方案 MessageListenerOrderly：消息按照顺序进行消费，每一个队列下的消息用一个线程去处理 MessageListenerConcurrently：并行的去处理消息，提高消费消息的性能

# 3. broker 与 消费者如何保证消息顺序 #

![image.png][image.png 2]

![get_code_N2NiODJkNTg5NDdhOTU5YWRjMmQ5NDU0YTE0NDYxOTYsMTY5MDY4NTI1NDI5Mw][]

1.  消息进入 broker 加锁按顺序进入 commitLog
2.  commitLog 重消费线程单线程处理放入 topic 的不同队列当中
3.  消费者按照 topic、queueId、消费偏移量去拉取一组消息
4.  消费者处理完这一组消息后再去根据新的消费偏移量按照顺序拉取消息

# 4. 代码实现 #

# 4.1 生产者 #

java复制代码        DefaultMQProducer producer = new DefaultMQProducer("order_producer_group");
            producer.setNamesrvAddr("127.0.0.1:9876");
            producer.setSendMsgTimeout(1000 * 15);
            producer.start();
    
            List<String> orderProcess = Arrays.asList("创建订单", "提交支付", "支付", "支付成功");
    
            for (int i = 0; i < 4; i++) {
                int orderId = i;
                new Thread(() -> {
                    for (String process : orderProcess) {
                        try {
                            producer.send(
                                    new Message("order_topic",
                                            String.format("orderId:%s，process：%s", orderId, process).getBytes(StandardCharsets.UTF_8)),
                                    new MessageQueueSelector() {
                                        @Override
                                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                                            Integer orderId = (Integer) arg;
                                            int index = orderId % mqs.size();
                                            return mqs.get(index);
                                        }
                                    },
                                    orderId
                            );
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
            }

# 4.2 消费者 #

java复制代码        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer_group");
            consumer.setNamesrvAddr("127.0.0.1:9876");
            /**
             * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
             * 如果非第一次启动，那么按照上次消费的位置继续消费
             */
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            consumer.subscribe("order_topic", "*");
    
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    context.setAutoCommit(true);
    
                    for (MessageExt msg : msgs) {
                        // 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序
                        System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody()));
                    }
    
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
    
            consumer.start();
    
            System.out.println("Consumer Started.");

# 4.3 消费者结果 #

> consumeThread=ConsumeMessageThread\_1queueId=1, content:orderId:1，process：创建订单  
> consumeThread=ConsumeMessageThread\_2queueId=3, content:orderId:3，process：创建订单  
> consumeThread=ConsumeMessageThread\_3queueId=0, content:orderId:0，process：创建订单  
> consumeThread=ConsumeMessageThread\_4queueId=2, content:orderId:2，process：创建订单  
> consumeThread=ConsumeMessageThread\_1queueId=1, content:orderId:1，process：提交支付  
> consumeThread=ConsumeMessageThread\_2queueId=3, content:orderId:3，process：提交支付  
> consumeThread=ConsumeMessageThread\_4queueId=2, content:orderId:2，process：提交支付  
> consumeThread=ConsumeMessageThread\_1queueId=1, content:orderId:1，process：支付  
> consumeThread=ConsumeMessageThread\_3queueId=0, content:orderId:0，process：提交支付  
> consumeThread=ConsumeMessageThread\_2queueId=3, content:orderId:3，process：支付  
> consumeThread=ConsumeMessageThread\_1queueId=1, content:orderId:1，process：支付成功  
> consumeThread=ConsumeMessageThread\_4queueId=2, content:orderId:2，process：支付  
> consumeThread=ConsumeMessageThread\_3queueId=0, content:orderId:0，process：支付  
> consumeThread=ConsumeMessageThread\_2queueId=3, content:orderId:3，process：支付成功  
> consumeThread=ConsumeMessageThread\_4queueId=2, content:orderId:2，process：支付成功  
> consumeThread=ConsumeMessageThread\_3queueId=0, content:orderId:0，process：支付成功

最终消费结果每一个订单都按照顺序消费，且每一个队列都被一个线程所处理

[image.png]: https://img-blog.csdnimg.cn/img_convert/3b065ea29830284e6b80a08f5975d693.webp?x-oss-process=image/format,png
[get_code_YjllYzUzNmY1ZWQyOTc4NWFkNGE3Y2VkODFmOTg1NmYsMTY5MDY4NTI1NDI5Mw]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=YjllYzUzNmY1ZWQyOTc4NWFkNGE3Y2VkODFmOTg1NmYsMTY5MDY4NTI1NDI5Mw==
[image.png 1]: https://img-blog.csdnimg.cn/img_convert/d4ea3ae02787116e6a13bc5ee58e50ff.webp?x-oss-process=image/format,png
[image.png 2]: https://img-blog.csdnimg.cn/img_convert/b8f35272fddec01409a8cae41e219cf5.webp?x-oss-process=image/format,png
[get_code_N2NiODJkNTg5NDdhOTU5YWRjMmQ5NDU0YTE0NDYxOTYsMTY5MDY4NTI1NDI5Mw]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=N2NiODJkNTg5NDdhOTU5YWRjMmQ5NDU0YTE0NDYxOTYsMTY5MDY4NTI1NDI5Mw==