RocketMQ顺序消息是怎么实现的？

顺序消费又分两种，全局顺序消费和局部顺序消费 ### 全局顺序消费 什么是全局顺序消费？所有发到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里局部顺序消息又是如何怎么实现的呢？ 要保证消息的顺序消费，有三个关键点 1. 消息顺序发送 2. 消息顺序存储 3. 消息顺序消费 第一点，消息顺序发送，多线程发送的消息无法保证有序性，因此，需要业务方在发送时，针对同一个业务编号(如同一笔订单)的消息需要保证在一个线程内顺序发送，在上一个消息发送成功后，在进行下一个消息的发送。对应到mq中，消息发送方法就得使用同步发送，异步发送无法保证顺序性 第二点，消息顺序存储，mq的topic下会存在多个queue，要保证消息的顺序存储，同一个业务编号的消息需要被发送到一个queue中。对应到mq中，需要使用MessageQueueSelector来选择要发送的queue，即对业务编号进行hash，然后根据队列数量对hash值取余，将消息发送到一个queue中 第三点，消息顺序消费，要保证消息顺序消费，同一个queue就只能被一个消费者所消费，因此对broker中消费队列加锁是无法避免的。同一时刻，一个消费队列只能被一个消费者消费，消费者内部，也只能有一个消费线程来消费该队列。即，同一时刻，一个消费队列只能被一个消费者中的一个线程消费 上面第一、第二点中提到，要保证消息顺序发送和消息顺序存储需要使用mq的同步发送和MessageQueueSelector来保证，具体Demo会有体现 至于第三点中的加锁操作会结合源码来具体分析 ### 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的消息被消费，不是正确的顺序消费，即普通的并行消息消费，无法保证消息消费的顺序性