SpringBoot集成RocketMQ使用延时消息与消息过滤-蒲公英云

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

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

先引入依赖

<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 &

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();
    }