RocketMQ-顺序消息Demo及实现原理分析_hosaos的博客_rocketmq 顺序消息

link之家

链接快照平台

输入网页链接，自动生成快照
标签化管理网页链接

全局顺序消费

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

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

这就需要设置topic下读写队列数量为1

为什么要设置读写队列数量为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会有体现

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

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的消息被消费，不是正确的顺序消费，即普通的并行消息消费，无法保证消息消费的顺序性
  
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个线程中，一样无法保证消息顺序消费，所以还得保证一个队列同一个时刻只能被一个消费者中一个线程消费 
下面顺序消息的源码分析中就针对这两点来进行分析，即 
如何保证一个队列只被一个消费者消费
如何保证一个消费者中只有一个线程能进行消费 
锁定MessageQueue
 
先看第一个问题，如何保证一个队列只被一个消费者消费。 
消费队列存在于broker端，如果想保证一个队列被一个消费者消费，那么消费者在进行消息拉取消费时就必须想mq服务器申请队列锁，消费者申请队列锁的代码存在于RebalanceService消息队列负载的实现代码中 
先明确一点，同一个消费组中的消费者共同承担topic下所有消费者队列的消费，因此每个消费者需要定时重新负载并分配其对应的消费队列，具体为消费者分配消费队列的代码实现在RebalanceImpl#rebalanceByTopic中，本文不多讲 
客户端实现
 
消费者重新负载，并且分配完消费队列后，需要向mq服务器发起消息拉取请求，代码实现在RebalanceImpl#updateProcessQueueTableInRebalance中，针对顺序消息的消息拉取，mq做了如下判断 

 核心思想就是，消费客户端先向broker端发起对messageQueue的加锁请求，只有加锁成功时才创建pullRequest进行消息拉取，下面看下lock加锁请求方法
 
 代码实现逻辑比较清晰，就是调用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，具体实现如下
 
 可以看到，构建了一个ConsumeRequest对象，并提交给了ThreadPoolExecutor来并行消费，看下顺序消费的ConsumeRequest的run方法实现 

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

 其中维护了一个ConcurrentMap<MessageQueue, Object> mqLockTable，使得一个messageQueue对应一个锁对象object 
获取到锁对象后，使用synchronized尝试申请线程级独占锁 
如果加锁成功，同一时刻只有一个线程进行消息消费
如果加锁失败，会延迟100ms重新尝试向broker端申请锁定messageQueue，锁定成功后重新提交消费请求 
至此，第三个关键点的解决思路也清晰了，基本上就两个步骤 
创建消息拉取任务时，消息客户端向broker端申请锁定MessageQueue，使得一个MessageQueue同一个时刻只能被一个消费客户端消费
消息消费时，多线程针对同一个消息队列的消费先尝试使用synchronized申请独占锁，加锁成功才能进行消费，使得一个MessageQueue同一个时刻只能被一个消费客户端中一个线程消费 
顺序消息重试机制
 
在使用顺序消息时，一定要注意其异常情况的出现，对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 版会自动不断地进行消息重试（每次间隔时间为 1 秒），重试最大值是Integer.MAX_VALUE.这时，应用会出现消息消费被阻塞的情况。因此，建议您使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生 
重要的事再强调一次：在使用顺序消息时，一定要注意其异常情况的出现！
                            RocketMQ-顺序消息Demo及实现原理分析
                            文章目录场景分析全局顺序消费局部顺序消费DemoProducer DemoNormal Consumer DemoOrder Consumer Demo源码分析锁定MessageQueue客户端实现broker端实现锁定ProcessQueue场景分析顺序消费是指消息的产生顺序和消费顺序相同假设有个下单场景，每个阶段需要发邮件通知用户订单状态变化。用户付款完成时系统给用户发送订单已付款邮件，...
				目录说明生产端消费端总结
