RocketMQ的有序消息

电玩女神 2022-04-05 02:57 241阅读 0赞

我们知道Topic的有序消息已经成为mq的标配。而RocketMQ中是这样区分消息类型的， 普通消息也叫做无序消息，简单来说就是没有顺序的消息，而**有序消息**就是按照一定的先后顺序的消息类型。举个例子，producer 依次发送 order id 为 1、2、3 的消息到 broker，consumer 接到的消息顺序也就是 1、2、3 ，而不会出现普通消息那样的 2、1、3 等情况。

**那有序消息该如何实现呢？**

理论上：我们都知道消息首先由 producer 到 broker，再从 broker 到 consumer，分这两步走。那么要保证消息的有序，势必这两步都是要保证有序的，即要保证消息是按有序发送到 broker，broker 也是有序将消息投递给 consumer，两个条件必须同时满足，缺一不可。

**实现方式呢？**

**全局有序消息**：由于一个 topic 只有一个 queue ，即使我们有多个 producer 实例和 consumer 实例也很难提高消息吞吐量。就好比过独木桥，大家只能一个挨着一个过去，效率低下。

**局部有序消息**：常见做法就是将 order id 进行处理，将 order id 相同的消息发送到 topicB 的同一个 queue，假设我们 topicB 有 2 个 queue，那么我们可以简单的对 id 取余，奇数的发往 queue0，偶数的发往 queue1，消费者按照 queue 去消费时，就能保证 queue0 里面的消息有序消费，queue1 里面的消息有序消费。

**科普一下RocketMQ的topic**

topic 只是消息的逻辑分类，内部实现其实是由 queue 组成。当 producer 把消息发送到某个 topic 时，默认是会消息发送到具体的 queue 上。由于一个 topic 可以有多个 queue，所以在性能比全局有序高得多。假设 queue 数是 n，理论上性能就是全局有序的 n 倍，当然 consumer 也要跟着增加才行。在实际情况中，这种局部有序消息是会比全局有序消息用的更多。

public class Producer {
        public static void main(String[] args) throws UnsupportedEncodingException {
            try {
                // 声明并初始化一个producer
                // 需要一个producer group名字作为构造方法的参数，这里为ordered_producer
                DefaultMQProducer orderedProducer = new DefaultMQProducer("ordered_producer");
    
                // 设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
                //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
                orderedProducer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
    
                // 调用start()方法启动一个producer实例
                orderedProducer.start();
    
                // 自定义一个tag数组
                String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
    
                // 发送10条消息到Topic为TopicTestOrdered，tag为tags数组按顺序取值，
                // key值为“KEY”拼接上i的值，消息内容为“Hello RocketMQ”拼接上i的值
                for (int i = 0; i < 10; i++) {
    
                    int orderId = i % 10;
                    Message msg =
                            new Message("TopicTestOrdered", tags[i % tags.length], "KEY" + i,
                                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // 这里设置需要延时的等级即可
                    msg.setDelayTimeLevel(3);
                    SendResult sendResult = orderedProducer.send(msg, new MessageQueueSelector() {
    
                        // 选择发送消息的队列
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
    
                            // arg的值其实就是orderId
                            Integer id = (Integer) arg;
    
                            // mqs是队列集合，也就是topic所对应的所有队列
                            int index = id % mqs.size();
    
                            // 这里根据前面的id对队列集合大小求余来返回所对应的队列
                            return mqs.get(index);
                        }
                    }, orderId);
                    System.out.println(sendResult);
                }
                orderedProducer.shutdown();      
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    public class Consumer {
    
        public static void main(String[] args) throws MQClientException {
    
            //声明并初始化一个consumer
            //需要一个consumer group名字作为构造方法的参数，这里为concurrent_consumer
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ordered_consumer");
    
            //同样也要设置NameServer地址
            consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
    
            //这里设置的是一个consumer的消费策略
            //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
            //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
            //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            //设置consumer所订阅的Topic和Tag
            consumer.subscribe("TopicTestOrdered", "TagA || TagC || TagD");
    
            //设置一个Listener，主要进行消息的逻辑处理
            //注意这里使用的是MessageListenerOrderly这个接口
            consumer.registerMessageListener(new MessageListenerOrderly() {
    
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
    
                    System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
    
                    //返回消费状态
                    //SUCCESS 消费成功
                    //SUSPEND_CURRENT_QUEUE_A_MOMENT 消费失败，暂停当前队列的消费
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
    
            //调用start()方法启动consumer
            consumer.start();
    
            System.out.println("Consumer Started.");
        }
    }

**延时消息呢？**

**延时消息**，简单来说就是当 producer 将消息发送到 broker 后，会延时一定时间后才投递给 consumer 进行消费。RcoketMQ的延时等级为：1s，5s，10s，30s，1m，2m，3m，4m，5m，6m，7m，8m，9m，10m，20m，30m，1h，2h。level=0，表示不延时。level=1，表示 1 级延时，对应延时 1s。level=2 表示 2 级延时，对应5s，以此类推。这种消息一般适用于消息生产和消费之间有时间窗口要求的场景。比如说我们网购时，下单之后是有一个支付时间，超过这个时间未支付，系统就应该自动关闭该笔订单。那么在订单创建的时候就会就需要发送一条延时消息（延时15分钟）后投递给 consumer，consumer 接收消息后再对订单的支付状态进行判断是否关闭订单。