RocketMQ发送顺序消息

RocketMQ发送顺序消息
如何实现rocketMQ的顺序消息呢？默认情况下生产者发送消息到broker，broker中一个Topic默认有四个队列，消息会被随机分配到各个队列上，消费者消费消息也需要从各个队列上取，并且如果消费者是多个线程的话，更不能保证消息的顺序消费。
从生产者、消费者的角度来看，生产者发送消息时把消息发送到同一个队列，消息者消费消息时只用一个线程消费，队列又是先进先出的，这样就满足的消息的顺序消费。
总结下：

1.生产者发送消息时发送到同一个队列
2.消费者使用单线程消费
3.队列先进先出

模拟一个订单的发送流程，创建三个订单，发送的消息分别是
订单号111 消息流程 创建-》发货-》收货 -》完成
订单号112 消息流程 创建-》完成
订单号113 消息流程 创建-》收货-》完成

代码开发思路：
1.生产者发送消息时通过订单号选择队列，这样一个订单的消息都会发送到同一个队列
2.消费者订阅对应的主题消息，单线程消费

接下来上代码
1.订单实体

package base.entity;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by Administrator on 2021/1/21.
 */
public class Order {


    private int orderId;
    private String orderDec;

    public int getOrderId() {
        return orderId;
    }

    public void setOrderId(int orderId) {
        this.orderId = orderId;
    }

    public String getOrderDec() {
        return orderDec;
    }

    public void setOrderDec(String orderDec) {
        this.orderDec = orderDec;
    }


    public static List<Order> getListOrder(){
        List<Order> list = new ArrayList<>();


        //111  创建-->发货-->收货-->完成
        //112  创建-->完成
        //113  创建-->收货-->完成

        //模拟3个订单以及每个订单发送消息的顺序
        //通过list集合先保证每一个订单的消息顺序


        Order order = new Order();
        order.setOrderId(111);
        order.setOrderDec("创建");
        list.add(order);

        Order orderOne = new Order();
        orderOne.setOrderId(111);
        orderOne.setOrderDec("发货");
        list.add(orderOne);

        Order orderTwo = new Order();
        orderTwo.setOrderId(111);
        orderTwo.setOrderDec("收货");
        list.add(orderTwo);

        Order orderThree = new Order();
        orderThree.setOrderId(111);
        orderThree.setOrderDec("完成");
        list.add(orderThree);


        Order orderFour = new Order();
        orderFour.setOrderId(112);
        orderFour.setOrderDec("创建");
        list.add(orderFour);

        Order orderFive = new Order();
        orderFive.setOrderId(112);
        orderFive.setOrderDec("完成");
        list.add(orderFive);


        Order orderSix = new Order();
        orderSix.setOrderId(113);
        orderSix.setOrderDec("创建");
        list.add(orderSix);

        Order orderSeven = new Order();
        orderSeven.setOrderId(113);
        orderSeven.setOrderDec("收货");
        list.add(orderSeven);

        Order orderEight = new Order();
        orderEight.setOrderId(113);
        orderEight.setOrderDec("完成");
        list.add(orderEight);
        return list;
    }
}

2.生产者发送消息

package base.producer;

import base.entity.Order;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;

public class SyncProducer {


    public static void main(String args[]) throws Exception{


        //创建生产者，并指定生产者组
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //指定nameserver
        producer.setNamesrvAddr("198.1.245.202:9876;198.1.245.204:9876");
        //启动生产者
        producer.start();
        //创建发送消息

        List<Order> list = Order.getListOrder();
        for(Order order:list){
            //topic:OrderTopi 已事先在broker-b上生产
            Message message = new Message("OrderTopic","order",order.getOrderDec().getBytes());
           SendResult sendResult = producer.send(message, new MessageQueueSelector() {
                /**
                 * 发送同一订单(订单号相同)的消息，选择同一队列
                 * 同一订单的消息在list里面是顺序的，顺序发送到同一队列
                 *
                 * @param list  该topic下的消息队列
                 * @param message 消息对象
                 * @param o 订单id,send方法中的第三个参数值会传到这个参数
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    System.out.println("该topic下的消息队列有" + list.size() + "个");
                    int queueNum = list.size();
                    //队列索引
                    int index = ((int)o)%queueNum;
                    return list.get(index);
                }
            },order.getOrderId());

           System.out.println("消息发送结果:" + sendResult);

        }
        //关闭生产者
        producer.shutdown();
    }
}

3.消费者消费消息

package base.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * Created by Administrator on 2021/1/20.
 */
public class Consumer {


    public static void main(String args[]) throws Exception{


        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("198.1.245.202:9876;198.1.245.204:9876");
        consumer.subscribe("OrderTopic","*");
        //注册消息监听器
        consumer.registerMessageListener(new MessageListenerOrderly() {

            /**
             * 一个队列的消息只用同一个线程来消费
             * 队列先进先出，又是同一个线程，可以保证消费的顺序
             *
             * 生产者和消费者这种模式结合，就可以达到顺序消息了
             * @param list
             * @param consumeOrderlyContext
             * @return
             */

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {

                for(MessageExt message:list){
                    System.out.println("当前线程" + Thread.currentThread().getName() + ",消息内容:" + new String(message.getBody()) + ",来自于broker" + message.getBrokerName());
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }
}

4.发送结果和消费结果
发送结果

消费结果

从消费结果来看，同一个线程消费的消息是顺序的，顺序消息成功实现。

查看控制台如下(OrderTopic主题在broker-b上)