RocketMQ——顺序消费（代码）

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

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

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

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

顺序消费简介

1.普通顺序消费

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

2.严格顺序消息

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

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

1.producer

  1. package com.gwd.rocketmq;
  3. import java.io.IOException;
  4. import java.text.SimpleDateFormat;
  5. import java.util.ArrayList;
  6. import java.util.Date;
  7. import java.util.List;
  9. import com.alibaba.rocketmq.client.exception.MQBrokerException;
  10. import com.alibaba.rocketmq.client.exception.MQClientException;
  11. import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
  12. import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
  13. import com.alibaba.rocketmq.client.producer.SendResult;
  14. import com.alibaba.rocketmq.common.message.Message;
  15. import com.alibaba.rocketmq.common.message.MessageQueue;
  16. import com.alibaba.rocketmq.remoting.exception.RemotingException;
  18. /** 
  19. * @FileName Producer.java
  20. * @Description:
  21. * @author gu.weidong
  22. * @version V1.0
  23. * @createtime 2018年7月3日 上午9:59:38 
  24. * 修改历史：
  25. * 时间           作者          版本        描述
  26. *====================================================  
  27. *
  28. */
  29. /**
  30.  * Producer，发送顺序消息
  31.  */
  32. public class Producer {
  34.     public static void main(String[] args) throws IOException {
  35.         try {
  36.             DefaultMQProducer producer = new DefaultMQProducer("sequence_producer");
  38.             producer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
  40.             producer.start();
  42.             String[] tags = new String[] { "TagA", "TagC", "TagD" };
  44.             // 订单列表
  45.             List<OrderDO> orderList =  new Producer().buildOrders();
  47.             Date date = new Date();
  48.             SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  49.             String dateStr = sdf.format(date);
  50.             for (int i = 0; i < 10; i++) {
  51.                 // 加个时间后缀
  52.                 String body = dateStr + " Hello RocketMQ " + orderList.get(i).getOrderId()+orderList.get(i).getDesc();
  53.                 Message msg = new Message("SequenceTopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());
  55.                 SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
  56.                     @Override
  57.                     public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
  58.                         Long id = Long.valueOf((String)arg);
  59.                         long index = id % mqs.size();
  60.                         return mqs.get((int)index);
  61.                     }
  62.                 }, orderList.get(i).getOrderId());//通过订单id来获取对应的messagequeue
  64.                 System.out.println(sendResult + ", body:" + body);
  65.             }
  67.             producer.shutdown();
  69.         } catch (MQClientException e) {
  70.             e.printStackTrace();
  71.         } catch (RemotingException e) {
  72.             e.printStackTrace();
  73.         } catch (MQBrokerException e) {
  74.             e.printStackTrace();
  75.         } catch (InterruptedException e) {
  76.             e.printStackTrace();
  77.         }
  78.         System.in.read();
  79.     }
  81.     /**
  82.      * 生成模拟订单数据 
  83.      */
  84.     private List<OrderDO> buildOrders() {
  85.         List<OrderDO> orderList = new ArrayList<OrderDO>();
  87.         OrderDO OrderDO = new OrderDO();
  88.         OrderDO.setOrderId("15103111039");
  89.         OrderDO.setDesc("创建");
  90.         orderList.add(OrderDO);
  92.         OrderDO = new OrderDO();
  93.         OrderDO.setOrderId("15103111065");
  94.         OrderDO.setDesc("创建");
  95.         orderList.add(OrderDO);
  97.         OrderDO = new OrderDO();
  98.         OrderDO.setOrderId("15103111039");
  99.         OrderDO.setDesc("付款");
  100.         orderList.add(OrderDO);
  102.         OrderDO = new OrderDO();
  103.         OrderDO.setOrderId("15103117235");
  104.         OrderDO.setDesc("创建");
  105.         orderList.add(OrderDO);
  107.         OrderDO = new OrderDO();
  108.         OrderDO.setOrderId("15103111065");
  109.         OrderDO.setDesc("付款");
  110.         orderList.add(OrderDO);
  112.         OrderDO = new OrderDO();
  113.         OrderDO.setOrderId("15103117235");
  114.         OrderDO.setDesc("付款");
  115.         orderList.add(OrderDO);
  117.         OrderDO = new OrderDO();
  118.         OrderDO.setOrderId("15103111065");
  119.         OrderDO.setDesc("完成");
  120.         orderList.add(OrderDO);
  122.         OrderDO = new OrderDO();
  123.         OrderDO.setOrderId("15103111039");
  124.         OrderDO.setDesc("推送");
  125.         orderList.add(OrderDO);
  127.         OrderDO = new OrderDO();
  128.         OrderDO.setOrderId("15103117235");
  129.         OrderDO.setDesc("完成");
  130.         orderList.add(OrderDO);
  132.         OrderDO = new OrderDO();
  133.         OrderDO.setOrderId("15103111039");
  134.         OrderDO.setDesc("完成");
  135.         orderList.add(OrderDO);
  136.         return orderList;
  137.     }
  138. }

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

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

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

