Java小案例-RocketMQ的11种消息类型，你知道几种？（顺序消息）

前言

上一节给大家讲了Rocket的普通消息和批量消息，这一节和大家聊一下顺序消息，关于普通消息和批量消息大家可以点下面这个链接直接看

RocketMQ的普通消息和批量消息

顺序消息

首先讲一下概念，所谓的顺序消息就是指消息发送顺序与消费顺序一致，包括全局顺序消息和局部顺序消息。全局顺序消息可以使用一个Topic对应1个队列来实现，但是吞吐量特别低。局部顺序消息是指在一个队列上消息有序的发送和消费，但是对应异步发送和广播消费模式不支持顺序。

RocketMQ可以保证同一个队列的消息绝对顺序，先进入队列的消息会先被消费者拉取到，但是无法保证一个Topic内消息的绝对顺序，意思就是只能保证队列内消息顺序不能保证topic中的队列的顺序

所以要想通过RocketMQ实现顺序消费，需要保证两点

• 生产者将需要保证顺序的消息发送到同一个队列

• 消费者按照顺序消费拉取到的消息

那么，第一个问题，如何消息发送到同一个队列？

前面有提到，RocketMQ发送消息的时候会选择一个队列进行发送

而RocketMQ默认是通过轮询算法来选择队列的，这就无法保证需要顺序消费的消息会存到同一个队列底下

所以，默认情况下是不行了，我们需要自定义队列的选择算法，才能保证消息都在同一个队列中

RocketMQ提供了自定义队列选择的接口MessageQueueSelector

比如我们可以实现这个接口，保证相同订单id的消息都选择同一个队列，在消息发送的时候指定一下就可以了

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        //可以根据业务的id从mqs中选择一个队列
        return null;
    }
}, new Object());

保证消息顺序发送之后，第二个问题，消费者怎么按照顺序消费拉取到的消息？

这个问题RocketMQ已经考虑到了，看看RocketMQ多么地贴心

RocketMQ在消费消息的时候，提供了两种方式：

• 并发消费

• 顺序消费

并发消费，多个线程同时处理同一个队列拉取到的消息

顺序消费，同一时间只有一个线程会处理同一个队列拉取到的消息

至于是并发消费还是顺序消费，需要我们自己去指定

对于顺序处理，只需要实现MessageListenerOrderly接口，处理消息就可以了

public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {

        // 创建一个消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sanyouConsumer");
        // 指定NameServer的地址
        consumer.setNamesrvAddr("192.168.200.143:9876");

        // 订阅sanyouTopic这个topic下的所有的消息
        consumer.subscribe("sanyouTopic", "*");
        // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("消费消息:%s", new String(msg.getBody()) + "\n");
                }

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}

如果想并发消费，换成实现MessageListenerConcurrently即可

这时候就会想

并发消费和顺序消费跟前面提到的集群消费和广播消费有什么区别？

集群消费和广播消费指的是一个消费者组里的每个消费者是去拉取全部队列的消息还是部分队列的消息，也就是选择需要拉取的队列，而并发和顺序消费的意思是，是对已经拉到的同一个队列的消息，是并发处理还是按照消息的顺序去处理

总结

RocketMQ的顺序消息是通过消息队列来保证的，同一个消息队列中的消息是有序的，但是不同的队列之间是没有顺序保证的。当我们发送一条顺序消息时，RocketMQ会根据我们选择的算法来选择一个消息队列，然后将消息发送到这个队列中。

当我们消费顺序消息时，RocketMQ会按照顺序从消息队列中读取消息。RocketMQ提供了两种方式来保证顺序消费：MessageListenerOrderly和ConsumeOrderlyEnable。MessageListenerOrderly是一种高级的方式，可以确保一个线程只消费一个队列中的消息，并且消息按照发送的顺序被消费。ConsumeOrderlyEnable是一种低级的方式，它只能确保一个进程消费一个队列中的消息，但是不能保证消息按照发送的顺序被消费。

联系方式

关于文章中大家有任何疑问可以通过关注公众号《编程乐学》进行留言，同时，公众号还有更多有趣的项目以及关于学习编程的笔记资料大家可以看看，欢迎大家进行留言。