RocketMQ之顺序消费：Demo及实现原理分析

小咪咪 2023-07-18 09:15 11阅读 0赞

**目录**

全局顺序消费

局部顺序消费

Demo

源码分析

顺序消息重试机制

--------------------

场景分析

**顺序消费**：是指消息的产生顺序和消费顺序相同，按照FIFO先进先出的原则严格保持一致。

假设有个下单场景，每个阶段需要发邮件通知用户订单状态变化。用户付款完成时系统给用户发送订单已付款邮件，订单已发货时给用户发送订单已发货邮件，订单完成时给用户发送订单已完成邮件。

发送邮件的操作为了不阻塞订单主流程，可以通过mq消息来解耦，下游邮件服务器收到mq消息后发送具体邮件，已付款邮件、已发货邮件、订单已完成邮件这三个消息，下游的邮件服务器需要顺序消费这3个消息并且顺序发送邮件才有意义。否则就会出现已发货邮件先发出，已付款邮件后发出的情况。

但是mq消费者往往是集群部署，一个消费组内存在多个消费者，同一个消费者内部，也可能存在多个消费线程并行消费，如何在消费者集群环境中，如何保证邮件mq消息发送与消费的顺序性呢？

顺序消费又分两种，全局顺序消费和局部顺序消费

### 全局顺序消费 ###

什么是全局顺序消费？所有发到mq的消息都被顺序消费，类似数据库中的binlog，需要严格保证全局操作的顺序性

那么RocketMQ中如何做才能保证全局顺序消费呢？

这就需要设置topic下读写队列数量为1，即quene队列只能为同一个；

为什么要设置读写队列数量为1呢？  
假设读写队列有多个，消息就会存储在多个队列中，消费者负载时可能会分配到多个消费队列同时进行消费，多队列并发消费时，无法保证消息消费顺序性

那么全局顺序消费有必要么？  
A、B都下了单，B用户订单的邮件先发送，A的后发送，不行么？其实，大多数场景下，mq下只需要保证局部消息顺序即可，即A的付款消息先于A的发货消息即可，A的消息和B的消息可以打乱，这样系统的吞吐量会更好，将队列数量置为1，极大的降低了系统的吞吐量，不符合mq的设计初衷

举个例子来说明局部顺序消费。

假设订单A的消息为A1，A2，A3，发送顺序也如此。订单B的消息为B1，B2，B3，A订单消息先发送，B订单消息后发送

消费顺序如下  
A1，A2，A3，B1，B2，B3是全局顺序消息，严重降低了系统的并发度  
A1，B1，A2，A3，B2，B3是局部顺序消息，可以被接受  
A2，B1，A1，B2，A3，B3不可接收，因为A2出现在了A1的前面

### 局部顺序消费 ###

那么在RocketMQ里局部顺序消息又是如何怎么实现的呢？

要保证消息的顺序消费，有三个关键点

*  消息顺序发送
 *  消息顺序存储
 *  消息顺序消费

第一点，消息顺序发送，多线程发送的消息无法保证有序性，因此，需要业务方在发送时，针对同一个业务编号(如同一笔订单)的消息需要保证在一个线程内顺序发送，在上一个消息发送成功后，在进行下一个消息的发送。对应到mq中，消息发送方法就得使用同步发送，异步发送无法保证顺序性

第二点，消息顺序存储，mq的topic下会存在多个queue，要保证消息的顺序存储，同一个业务编号的消息需要被发送到一个queue中。对应到mq中，需要使用MessageQueueSelector来选择要发送的queue，即对业务编号进行hash，然后根据队列数量对hash值取余，将消息发送到一个queue中

第三点，消息顺序消费，要保证消息顺序消费，同一个queue就只能被一个消费者所消费，因此对broker中消费队列加锁是无法避免的。同一时刻，一个消费队列只能被一个消费者消费，消费者内部，也只能有一个消费线程来消费该队列。即，同一时刻，一个消费队列只能被一个消费者中的一个线程消费

上面第一、第二点中提到，要保证消息顺序发送和消息顺序存储需要使用mq的同步发送和MessageQueueSelector来保证，具体Demo会有体现

至于第三点中的加锁操作会结合源码来具体分析

### Demo ###

producer中模拟了两个线程，并发顺序发送100个消息的情况，发送的消息中，key为消息发送编号i，消息body为orderId，大家注意下MessageQueueSelector的使用；

consumer的demo有两个，第一个为正常集群消费的consumer，另外一个是顺序消费的consumer，从结果中观察消息消费顺序；

理想情况下消息顺序消费的结果应该是，同一个orderId下的消息的编号i值应该顺序递增，但是不同orderId之间的消费可以并行，即局部有序即可；

**Producer Demo**