2.错误的Consumer

  1. package com.gwd.rocketmq;
  3. import java.util.List;
  4. import java.util.Random;
  5. import java.util.concurrent.TimeUnit;
  7. import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
  8. import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  9. import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  10. import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  11. import com.alibaba.rocketmq.client.exception.MQClientException;
  12. import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
  13. import com.alibaba.rocketmq.common.message.MessageExt;
  15. /** 
  16. * @FileName WrongConsumer.java
  17. * @Description:
  18. * @author gu.weidong
  19. * @version V1.0
  20. * @createtime 2018年7月3日 下午3:13:16 
  21. * 修改历史：
  22. * 时间           作者          版本        描述
  23. *====================================================  
  24. *
  25. */
  26. public class WrongConsumer {
  27.     public static void main(String[] args) throws MQClientException {
  28.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
  29.         consumer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
  30.         /**
  31.          * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
  32.          * 如果非第一次启动，那么按照上次消费的位置继续消费
  33.          */
  34.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  36.         consumer.subscribe("SequenceTopicTest", "TagA || TagC || TagD");
  38.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  40.             Random random = new Random();
  42.             @Override
  43.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
  44.                 System.out.print(Thread.currentThread().getName() + " Receive New Messages: " );
  45.                 for (MessageExt msg: msgs) {
  46.                     System.out.println(msg + ", content:" + new String(msg.getBody()));
  47.                 }
  48.                 try {
  49.                     //模拟业务逻辑处理中...
  50.                     TimeUnit.SECONDS.sleep(random.nextInt(10));
  51.                 } catch (Exception e) {
  52.                     e.printStackTrace();
  53.                 }
  54.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  55.             }
  56.         });
  58.         consumer.start();
  60.         System.out.println("Consumer Started.");
  61.     }
  62. }

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

70

3.正确的Consumer

  1. package com.gwd.rocketmq;
  3. import java.util.List;
  4. import java.util.Random;
  5. import java.util.concurrent.TimeUnit;
  7. import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
  8. import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
  9. import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
  10. import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
  11. import com.alibaba.rocketmq.client.exception.MQClientException;
  12. import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
  13. import com.alibaba.rocketmq.common.message.MessageExt;
  15. /** 
  16. * @FileName Consumer.java
  17. * @Description:
  18. * @author gu.weidong
  19. * @version V1.0
  20. * @createtime 2018年7月3日 上午10:05:26 
  21. * 修改历史：
  22. * 时间           作者          版本        描述
  23. *====================================================  
  24. *
  25. */
  26. /**
  27.  * 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）
  28.  */
  29. public class Consumer {
  31.     public static void main(String[] args) throws MQClientException {
  32.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
  33.         consumer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
  34.         /**
  35.          * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
  36.          * 如果非第一次启动，那么按照上次消费的位置继续消费
  37.          */
  38.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  40.         consumer.subscribe("SequenceTopicTest", "TagA || TagC || TagD");
  42.         consumer.registerMessageListener(new MessageListenerOrderly() {
  44.             Random random = new Random();
  46.             @Override
  47.             public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
  48.                 context.setAutoCommit(true);
  49.                 System.out.print(Thread.currentThread().getName() + " Receive New Messages: " );
  50.                 for (MessageExt msg: msgs) {
  51.                     System.out.println(msg + ", content:" + new String(msg.getBody()));
  52.                 }
  53.                 try {
  54.                     //模拟业务逻辑处理中...
  55.                     TimeUnit.SECONDS.sleep(random.nextInt(10));
  56.                 } catch (Exception e) {
  57.                     e.printStackTrace();
  58.                 }
  59.                 return ConsumeOrderlyStatus.SUCCESS;
  60.             }
  61.         });
  63.         consumer.start();
  64.         System.out.println("Consumer Started.");
  65.     }
  66. }

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

显示结果：

70 1

4.多个消费者

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

70 2

70 3

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

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

发表评论

表情：
评论列表 （有 0 条评论，422人围观）

还没有评论，来说两句吧...

相关阅读

    相关 使用RocketMQ实现消息顺序消费

    消息的顺序消费在很多交易型的业务场景中都会被要求实现，而且，消息队列的顺序消费解决方案在很多互联网公司的面试中经常会被问到。 索尔老师在使用了多个消息队列后发现，虽然每个消息

    港控/mmm°港控/mmm°/ 0 赞/ 128 阅读

    相关 rocketMq 顺序消费

    什么是顺序消费？ 消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了 3 条消息，分别是订单创建、订单付款、订单完成。消费时，要按照这个顺序消费才

    Dear 丶Dear 丶/ 0 赞/ 163 阅读

    相关 RocketMQ 顺序消费机制

    顺序消息是指对于一个指定的 Topic ，消息严格按照先进先出（FIFO）的原则进行消息发布和消费，即先发布的消息先消费，后发布的消息后消费。 顺序消息分为分区顺序消息和全局

    柔光的暖阳◎柔光的暖阳◎/ 0 赞/ 86 阅读

    相关 springboot整合rocketmq实现顺序消费

    消息队列已然成为当下非常火热的中间件，而rocketmq作为阿里开源的中间件产品，历经数次超大并发的考验，已然成为中间件产品的首选。而有时候我们在使用消息队列的时候，往往需要能

    迷南。迷南。/ 0 赞/ 84 阅读