RocketMQ保证顺序消费demo

上一篇 <<<SpringBoot整合RocketMQ示例
下一篇 >>>RocketMQ如何动态扩容和缩容

1.解决顺序消费的核心思路

a、相同的业务逻辑一定要放到同一个队列中
b、每个队列必须要对应同一个消费者消费
c、RocketMQ每个队列1：1均等匹配消费者

2.代码示例

2.1 jar包引入

 <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.5.1</version>
    </dependency>

2.2 相同行为的数据投放到相同的队列中—默认是使用轮询策略的

Long orderId = System.currentTimeMillis();
String insertSql = getSqlMsg("insert", orderId);
String updateSql = getSqlMsg("update", orderId);
String deleteSql = getSqlMsg("delete", orderId);
Message insertMsg = new Message(T.TOPIC, insertSql.getBytes());
Message updateMsg = new Message(T.TOPIC, updateSql.getBytes());
Message deleteMsg = new Message(T.TOPIC, deleteSql.getBytes());
producer = rocketMQTemplate.getProducer();
producer.send(insertMsg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        // 所有的消息使用同一个订单号，存放到同一个队列中。
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);
producer.send(updateMsg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        // 所有的消息使用同一个订单号，存放到同一个队列中
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);
producer.send(deleteMsg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        // 所有的消息使用同一个订单号，存放到同一个队列中
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);

在Springboot中可以这样写:
@Service
@RocketMQMessageListener(topic = "myTest", consumerGroup = "myTest",consumeMode = ConsumeMode.ORDERLY,consumeThreadMax = 1)
public class OrdeConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String o) {
        System.out.println("o:" + JSON.toJSONString(o));
    }
}

ConsumeMode 支持并发和顺序两种形式
consumeThreadMax 配置为1，则只能使用同一个线程消费
默认值是20，可以从DefaultMqPushConsumer类中查找到
建议配置成和队列数相同，不过打印出来的日志会显示线程不是同一个，这是线程复用机制，底层也是保证顺序消费的

2.3 消费者同一线程读取同一个队列

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
consumer.setNamesrvAddr(T.NAMESERVER);
consumer.setConsumerGroup(T.consumerGroup);
consumer.subscribe(T.TOPIC, "");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        //设置为true，则自动应答,不会重复获取
        //设置为false，则不会自动应答，程序会重复读取
        context.setAutoCommit(true);
        MessageQueue queue = context.getMessageQueue();
        for (MessageExt ext : msgs) {
            try {
                Thread.sleep((new Random()).nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+" 消费者获取消息 brokerName: "+queue.getBrokerName()+" topic:"+queue.getTopic()+" queueId:"+queue.getQueueId()+",队列id：" + ext.getQueueId() + ", 消息" +
                    new String(ext.getBody()) + "msgid:" + ext.getMsgId());
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});
consumer.start();

2.4 执行结果

ConsumeMessageThread_1 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:1,队列id：1, 消息{"orderId":1577796057489,"type":"insert"}msgid:0A2581022BE518B4AAC29EEF11E70000
ConsumeMessageThread_1 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:1,队列id：1, 消息{"orderId":1577796057489,"type":"update"}msgid:0A2581022BE518B4AAC29EEF11F40001
ConsumeMessageThread_1 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:1,队列id：1, 消息{"orderId":1577796057489,"type":"delete"}msgid:0A2581022BE518B4AAC29EEF11F70002
ConsumeMessageThread_2 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:3,队列id：3, 消息{"orderId":1577796132835,"type":"insert"}msgid:0A2581022BE818B4AAC29EF0382B0000
ConsumeMessageThread_2 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:3,队列id：3, 消息{"orderId":1577796132835,"type":"update"}msgid:0A2581022BE818B4AAC29EF038360001
ConsumeMessageThread_2 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:3,队列id：3, 消息{"orderId":1577796132835,"type":"delete"}msgid:0A2581022BE818B4AAC29EF038390002
ConsumeMessageThread_3 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:0,队列id：0, 消息{"orderId":1577796180976,"type":"insert"}msgid:0A2581022BEF18B4AAC29EF0F43E0000
ConsumeMessageThread_3 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:0,队列id：0, 消息{"orderId":1577796180976,"type":"update"}msgid:0A2581022BEF18B4AAC29EF0F44A0001
ConsumeMessageThread_3 消费者获取消息 brokerName: broker-a topic:tp1231001 queueId:0,队列id：0, 消息{"orderId":1577796180976,"type":"delete"}msgid:0A2581022BEF18B4AAC29EF0F44D0002

推荐阅读：
<<<消息中间件的核心思想
<<<消息中间件常见问题汇总
<<<基于Netty简单手写消息中间件思路
<<<消息队列常用名词与中间件对比
<<<Rabbitmq基础知识
<<<Rabbitmq示例之点对点简单队列
<<<Rabbitmq示例之工作(公平)队列
<<<Rabbitmq示例之发布订阅模式
<<<Rabbitmq示例之路由模式Routing
<<<Rabbitmq示例之通配符模式Topics
<<<Rabbitmq示例之RPC模式
<<<Rabbitmq队列模式总结
<<<Rabbitmq如何保证消息不丢失
<<<Springboot利用AmqpTemplate整合Rabbitmq
<<<Rabbitmq如何保证幂等性
<<<Rabbitmq的重试策略
<<<Rabbitmq通过死信队列实现过期监听
<<<Rabbitmq解决分布式事务思路
<<<Rabbitmq解决分布式事务demo
<<<Rabbitmq环境安装
<<<Kafka中的专业术语都有哪些
<<<Kafka的设计原理介绍
<<<Kafka集群如何实现相互感知
<<<Kafka如何实现分区及指定分区消费
<<<Kafka如何保证消息顺序消费
<<<Kafka如何保证高吞吐量
<<<Kafka集群环境搭建
<<<RocketMQ架构原理
<<<RocketMQ、RabbitMQ和Kafka的对比
<<<SpringBoot整合RocketMQ示例
<<<RocketMQ如何动态扩容和缩容
<<<RocketMQ如何解决分布式事务
<<<RocketMQ单机版本安装
<<<RocketMQ集群环境程序启用相关知识点
<<<RocketMQ单机做主备实操
<<<RocketMQ所有配置说明