public class Producer {
        public static void main(String[] args)  {
            try {
                MQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
                ((DefaultMQProducer) producer).setNamesrvAddr("111.231.110.149:9876");
                producer.start();
                
                //顺序发送100条编号为0到99的，orderId为1 的消息
                new Thread(() -> {
                    Integer orderId = 1;
                    sendMessage(producer, orderId);
                }).start();
                //顺序发送100条编号为0到99的，orderId为2 的消息
                new Thread(() -> {
                    Integer orderId = 2;
                    sendMessage(producer, orderId);
                }).start();
                //sleep 30秒让消息都发送成功再关闭
                Thread.sleep(1000*30);
    
                producer.shutdown();
            } catch (MQClientException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        private static void sendMessage(MQProducer producer, Integer orderId) {
            for (int i = 0; i < 100; i++) {
                try {
                    Message msg =
                            new Message("TopicTestjjj", "TagA", i + "",
                                    (orderId + "").getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            Integer id = (Integer) arg;
                            int index = id % mqs.size();
                            return mqs.get(index);
                        }
                    }, orderId);
                    System.out.println("message send,orderId:"+orderId);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

**Normal Consumer Demo**  
模拟了一个消费者中多线程并行消费消息的情况，使用的消费监听器为**MessageListenerConcurrently**

public class Consumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
    
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
    
            consumer.setNamesrvAddr("111.231.110.149:9876");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            consumer.subscribe("TopicTestjjj", "*");
            //单个消费者中多线程并行消费
            consumer.setConsumeThreadMin(3);
            consumer.setConsumeThreadMin(6);
    
            consumer.registerMessageListener(new MessageListenerConcurrently() {
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
    //                    System.out.println("收到消息," + new String(msg.getBody()));
                        System.out.println("queueId:"+msg.getQueueId()+",orderId:"+new String(msg.getBody())+",i:"+msg.getKeys());
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            consumer.start();
    
            System.out.printf("Consumer Started.%n");
        }
    }

看下结果输出，如图，同一个orderId下，编号为10的消息先于编号为9的消息被消费，不是正确的顺序消费，即普通的并行消息消费，无法保证消息消费的顺序性

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center][]

**Order Consumer Demo**  
顺序消费的消费者例子如下，使用的监听器是**MessageListenerOrderly**

public class Consumer {
    
        public static void main(String[] args) throws MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
            consumer.setNamesrvAddr("111.231.110.149:9876");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            consumer.subscribe("TopicTestjjj", "TagA");
    
            //消费者并行消费
            consumer.setConsumeThreadMin(3);
            consumer.setConsumeThreadMin(6);
    
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
    //                context.setAutoCommit(false);
                    for (MessageExt msg : msgs) {
                        System.out.println("queueId:"+msg.getQueueId()+",orderId:"+new String(msg.getBody())+",i:"+msg.getKeys());
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
    
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }
    
    }

结果如下，同一个orderId下，消息顺序消费，不同orderId并行消费，符合预期

###    源码分析 ###

在源码分析之前，先来思考下几个问题

前面已经提到实现消息顺序消费的关键点有三个，其中前两点已经明确了解决思路

*  第一点，消息顺序顺序发送，可以由业务方在单线程使用同步发送消息的方式来保证
 *  第二点，消息顺序存储，可以由业务方将同一个业务编号的消息发送到一个队列中来实现
 *  还剩下第三点，消息顺序消费，实现消息顺序消费的关键点又是什么呢？

举个例子，假设业务方针对某个订单发送了N个顺序消息，这N个消息都发送到了mq服务端的一个队列中，假设消费者集群中有3个消费者，每个消费者中又是开了N个线程多线程消费

第一种情形，假设3个消费者同时拉取一个队列的消息进行消费，结果会怎么样？N个消息可能会分配在3个消费者中进行消费，多机并行的情况下，消费能力的不同，无法保证这N个消息被顺序消费，所以得保证一个消费队列同一个时刻只能被一个消费者消费

假设又已经保证了一个队列同一个时刻只能被一个消费者消费，那就能保证顺序消费了？同一个消费者多线程进行消费，同样会使得的N个消费被分配到N个线程中，一样无法保证消息顺序消费，所以还得保证一个队列同一个时刻只能被一个消费者中一个线程消费

下面顺序消息的源码分析中就针对这两点来进行分析，即

1.  如何保证一个队列只被一个消费者消费
2.  如何保证一个消费者中只有一个线程能进行消费

**（）一）锁定MessageQueue（添加队列锁，锁定队列）**

先看第一个问题，如何保证一个队列只被一个消费者消费。

消费队列存在于broker端，如果想保证一个队列被一个消费者消费，那么消费者在进行消息拉取消费时就必须想mq服务器申请队列锁，消费者申请队列锁的代码存在于RebalanceService消息队列负载的实现代码中

先明确一点，同一个消费组中的消费者共同承担topic下所有消费者队列的消费，因此每个消费者需要定时重新负载并分配其对应的消费队列，具体为消费者分配消费队列的代码实现在RebalanceImpl\#rebalanceByTopic中，本文不多讲

客户端实现  
消费者重新负载，并且分配完消费队列后，需要向mq服务器发起消息拉取请求，代码实现在RebalanceImpl\#updateProcessQueueTableInRebalance中，针对顺序消息的消息拉取，mq做了如下判断

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center 1][]

核心思想就是，消费客户端先向broker端发起对messageQueue的加锁请求，只有加锁成功时才创建pullRequest进行消息拉取，下面看下lock加锁请求方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center 2][]

代码实现逻辑比较清晰，就是调用lockBatchMQ方法发送了一个加锁请求，那么broker端收到加锁请求后的处理逻辑又是怎么样？

broker端实现  
broker端收到加锁请求的处理逻辑在RebalanceLockManager\#tryLockBatch方法中，RebalanceLockManager中关键属性如下

//默认锁过期时间 60秒
        private final static long REBALANCE_LOCK_MAX_LIVE_TIME = Long.parseLong(System.getProperty(
            "rocketmq.broker.rebalance.lockMaxLiveTime", "60000"));
     //重入锁
        private final Lock lock = new ReentrantLock();
     //key为消费者组名称，value是一个key为MessageQueue，value为LockEntry的map
        private final ConcurrentMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>> mqLockTable =
            new ConcurrentHashMap<String, ConcurrentHashMap<MessageQueue, LockEntry>>(1024);

LockEntry对象中关键属性如下

//消费者id
    private String clientId;
    //最后加锁时间
    private volatile long lastUpdateTimestamp = System.currentTimeMillis();

isLocked方法如下

public boolean isLocked(final String clientId) {
                boolean eq = this.clientId.equals(clientId);
                return eq && !this.isExpired();
            }
    
