RocketMQ-顺序消息

发表于

这一讲我们来讲顺序消息。

顺序消息是什么

首先，什么是顺序消息？

指的是按照消息的发送顺序来消费。RocketMQ中可以保证消息严格有序，可以分为局部有序和全局有序。

局部有序

什么是局部有序？

在每个MessageQueue里面的每一条消息之间都是保持相对有序的。但是不保证所有MessageQueue的消息都严格有序。

举例例子：
订单1：创建-下单-付款-完成
订单2：创建-下单-付款

订单1和订单2，分别在不同的MessageQueue上，它们只需要保证MessageQueue里面的消息有序即可。

一个MessageQueue只能由一个消费者消费，且只能单线程消费。但是这个消费者可以开启多线程，同时消费多个MessageQueue。

全局有序

既然你已经知道了局部有序，那全局有序就更加简单了。

就是只有一个MessageQueue。这样子所有的消息，都会被发送到这个MessageQueue上。这样子就能保证所有的消息严格有序。

一个MessageQueue只能由一个消费者消费，且只能单线程消费。

生产者顺序发送消息

接下来，我们用代码来展示一下局部有序：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
ini复制代码public class OrderedProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("order_producer_group");


        producer.setNamesrvAddr("localhost:9876");
        //启动生产者
        producer.start();


        List<OrderEntity> list = buildOrderList();
        for (int i = 0; i < list.size(); i++) {
            int orderId = list.get(i).getId();
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("orderTopic", "TagA", "KEY" + i,
                    (list.get(i).toString()).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("订单id:"+orderId+  "  发送结果:"+ sendResult);
        }
        //关闭生产者
        producer.shutdown();
    }

    private static List<OrderEntity> buildOrderList() {
        List<OrderEntity> res =  new ArrayList<>();

        OrderEntity order1 = new OrderEntity(147,"加入购物车");
        OrderEntity order2 = new OrderEntity(147,"下单");
        OrderEntity order3 = new OrderEntity(147,"付款");
        OrderEntity order4 = new OrderEntity(147,"完成");

        OrderEntity order5 = new OrderEntity(258,"加入购物车");
        OrderEntity order6 = new OrderEntity(258,"下单");

        OrderEntity order7 = new OrderEntity(369,"加入购物车");
        OrderEntity order8 = new OrderEntity(369,"下单");
        OrderEntity order9 = new OrderEntity(369,"付款");

        res.add(order1);
        res.add(order2);
        res.add(order3);
        res.add(order4);
        res.add(order5);
        res.add(order6);
        res.add(order7);
        res.add(order8);
        res.add(order9);

        return res;
    }
}
//运行结果：
订单id:147  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B1F80000, offsetMsgId=0AFCA6FA00002A9F000000000009FBA7, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=3], queueOffset=44]
订单id:147  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B1FD0001, offsetMsgId=0AFCA6FA00002A9F000000000009FC96, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=3], queueOffset=45]
订单id:147  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B1FF0002, offsetMsgId=0AFCA6FA00002A9F000000000009FD7C, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=3], queueOffset=46]
订单id:147  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B2010003, offsetMsgId=0AFCA6FA00002A9F000000000009FE62, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=3], queueOffset=47]
订单id:258  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B2020004, offsetMsgId=0AFCA6FA00002A9F000000000009FF48, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=2], queueOffset=42]
订单id:258  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B2040005, offsetMsgId=0AFCA6FA00002A9F00000000000A0037, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=2], queueOffset=43]
订单id:369  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B2050006, offsetMsgId=0AFCA6FA00002A9F00000000000A011D, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=1], queueOffset=63]
订单id:369  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B2060007, offsetMsgId=0AFCA6FA00002A9F00000000000A020C, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=1], queueOffset=64]
订单id:369  发送结果:SendResult [sendStatus=SEND_OK, msgId=7F0000010A0118B4AAC22BE4B2070008, offsetMsgId=0AFCA6FA00002A9F00000000000A02F2, messageQueue=MessageQueue [topic=orderTopic, brokerName=aarondeMacBook-Pro.local, queueId=1], queueOffset=65]

总结：根据不同订单id的取模，把不同订单的消息分配到不同的MessageQueue，把相同订单消息分配到相同的MessageQueue。

你看，订单id=147的消息，都发送都queue3中；

订单id=258的消息，都发送都queue2中；

订单id=369的消息，都发送都queue1中。

消费者顺序消费消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
ini复制代码public class OrderedConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer");
        consumer.setNamesrvAddr("localhost:9876");
        //设置从哪里开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("orderTopic", "TagA");
        //确保一个queue只被一个线程消费
        consumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                       ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("["+Thread.currentThread().getName()+"] "+new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        //启动消费者
        consumer.start();

        System.out.println("消费者已启动");
    }
}
\\运行结果
[ConsumeMessageThread_1] OrderEntity{id=147, name='加入购物车'}
[ConsumeMessageThread_2] OrderEntity{id=258, name='加入购物车'}
[ConsumeMessageThread_3] OrderEntity{id=369, name='加入购物车'}
[ConsumeMessageThread_1] OrderEntity{id=147, name='下单'}
[ConsumeMessageThread_3] OrderEntity{id=369, name='下单'}
[ConsumeMessageThread_1] OrderEntity{id=147, name='付款'}
[ConsumeMessageThread_2] OrderEntity{id=258, name='下单'}
[ConsumeMessageThread_3] OrderEntity{id=369, name='付款'}
[ConsumeMessageThread_1] OrderEntity{id=147, name='完成'}

总结：可以看到，线程1，都是消费了id=147的消息，证明queue3只被一个线程所消息。

对比分析

我们看一下顺序消息的消费者，消费的时候，我们用的是MessageListenerOrderly。是用来告诉消费者，要顺序消费信息，并且只能一个线程去单独消费消息。

普通消息的消费者：MessageListenerConcurrently。看到Concurrent就知道是并发的意思，就是可以并发消费消息。

适用场景

天上飞的理论，终究还得有落地的实现。

适用场景：业务中，需要保持顺序的。比如：数据库的binlog消息，订单的创建、下单、付款等消息。

好了，这一节说得差不多了。

有问题的话，欢迎留言交流。

每日一问

最后，全局有序，你就把MessageQueue设置为1就好。那问题来了，如何设置MessageQueue为1呢？

欢迎留言

后续文章

  • RocketMQ-入门(已更新)
  • RocketMQ-架构和角色(已更新)
  • RocketMQ-消息发送(已更新)
  • RocketMQ-消费信息
  • RocketMQ-消费者的广播模式和集群模式(已更新)
  • RocketMQ-顺序消息(已更新)
  • RocketMQ-延迟消息
  • RocketMQ-批量消息
  • RocketMQ-过滤消息
  • RocketMQ-事务消息
  • RocketMQ-消息存储
  • RocketMQ-高可用
  • RocketMQ-高性能
  • RocketMQ-主从复制
  • RocketMQ-刷盘机制
  • RocketMQ-幂等性
  • RocketMQ-消息重试
  • RocketMQ-死信队列
    …

欢迎各位入(guan)股(zhu)，后续文章干货多多。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%