精讲RocketMQ顺序消息原理以及代码

发表时间：2022-03-25来源：网络

编程界的小学生

一、顺序消息解释1、概述2、图解3、再次分析二、解决方案1、方案一2、代码一2.1、生产者2.2、消费者2.3、输出结果3、情况二

一、顺序消息解释

1、概述

RocketMQ的消息是存储到Topic的queue里面的，queue本身是FIFO（First Int First Out）先进先出队列。所以单个queue是可以保证有序性的。但是问题是1个topic有N个queue，作者这么设计的好处也很明显，天然支持集群和负载均衡的特性，将海量数据均匀分配到各个queue上，你发了10条消息到同一个topic上，这10条消息会自动分散在topic下的所有queue中，所以消费的时候不一定是先消费哪个queue，后消费哪个queue，这就导致了无序消费。

2、图解

3、再次分析

一个Producer发送了m1、m2、m3、m4四条消息到topic上，topic有四个队列，由于自带的负载均衡策略，四个队列上分别存储了一条消息。queue1上存储的m1，queue2上存储的m2，queue3上存储的m3，queue4上存储的m4，Consumer消费的时候是多线程消费，所以他无法保证先消费哪个队列或者哪个消息，比如发送的时候顺序是m1，m2，m3，m4，但是消费的时候由于Consumer内部是多线程消费的，所以可能先消费了queue4队列上的m4，然后才是m1，这就导致了无序。

二、解决方案

1、方案一

很简单，问题产生的关键在于多个队列都有消息，我消费的时候又不知道哪个队列的消息是最新的。那么思路就有了，发消息的时候你要想保证有序性的话，就都给我发到一个queue上，然后消费的时候因为只有那一个queue上有消息且queue是FIFO，先进先出，所以正常消费就完了。很完美。而且RocketMQ也给我们提供了这种发消息的时候选择queue的api（Message）。直接上代码。

2、代码一

2.1、生产者

import java.util.List; import org.apache.rocketmq.client.producer.DefaultM; import org.apache.rocketmq.client.producer.Message; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageQueue; /** * 消息发送者 */ public class Producer5 { public static void main(String[] args)throws Exception { DefaultM producer = new DefaultM("my-order-producer"); producer.setNamesrvAddr("124.57.180.156:9876"); producer.start(); for (int i = 0; i // 手动选择一个queue @Override public MessageQueue select( // 当前topic 里面包含的所有queue List mqs, // 具体要发的那条消息 Message msg, // 对应到 send（）里的 args，也就是2000前面的那个0 // 实际业务中可以把0换成实际业务系统的主键，比如订单号啥的，然后这里做hash进行选择queue等。能做的事情很多，我这里做演示就用第一个queue，所以不用arg。 Object arg) { // 向固定的一个queue里写消息，比如这里就是向第一个queue里写消息 MessageQueue queue = mqs.get(0); // 选好的queue return queue; } }, // 自定义参数：0 // 2000代表2000毫秒超时时间 0, 2000); } } }

2.2、消费者

import org.apache.rocketmq.client.consumer.DefaultM; import org.apache.rocketmq.client.consumer.listener.*; import org.apache.rocketmq.common.message.MessageExt; import java.util.List; /** * Description: * * @author TongWei.Chen 2020-06-22 11:17:47 */ public class ConsumerOrder { public static void main(String[] args) throws Exception { DefaultM consumer = new DefaultM("my-consumer"); consumer.setNamesrvAddr("124.57.180.156:9876"); consumer.subscribe("orderTopic", "*"); consumer.registerMessageListener(new MessageListenerOrderly() { @Override public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) { for (MessageExt msg : msgs) { System.out.println(new String(msg.getBody()) + " Thread:" + Thread.currentThread().getName() + " :" + msg.get()); } return ConsumeOrderlyStatus.SUCCESS; } }); consumer.start(); System.out.println("Consumer start..."); } }

2.3、输出结果

Consumer start... hello!0 Thread:ConsumeMessageThread_1 :0 hello!1 Thread:ConsumeMessageThread_1 :0 hello!2 Thread:ConsumeMessageThread_1 :0 hello!3 Thread:ConsumeMessageThread_1 :0 hello!4 Thread:ConsumeMessageThread_1 :0

很完美，有序输出！

3、情况二

比如你新需求：把未支付的订单都放到queue1里，已支付的订单都放到queue2里，支付异常的订单都放到queue3里，然后你消费的时候要保证每个queue是有序的，不能消费queue1一条直接跑到queue2去了，要逐个queue去消费。

这时候思路是发消息的时候利用自定义参数arg，消息体里肯定包含支付状态，判断是未支付的则选择queue1，以此类推。这样就保证了每个queue里只包含同等状态的消息。那么消费者目前是多线程消费的，肯定乱序。三个queue随机消费。解决方案更简单，直接将消费端的线程数改为1个，这样队列是FIFO，他就逐个消费了。RocketMQ也为我们提供了这样的api，如下两句:

// 最大线程数1 consumer.setConsumeThreadMax(1); // 最小线程数 consumer.setConsumeThreadMin(1);

上一篇：8岁小学生开直播教编程，引发百万点击！10后“程序员”的养成~

下一篇：小孩子学编程先学什么