RocketMQ与其它消息队列一样，一个Topic利用多个队列来存储数据，单个队列内的数据是顺序存储的，但队列间的数据无法保证顺序性。RocketMQ目前支持保证某类数据或部分数据的顺序性。
核心思想是：发送消息时，可以通过实现MessageQueueSelector接口，选择消息发送到哪个队列，从而保证某类数据的顺序性。同时，可以在send方法中指定入参，方便MessageQueueSelector接口内部根据入参选择指定的队列。
@Test
public void
				1. 前言
顺序消息是RocketMQ的特性之一，它可以让Consumer消费消息的顺序严格按照消息的发送顺序来进行。例如：一条订单产生的三条消息：订单创建、订单付款、订单完成。消费时要按照这个顺序依次消费才有意义，但是不同的订单之间这些消息是可以并行消费的。
顺序消息可以分为全局有序和分区有序，绝大部分场景下，分区有序就已经能够满足需求了，因此本文会重点分析。
全局有序：某个Topic下所有的消息都是有序的，所有消息按照严格的先进先出的顺序进行生产和消费，要求Topic下只能有一个分区队列，且Cons
				kafka篇二11、讲一讲 kafka 的 ack 的三种机制12、消费者如何不自动提交偏移量，由应用提交？13、消费者故障，出现活锁问题如何解决？14、如何控制消费的位置15、kafka 分布式（不是单机）的情况下，如何保证消息的顺序消费?16、kafka 的高可用机制是什么？17、kafka 如何减少数据丢失18、kafka 如何不消费重复数据？比如扣款，我们不能重复的扣。
11、讲一讲 kafka 的 ack 的三种机制
request.required.acks 有三个值 0 1 -1(all)
在项目中使用MQ消息队列时，某些业务场景可能需要保证消息的顺序消费执行。
比如在订单流程场景中：创建订单、支付订单、订单完成这三个订单状态需要保证顺序执行，不能先支付订单，再创建订单。也不能订单完成，再去支付订单。
在项目中如果使用默认的普通消息，那么就算生产者按照顺序发布消息后，消费端也有可能会不会按照顺序进行消费消息。
RocketMQ
RocketMQ在其官方文档中指出，除了支持无序消息以外，还支持顺序消息。
无序消息：普通消息，事务消息，延迟消息和定时消息
顺序消息：全局顺序消息，
				一、rocketmq顺序消费的原理
1、消息的有序性是指消息的消费顺序能够与消息的发送顺序一致。但是有时候我们从业务需要上面并不需要保证所有消息严格按照消费顺序完全一致。例如，一个订单的下单、付款、出库等操作是不同替换顺序。但是有A订单和B订单，并不需要保证A订单与B订单的顺序。
2、RocketMQ采用了局部顺序一致性的机制，一组消息发送到同一个队列中来保证发送顺序的有序性，然后再由消费者进行。消费的时候通过一个队列只会被一个线程取到  ，第二个线程无法访问这个队列 来保证队列有序性。rocketmq可以
// 有序消费MessageListenerOrderly
consumer.registerMessageListener((MessageListenerOrderly) (msgs, content) -> {
    // msgs中只收集同一个topic，同一个tag，并且key相同的message
    try {
        for (MessageExt msg : msgs) {
            String body = ne
				Java 操作RocketMQ简单Demo,顺序与事务案例，不容错过！
前言：MQ的概述与认识这里就暂时不赘述，以及RocketMQ，想了解这方面的知识点，可以推荐伙伴们看 “ 墨家巨子@俏如来 ”大佬写的文章
注：- - 启动Broker时，需要指定可以支持自动创建主题（start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true），如果出现闪退的情况，可以把 MQ文件夹\bin’下，Bean目录下的runbroker.cmd 中JVM占用
				1，单机模式的配置：进入目录\rocketmq-all-4.7.0-bin-release\conf\2m-noslave，修改broker-a.properties目录如下：
# 集群名称
brokerClusterName=DefaultCluster
# broker名字,注意此处不同的配置文件填写的不一样
brokerName=broker-a
# 0 表示Master,>0 表示Slave
brokerId=0
# nameServer地址,分号分割
namesrvAddr=127.0.