            public boolean isExpired() {
                boolean expired =
                    (System.currentTimeMillis() - this.lastUpdateTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;
    
                return expired;
            }

对messageQueue进行加锁的关键逻辑如下：

如果messageQueue对应的lockEntry为空，标志队列未加锁，返回加锁成功；

如果lockEntry对应clientId为自己并且没过期，标志同一个客户端重复加锁，返回加锁成功(可重入)；

如果锁已经过期，返回加锁成功；

总而言之，broker端通过对ConcurrentMap<String/\* group \*/, ConcurrentHashMap<MessageQueue, LockEntry>> mqLockTable的维护来达到messageQueue加锁的目的，使得同一时刻，一个messageQueue只能被一个消费者消费

**（二）synchronized申请线程独占锁（添加线程锁）**

假设消费者对messageQueue的加锁已经成功，那么就进入到了第二个步骤，创建pullRequest进行消息拉取，消息拉取部分的代码实现在PullMessageService中，消息拉取完后，需要提交到ConsumeMessageService中进行消费，顺序消费的实现为ConsumeMessageOrderlyService，提交消息进行消费的方法为ConsumeMessageOrderlyService\#submitConsumeRequest，具体实现如下

![20190530201839905.png_pic_center][]

可以看到，构建了一个ConsumeRequest对象，并提交给了ThreadPoolExecutor来并行消费，看下顺序消费的ConsumeRequest的run方法实现

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center 3][]

里面先从messageQueueLock中获取了messageQueue对应的一个锁对象，看下messageQueueLock的实现

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center 4][]

其中维护了一个ConcurrentMap<MessageQueue, Object> mqLockTable，使得一个messageQueue对应一个锁对象object

获取到锁对象后，使用synchronized尝试申请线程级独占锁

1.  如果加锁成功，同一时刻只有一个线程进行消息消费；
2.  如果加锁失败，会延迟100ms重新尝试向broker端申请锁定messageQueue，锁定成功后重新提交消费请求；

至此，第三个关键点的解决思路也清晰了，基本上就两个步骤

1.  创建消息拉取任务时，消息客户端向broker端申请锁定MessageQueue，使得一个MessageQueue同一个时刻只能被一个消费客户端消费
2.  消息消费时，多线程针对同一个消息队列的消费先尝试使用synchronized申请独占锁，加锁成功才能进行消费，使得一个MessageQueue同一个时刻只能被一个消费客户端中一个线程消费

## ##

### 顺序消息重试机制 ###

在使用顺序消息时，一定要注意其异常情况的出现，对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 版会自动不断地进行消息重试（每次间隔时间为 1 秒），重试最大值是Integer.MAX\_VALUE.这时，应用会出现消息消费被阻塞的情况。因此，建议您使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生

重要的事再强调一次：在使用顺序消息时，一定要注意其异常情况的出现！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center]: /images/20230528/163b0a234313415e83d65965b4568171.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20230528/e9f813df72c1477db39ba32ae162f7b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20230528/e8df4a2197114e1fbe045e7be82fc30c.png
[20190530201839905.png_pic_center]: /images/20230528/157508e7200942eda560a5b42dac0355.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20230528/e75f3acb543240e1ade87929bc8e70c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvc2Fvcw_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20230528/a5c5f4cef9574888b1ec979eaceef9d7.png