RocketMQ——顺序消费（代码）

柔情只为你懂 2021-09-27 01:25 366阅读 0赞

关于rocketmq顺序消费的理解和图示可以查看该博文：[RocketMQ——顺序消费和重复消费][RocketMQ]

本博客主要是以代码示例来了解顺序消费的相关内容,建议在此之前先了解下顺序消费的原理。

注：RocketMQ可以严格的保证消息有序，但这个顺序，不是全局顺序，只是分区（queue）顺序，如果想要全局顺序，那么需要保证只有一个分区。

### 顺序消费简介 ###

#### 1.普通顺序消费 ####

顺序消费的一种，正常情况下可以保证完全的顺序消息，但是一旦发生通信异常，Broker重启，由于队列总数法还是能变化，哈希取模后定位的队列会变化，产生短暂的消息顺序不一致。

#### 2.严格顺序消息 ####

顺序消息的一种，无论正常异常情况都能保证顺序，但是牺牲了分布式failover特性，即broker集群中要有一台机器不可用，则整个集群都不可用，服务可用性大大降低。如果服务器部署为同步双写模式，此缺陷可通过备机自动切换为主避免，不过仍然会存在几分钟的服务不可用。

目前已知的应用只有数据库binlog同步强依赖严格顺序消息，其他应用绝大部分都可以容忍短暂乱序，推荐使用普通的顺序消息。

### 1.producer ###

package com.gwd.rocketmq;
    
    import java.io.IOException;
    import java.text.SimpleDateFormat;
    import java.util.ArrayList;
    import java.util.Date;
    import java.util.List;
    
    import com.alibaba.rocketmq.client.exception.MQBrokerException;
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
    import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
    import com.alibaba.rocketmq.client.producer.SendResult;
    import com.alibaba.rocketmq.common.message.Message;
    import com.alibaba.rocketmq.common.message.MessageQueue;
    import com.alibaba.rocketmq.remoting.exception.RemotingException;
    
    /** 
    * @FileName Producer.java
    * @Description:
    * @author gu.weidong
    * @version V1.0
    * @createtime 2018年7月3日 上午9:59:38 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    /**
     * Producer，发送顺序消息
     */
    public class Producer {
    	
        public static void main(String[] args) throws IOException {
            try {
                DefaultMQProducer producer = new DefaultMQProducer("sequence_producer");
     
                producer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
     
                producer.start();
     
                String[] tags = new String[] { "TagA", "TagC", "TagD" };
                
                // 订单列表
                List<OrderDO> orderList =  new Producer().buildOrders();
                
                Date date = new Date();
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String dateStr = sdf.format(date);
                for (int i = 0; i < 10; i++) {
                    // 加个时间后缀
                    String body = dateStr + " Hello RocketMQ " + orderList.get(i).getOrderId()+orderList.get(i).getDesc();
                    Message msg = new Message("SequenceTopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());
     
                    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            Long id = Long.valueOf((String)arg);
                            long index = id % mqs.size();
                            return mqs.get((int)index);
                        }
                    }, orderList.get(i).getOrderId());//通过订单id来获取对应的messagequeue
     
                    System.out.println(sendResult + ", body:" + body);
                }
                
                producer.shutdown();
     
            } catch (MQClientException e) {
                e.printStackTrace();
            } catch (RemotingException e) {
                e.printStackTrace();
            } catch (MQBrokerException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.in.read();
        }
        
        /**
         * 生成模拟订单数据 
         */
        private List<OrderDO> buildOrders() {
        	List<OrderDO> orderList = new ArrayList<OrderDO>();
     
        	OrderDO OrderDO = new OrderDO();
            OrderDO.setOrderId("15103111039");
        	OrderDO.setDesc("创建");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103111065");
        	OrderDO.setDesc("创建");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103111039");
        	OrderDO.setDesc("付款");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103117235");
        	OrderDO.setDesc("创建");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103111065");
        	OrderDO.setDesc("付款");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103117235");
        	OrderDO.setDesc("付款");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103111065");
        	OrderDO.setDesc("完成");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103111039");
        	OrderDO.setDesc("推送");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103117235");
        	OrderDO.setDesc("完成");
        	orderList.add(OrderDO);
        	
        	OrderDO = new OrderDO();
        	OrderDO.setOrderId("15103111039");
        	OrderDO.setDesc("完成");
        	orderList.add(OrderDO);
        	return orderList;
        }
    }

