SpringBoot集成RocketMQ使用延时消息与消息过滤

我会带着你远行 2024-03-31 13:17 90阅读 0赞

#### 1.首先了解一下什么是延时消息 ####

延时消息用来指定消息发送到消息队列(RocketMQ)的服务端后，延时一段时间之后才被投递到客户端进行消费(例如半分钟之后)，适用于解决一些消息的生产和消费有窗口弹出要求的场景。例如：电商交易中超过时间未支付则关闭订单，在订单创建时，发送一条延时消息，这条消息将在30分钟以后投递给消费者，消费者受到此消息之后，判断对应的订单是否已支付，如果支付未完成则关闭订单，删除数据，恢复库存，如果已完成支付则忽略。比如：

![dc167d2478094475ad40f9b17fedda78.png][]

#### 先引入依赖 ####

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

#### 1.创建延时消息生产者样例 ####

// 延时消息
        public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
    
            //创建 producer
            DefaultMQProducer producer = new DefaultMQProducer("tmXBL");
            //设置ip地址
            producer.setNamesrvAddr("192.168.1.7:9876");
            //开始加载
            producer.start();
            //编辑消息
            String body = "{userName:'Lix',hobby:'延时消息'}";
            //创建消息
            Message message = new Message("topicXBL","tagsXBL",body.getBytes());
            //设置延时消息
            //注意，这是设置消息的延时等级
            //1  2  3  4  5  6  7  8  9  10  11  12  13  14  15  16  17  18
            //1s 5s 10s30s1m 2m 3m 4m 5m 6m  7m  8m  9m  10m 20m 30m 1h  2h
            message.setDelayTimeLevel(6);
            //发送消息
            SendResult send = producer.send(message);
            System.out.println(send);
            //关闭链接
            producer.shutdown();
        }

说明：现在往rocketMQ中发送一条延时消息（2分钟后roketMQ才会接收到）

#### 2.创建延时消息消费者样例 ####

//消费延时消息
        public static void main(String[] args) throws MQClientException {
    
            //创建消费对象
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_XBL");
            //通过NameSrv设置ip+端口
            consumer.setNamesrvAddr("192.168.1.7:9876");
            //设置要消费信息的范围
            consumer.subscribe("topicXBL","tagsXBL");
            //设置消息监听
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    //setDelayLevelWhenNextConsume设置消费间隔
                    consumeConcurrentlyContext.setDelayLevelWhenNextConsume(3);
                    //获取每条消息的延时时间
                    for (MessageExt messageExt : list) {
                        byte[] body = messageExt.getBody();
                        String str = new String(body);
                        System.out.println(str);
                        long storeTimestamp = messageExt.getStoreTimestamp();
                        System.out.println("存储时间为："+storeTimestamp);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            //直接消费
            consumer.start();
            //    //消费者需要关闭吗？ 不用
            //    //1、我们在项目中使用消息消费者，是需要持续不断的读取MQ中的消息，
            //    //所以不用关闭
            //    //2、在读取消息时，是自动开启另一线程，和当前代码不是同时执行
            //    //可能造成，上面的consumeMessage方法还没走完，却已经consumer.shutdown了
            //    //consumer.shutdown();
        }

#### 3.消息过滤 ####

注意，在我们需要使用过滤时，需要提前设置broker的允许过滤，否则将会报错：Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: CODE: 1 DESC: The broker does not support consumer to filter message by SQL92

怎么设置？

进入rocketmq的conf文件夹中，编辑broker.conf

添加一行：

enablePropertyFilter = true

保存退出即可

然后关闭RocketMQ

返回目录后使用

nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &

![ba8e49e967f24fdd9b54dcf37adb1539.png][]

![ee25013baf3945af91c088195a1c2936.png][]

#### 4. 准备一条消息后续过滤 ####

//RocketMQ的消息过滤
        public static void main(String[] args) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
           //我们使用RocketMQ消费消息时，大多数情况是会使用到Tags的，通过Tags我们可以过滤掉我们想要的数据
            //列如：consumer.subscribe("topicTmXBL","TagsTmXBL||TagsTmXBL2");
            //这种情况是消费者直接找标签TagsXBL或者TagsXBL2的数据，但是限制是一个消息只能有一个标签
            //对于复杂场景来说，可能就不够用了，我们可以使用SQL表达式来筛选信息，SQL特性可以通过
            //发送消息的属性来进行计算，使用RocketMQ可以实现一些简单的逻辑
            //例如 AND 、BETWEEN 、 >= 、 <= 、 <> 、 OR 、 IS 、 NULL 等等
    
            //首先需要知道：我们想要使用的SQL特性，不是说直接过滤Tags， 因为一个标签就是一个单词而已
            //一个标签对应一个业务， 我们可以在消息生成者的代码中，额外加一些属性，就可以完成过滤了 
    
    
            //前两步不变
            DefaultMQProducer producer = new DefaultMQProducer("tm_XBL");
            producer.setNamesrvAddr("192.168.1.7:9876");
            //需要将producer对象启动起来
            producer.start();
    
            //编辑信息
            String body="userName:'Tom',hobby:'Jerry'";
            //创建消息
            Message message=new Message("topicTmXBL","tagsTmXBL","keysTmXBL",body.getBytes());
    
            //加上条件属性
            message.putUserProperty("isMember","0");
            message.putUserProperty("MemberLever","9");
    
            //发送信息
            producer.send(message);
    
            //关闭链接
            producer.shutdown();
    
        }

说明：这条消息跟其他消息不同点就我们使用 message.putUserProperty()设置2条属性，为后续过滤产生了条件

#### 5.执行消费者过滤 ####

//消费消息时进行过滤
        public static void main(String[] args) throws MQClientException {
    
            //创建消费者对象
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tmConsumerXBL233");
            //通过NameSrv设置IP+端口
            consumer.setNamesrvAddr("192.168.1.7:9876");
            //消费消息时进行消息过滤，需要注意，如果一个消息被过滤掉了
            //(实际上已经消费了，但是不符合条件没有筛选出来)，则当前消费组也在后续的消费中，不会继续消费已经过滤掉的消息。
            consumer.subscribe("topicTmXBL", MessageSelector.bySql("isMember <= 0 AND MemberLever >= 5  "));
            //设置消息监听
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                    //list就是根据条件得到的数据
                    //consumeOrderlyContext就是设置各种属性的上下文对象
                    //是否设置消费的消息在 被消费后被标记为以消费，相当于是删除的意思
                    consumeOrderlyContext.setAutoCommit(true);
                    list.forEach(a->{
                        byte[] body = a.getBody();
                        String str = new String(body);
                        System.out.println(str);
                    });
    
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
    
        }

说明：现在使用MessageSelector.bySql(定义的sql条件)输出结果会显示一条数据，如果更改条件使得条件不成立则消息会被过滤，且当前消费组无法重复消费。

#### 总结： ####

本文主要讲解rocketMQ在java中如何使用延时消息与消息过滤，使用起来比较简单多多支持点赞哈

[dc167d2478094475ad40f9b17fedda78.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/e87725dc2449462c9c2041711ff2aba8.png
[ba8e49e967f24fdd9b54dcf37adb1539.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/a883431a55f24c969e12be27dc61c719.png
[ee25013baf3945af91c088195a1c2936.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/7838a41705d64cc0b895fee74ba20bb2.png