rocketmq实战之消息的有序发送与有序消费

古城微笑少年丶 2023-05-31 03:28 33阅读 0赞

rocketmq支持有序的发送消息，有序的消费消息，这里便来看一下如何实现？

> 实战方面均以`RocketMQTemplate`形式展现，集成方案详见[《springboot中rocketmq的集成与使用》][springboot_rocketmq]

消息的有序发送方面，我们可以直接使用`syncSendOrderly`（同步有序发送）和`asyncSendOrderly`（异步有序发送）两种类型的方法进行发送消息，他们的区别就如命名一样是同步和异步的区别。

下面给出同步有序发送的样例代码：

public class OrderProducer { 
    
        private static final Integer NUM = 10;
    
        @Resource
        private RocketMQTemplate rocketMQTemplate;
    
        public void sendSyncOrderly() { 
            String message = "orderly message: ";
            for (int i = 0; i < NUM; i++) { 
                // 模拟有序发送消息
                rocketMQTemplate.syncSendOrderly("topic-orderly", message + i, "select_queue_key");
            }
        }
    }

有序消息的发送比普通消息的发送多了一类参数“select\_queue\_key”，在`rocketmq-spring`中叫做`hashKey`，该参数的左右即是在发送消息的时候，固定发送到一个队列（默认情况下rocketmq中的topic有4个队列）以保证顺序。

有序消费的样例代码如下：

@Slf4j
    @RocketMQMessageListener(topic = "topic-orderly",
            consumerGroup = "orderly-consumer-group",consumeMode = ConsumeMode.ORDERLY)
    public class OrderConsumer implements RocketMQListener<String> { 
    
        @Override
        public void onMessage(String message) { 
            log.info("========{}=======", message);
        }
    }

这里主要是设置消费模式`consumeMode = ConsumeMode.ORDERLY`，默认情况下是并发消费模式（`ConsumeMode.CONCURRENTLY`）。

2019-10-22 21:59:01.654  INFO 1400 --- [           main] c.l.s.r.SpringbootAcRocketmqApplication  : Started SpringbootAcRocketmqApplication in 2.611 seconds (JVM running for 3.824)
    2019-10-22 21:59:04.474  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 0=======
    2019-10-22 21:59:04.475  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD27200000 cost: 0 ms
    2019-10-22 21:59:04.475  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 1=======
    2019-10-22 21:59:04.475  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD27280002 cost: 0 ms
    2019-10-22 21:59:04.476  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 2=======
    2019-10-22 21:59:04.476  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD272A0004 cost: 0 ms
    2019-10-22 21:59:04.476  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 3=======
    2019-10-22 21:59:04.476  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD272C0006 cost: 0 ms
    2019-10-22 21:59:04.476  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 4=======
    2019-10-22 21:59:04.476  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD272D0008 cost: 0 ms
    2019-10-22 21:59:04.476  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 5=======
    2019-10-22 21:59:04.476  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD272F000A cost: 0 ms
    2019-10-22 21:59:04.477  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 6=======
    2019-10-22 21:59:04.477  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD2731000C cost: 0 ms
    2019-10-22 21:59:04.477  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 7=======
    2019-10-22 21:59:04.477  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD2733000E cost: 0 ms
    2019-10-22 21:59:04.477  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 8=======
    2019-10-22 21:59:04.477  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD27350010 cost: 0 ms
    2019-10-22 21:59:04.478  INFO 1400 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 9=======
    2019-10-22 21:59:04.478  INFO 1400 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume C0A801040578512DDF1770DD27360012 cost: 0 ms

如果消费者不设置为有序消费的话，则消费的消息则是乱序的：

2019-10-22 22:00:59.922  INFO 2198 --- [MessageThread_2] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 1=======
    2019-10-22 22:00:59.926  INFO 2198 --- [MessageThread_8] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 7=======
    2019-10-22 22:00:59.922  INFO 2198 --- [MessageThread_6] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 5=======
    2019-10-22 22:00:59.922  INFO 2198 --- [MessageThread_3] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 2=======
    2019-10-22 22:00:59.922  INFO 2198 --- [MessageThread_1] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 0=======
    2019-10-22 22:00:59.925  INFO 2198 --- [MessageThread_4] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 3=======
    2019-10-22 22:00:59.926  INFO 2198 --- [MessageThread_7] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 6=======
    2019-10-22 22:00:59.927  INFO 2198 --- [MessageThread_9] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 8=======
    2019-10-22 22:00:59.925  INFO 2198 --- [MessageThread_5] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 4=======
    2019-10-22 22:00:59.927  INFO 2198 --- [essageThread_10] c.l.sbac.rocketmq.order.OrderConsumer    : ========orderly message: 9=======

[springboot_rocketmq]: https://lazycece.github.io/2019/06/16/springboot%E4%B8%ADrocketmq%E7%9A%84%E9%9B%86%E6%88%90%E4%B8%8E%E4%BD%BF%E7%94%A8/