此处需要注意，producer.send(msg, new MessageQueueSelector()),如果需要全局有序，只需要使new MessageQueueSelector().select(List<MessageQueue> mqs, Message msg, Object arg)方法返回值唯一且不变，例如：

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            Long id = Long.valueOf((String)arg);
                            long index = id % mqs.size();
                            return mqs.get((int)index);
                        }
                    }, orderList.get(0).getOrderId());//通过订单id来获取对应的messagequeue

这边获取到的queue永远都是唯一的且确定的（此处只是举个简单的例子，orderList.get(i).getOrderId()改为0亦可）

### 2.错误的Consumer ###

package com.gwd.rocketmq;
    
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
    import com.alibaba.rocketmq.common.message.MessageExt;
    
    /** 
    * @FileName WrongConsumer.java
    * @Description:
    * @author gu.weidong
    * @version V1.0
    * @createtime 2018年7月3日 下午3:13:16 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    public class WrongConsumer {
    	public static void main(String[] args) throws MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
            consumer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
            /**
             * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
             * 如果非第一次启动，那么按照上次消费的位置继续消费
             */
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
     
            consumer.subscribe("SequenceTopicTest", "TagA || TagC || TagD");
     
            consumer.registerMessageListener(new MessageListenerConcurrently() {
     
                Random random = new Random();
     
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    System.out.print(Thread.currentThread().getName() + " Receive New Messages: " );
                    for (MessageExt msg: msgs) {
                        System.out.println(msg + ", content:" + new String(msg.getBody()));
                    }
                    try {
                        //模拟业务逻辑处理中...
                        TimeUnit.SECONDS.sleep(random.nextInt(10));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
     
            consumer.start();
     
            System.out.println("Consumer Started.");
        }
    }

注意：要想要有顺序，那么这边吃监听器就不能是MessageListenerConcurrently了，其显示效果如下：

![70][]

### 3.正确的Consumer ###

package com.gwd.rocketmq;
    
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
    import com.alibaba.rocketmq.common.message.MessageExt;
    
    /** 
    * @FileName Consumer.java
    * @Description:
    * @author gu.weidong
    * @version V1.0
    * @createtime 2018年7月3日 上午10:05:26 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    /**
     * 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）
     */
    public class Consumer {
     
        public static void main(String[] args) throws MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
            consumer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
            /**
             * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
             * 如果非第一次启动，那么按照上次消费的位置继续消费
             */
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
     
            consumer.subscribe("SequenceTopicTest", "TagA || TagC || TagD");
     
            consumer.registerMessageListener(new MessageListenerOrderly() {
     
                Random random = new Random();
     
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    context.setAutoCommit(true);
                    System.out.print(Thread.currentThread().getName() + " Receive New Messages: " );
                    for (MessageExt msg: msgs) {
                        System.out.println(msg + ", content:" + new String(msg.getBody()));
                    }
                    try {
                        //模拟业务逻辑处理中...
                        TimeUnit.SECONDS.sleep(random.nextInt(10));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
     
            consumer.start();
            System.out.println("Consumer Started.");
        }
    }

这边的Consumer和上面的最明显的区别在于对应的监听器是MessageListenerOrderly，MessageListenerOrderly是能够保证顺序消费的。

显示结果：

![70 1][]

### 4.多个消费者 ###

那如果有多个消费者呢？因为消息发送时被分配到多个队列，这些队列又会被分别发送给消费者唯一消费，现在启动两个消费者，其消费情况如下图：

![70 2][]

![70 3][]

结论：多个消费者时，各个消费者的消息依旧是顺序消费，且不会重复消费

原文转自：https://blog.csdn.net/earthhour/article/details/78323026 ，在此基础上部分代码略作修改

[RocketMQ]: https://blog.csdn.net/gwd1154978352/article/details/80691916
[70]: /images/20210923/1f86ae9441db4a9088c1502bf9a9a17c.png
[70 1]: /images/20210923/ca07c41df5794dcf8b62fc3f22e679e4.png
[70 2]: /images/20210923/bad136a6e6c840979cbc0084ab1290f6.png
[70 3]: /images/20210923/9c6263ed3eb24c2096dbf2616b5735ab.png