RocketMQ顺序消息和保证顺序原理

向右看齐 2022-10-04 14:58 227阅读 0赞

### RocketMQ顺序消息和保证顺序原理 ###

*   *  简介
     *  RocketMQ顺序消息
     *  实现顺序消息
     *   *  1.消息同步发送
         *  2.重新选择器
         *  3.消费者端
     *  顺序消息原理

## 简介 ##

**顺序消息**作为MQ的一个很重要的功能，一定要只能它是怎么使用的，并且要知道它设计思路如何？下面就根据优秀的RocketMQ源码来看下它是怎么实现消息顺序的。

## RocketMQ顺序消息 ##

首先你要知道[RocketMQ的存储模型][RocketMQ]，根据Broker的存储模型可以知道，一个topic的消息队列是维护在多个Broker的多个queue队列中，所以如果要保证多个Broker的多个queue队列都有序，是不可能的，就算能保证，那消耗的性能和复杂度，也一定是不能接受的。所以**RocketMQ采用分区顺序，不保证全局顺序。**

> 其实`RocketMQ可以做到全局顺序`，就是只用一个Broker一个队列，这样就能保证消息全局顺序，但是考虑性能原因，一般不会有人这么做。  
> 而且分区顺序基本可以满足日常业务需求。

下面就介绍下使用RocketMQ怎么实现顺序消息。

## 实现顺序消息 ##

为了演示RocketMQ的顺序消息，这里以电商场景中的，创建订单、支付、发货、完成等步骤为例。

### 1.消息同步发送 ###

因为默认的消息发送就是同步的，所以这里不用配置。

> 为什么要同步？因为如果异步发送 可能因为网络原因 导致 消息后发先到的情况。

![在这里插入图片描述][20210714171130272.png]

### 2.重新选择器 ###

重写 **MessageQueueSelector**，使同一个订单的消息发送到同一个队列。

public static void main(String[] args) throws UnsupportedEncodingException { 
            try { 
                List<String> orderTasks = Arrays.asList("创建订单", "支付", "发货", "完成");
    
                DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
                producer.setNamesrvAddr("127.0.0.1:9876");
                producer.start();
    
                for (int orderId = 0; orderId < 100; orderId++) { 
                    int finalOrderId = orderId;
                    new Thread(new Runnable() { 
                        @Override
                        public void run() { 
                            for (String task : orderTasks) { 
                                try { 
                                    Message msg = new Message("TopicTest5", "taggg", ("orderId:" + finalOrderId + ", task: " + task).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);
                                        }
                                    }, finalOrderId);
                                } catch (Exception e) { 
                                    e.printStackTrace();
                                }
                            }
                        }
                    }).start();
                }
                //sleep 30秒让消息都发送成功再关闭
                Thread.sleep(1000 * 30);
                producer.shutdown();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }

### 3.消费者端 ###

消费者端使用**MessageListenerOrderly**接收消息，保证消息顺序发送。

public static void main(String[] args) throws MQClientException { 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.setNamesrvAddr("127.0.0.1:9876");
            consumer.subscribe("TopicTest5", "*");
    
            // 使用MessageListenerOrderly消费消息
            consumer.registerMessageListener(new MessageListenerOrderly() { 
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { 
                    context.setAutoCommit(true);
                    for (MessageExt msg : msgs) { 
                        String msgBody = new String(msg.getBody(), "utf-8");
                        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgBody);
                        Thread.sleep(100);
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                    
                }
            });
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }

上面就能顺利实现顺序消息。  
下面就说下它为了实现顺序消息，在代码里是怎么控制的。

## 顺序消息原理 ##

要想消息顺序，只要保证下面的链路顺序：

> Producer -> Queue -> Consumer

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

第一点，消息顺序顺序发送，可以由业务方在单线程使用同步发送消息的方式来保证  
第二点，消息顺序存储，可以由业务方将同一个业务编号的消息发送到一个队列中来实现

还剩下第三点，消息顺序消费，实现消息顺序消费的关键点又是什么呢？

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

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

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

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

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

**`一个队列只被一个消费者消费`**

Consumer启动的时候会进行 **topic 队列的负载分配**，分配后如果是**顺序消息**，会向Broker发送请求申请对 **分配的队列 进行加锁**，Broker会维护一个Map类型**mqLockTable**，保存group锁定了哪些队列，这些队列被哪些client锁定。

具体为消费者分配消费队列的代码实现在`RebalanceImpl#rebalanceByTopic`中，大家可以自己看负载分配的逻辑。  
真正为当前消费者向Broker申请Queue锁的代码是在`RebalanceImpl#updateProcessQueueTableInRebalance`中：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5nMTgyMzgwMzI4OTE_size_16_color_FFFFFF_t_70]  
这个lock里就是向Broker申请Queue锁的代码，申请成功返回true，失败返回false。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5nMTgyMzgwMzI4OTE_size_16_color_FFFFFF_t_70 1]

Broker端收到加锁请求的处理逻辑在`RebalanceLockManager#tryLockBatch`方法中。Broker维护了group-client和queue对应关系的锁，也就是`RebalanceLockManager`的**mqLockTable**，下面是mqLockTable的定义：

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

client-group对queue加锁的逻辑维护在**tryLockBatch**中：  
这里就不贴代码了，感兴趣自己去看，总之就是维护一个`ConcurrentMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>>` 用`isLocked()和isExpired()`来判断是否能加锁成功。

**isLocked**

public boolean isLocked(final String clientId) { 
                boolean eq = this.clientId.equals(clientId);
                return eq && !this.isExpired();
            }

**isExpired**

public boolean isExpired() { 
                boolean expired =
                    (System.currentTimeMillis() - this.lastUpdateTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;
                return expired;
            }

Broker维护`ConcurrentMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>>` 的最终意义就是，让同一个queue只能被同一个group下的一个client消费。这就保证了消费顺序的第一步。

**`一个消费者中只有一个线程能进行消费`**

第一步成功，会创建**PullRequest**拉取消息，拉取消息后，创建一个**ConsumeRequest**提交给线程池消费，进行消费的逻辑是在`ConsumeMessageOrderlyService.ConsumeRequest#run`。进行消费时，会先获取MessageQueue锁对象，并且使用`synchronized`，保证一个queue只有一个线程能进行消费。看下`ConsumeRequest#run`源码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5nMTgyMzgwMzI4OTE_size_16_color_FFFFFF_t_70 2][]

上面使用两把锁已经能保证正常情况下的消息顺序消费。但是在生产环境中经常会发生 `消费者客户端数量变动(比如client的上下线)，这样会导致client端队列重新负载。`

未完待续。。。。。  
最后会对ProccessQueue进行加锁，保证处理中的消息 消费完成后，其他消费者才能消费。这样就算发生队列重新负载，其他消费者也不能消费之前的消息，最大可能保证了重复消费

[RocketMQ]: https://blog.csdn.net/jiang18238032891/article/details/118605487
[20210714171130272.png]: /images/20221004/82cd3a94bb624c9b86ff97e2b9a538ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5nMTgyMzgwMzI4OTE_size_16_color_FFFFFF_t_70]: /images/20221004/aecf511e68ea4b48bb37e55621e47b6f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5nMTgyMzgwMzI4OTE_size_16_color_FFFFFF_t_70 1]: /images/20221004/a725aae7caff4cd4bbb37db6d7c74e0e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5nMTgyMzgwMzI4OTE_size_16_color_FFFFFF_t_70 2]: /images/20221004/db0c46ef889f4c4ab101468d98b95d32.png