RocketMQ 基础教程
首页 死磕 Java 优质系列 Java 技术总览 Java 面试宝典 关于我

RocketMQ 基础教程-应用篇-消息过滤

转
 2024-01-16  阅读(6)
©版权
原文作者：Java 充电社 原文地址： http://www.itsoku.com/course/21/372

消息者在进行消息订阅时，除了可以指定要订阅消息的Topic外，还可以对指定Topic中的消息根据指定条件进行过滤，即可以订阅比Topic更加细粒度的消息类型。

对于指定Topic消息的过滤有两种过滤方式：Tag过滤与SQL过滤。

1、Tag过滤

通过consumer的subscribe()方法指定要订阅消息的Tag。如果订阅多个Tag的消息，Tag间使用或运算符(双竖线||)连接。

    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
    consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

2、SQL过滤

SQL过滤是一种通过特定表达式对事先埋入到消息中的用户属性进行筛选过滤的方式。通过SQL过滤，可以实现对消息的复杂过滤。不过，只有使用PUSH模式的消费者才能使用SQL过滤。

SQL过滤表达式中支持多种常量类型与运算符。

支持的常量类型：

  • 数值：比如：123，3.1415
  • 字符：必须用单引号包裹起来，比如：'abc'
  • 布尔：TRUE 或 FALSE
  • NULL：特殊的常量，表示空

支持的运算符有：

  • 数值比较：>，>=，<，<=，BETWEEN，=
  • 字符比较：=，<>，IN
  • 逻辑运算 ：AND，OR，NOT
  • NULL判断：IS NULL 或者 IS NOT NULL

默认情况下Broker没有开启消息的SQL过滤功能，需要在Broker加载的配置文件中添加如下属性，以开启该功能：

    enablePropertyFilter = true

在启动Broker时需要指定这个修改过的配置文件。

例如对于单机Broker的启动，其修改的配置文件是conf/broker.conf，启动时使用如下命令：

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

3、代码举例

3.1、定义Tag过滤Producer

    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    
    public class FilterByTagProducer {
        public static void main(String[] args) throws Exception {
            DefaultMQProducer producer = new DefaultMQProducer("pg");
            producer.setNamesrvAddr("rocketmqOS:9876");
            producer.start();
            String[] tags = {"myTagA", "myTagB", "myTagC"};
            for (int i = 0; i < 10; i++) {
                byte[] body = ("Hi," + i).getBytes();
                String tag = tags[i % tags.length];
                Message msg = new Message("myTopic", tag, body);
                SendResult sendResult = producer.send(msg);
                System.out.println(sendResult);
            }
            producer.shutdown();
        }
    }

3.2、定义Tag过滤Consumer

    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    public class FilterByTagConsumer {
        public static void main(String[] args) throws Exception {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("pg");
            consumer.setNamesrvAddr("rocketmqOS:9876");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.subscribe("myTopic", "myTagA || myTagB");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt me : msgs) {
                        System.out.println(me);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.println("Consumer Started");
        }
    }

3.3、定义SQL过滤Producer

    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    
    public class FilterBySQLProducer {
        public static void main(String[] args) throws Exception {
            DefaultMQProducer producer = new DefaultMQProducer("pg");
            producer.setNamesrvAddr("rocketmqOS:9876");
            producer.start();
            for (int i = 0; i < 10; i++) {
                try {
                    byte[] body = ("Hi," + i).getBytes();
                    Message msg = new Message("myTopic", "myTag", body);
                    msg.putUserProperty("age", i + "");
                    SendResult sendResult = producer.send(msg);
                    System.out.println(sendResult);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            producer.shutdown();
        }
    }

3.4、定义SQL过滤Consumer

    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.MessageSelector;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    public class FilterBySQLConsumer {
        public static void main(String[] args) throws Exception {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("pg");
            consumer.setNamesrvAddr("rocketmqOS:9876");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.subscribe("myTopic", MessageSelector.bySql("age between 0 and 6"));
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                    for (MessageExt me : msgs) {
                        System.out.println(me);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.println("Consumer Started");
        }
    }

Java 面试宝典是大明哥全力打造的 Java 精品面试题，它是一份靠谱、强大、详细、经典的 Java 后端面试宝典。它不仅仅只是一道道面试题，而是一套完整的 Java 知识体系，一套你 Java 知识点的扫盲贴。

它的内容包括：

  • 大厂真题：Java 面试宝典里面的题目都是最近几年的高频的大厂面试真题。
  • 原创内容：Java 面试宝典内容全部都是大明哥原创，内容全面且通俗易懂，回答部分可以直接作为面试回答内容。
  • 持续更新：一次购买，永久有效。大明哥会持续更新 3+ 年，累计更新 1000+，宝典会不断迭代更新，保证最新、最全面。
  • 覆盖全面：本宝典累计更新 1000+，从 Java 入门到 Java 架构的高频面试题，实现 360° 全覆盖。
  • 不止面试：内容包含面试题解析、内容详解、知识扩展，它不仅仅只是一份面试题，更是一套完整的 Java 知识体系。
  • 宝典详情：https://www.yuque.com/chenssy/sike-java/xvlo920axlp7sf4k
  • 宝典总览：https://www.yuque.com/chenssy/sike-java/yogsehzntzgp4ly1
  • 宝典进展：https://www.yuque.com/chenssy/sike-java/en9ned7loo47z5aw

目前 Java 面试宝典累计更新 400+ 道，总字数 42w+。大明哥还在持续更新中，下图是大明哥在 2024-12 月份的更新情况：

想了解详情的小伙伴，扫描下面二维码加大明哥微信【daming091】咨询

同时，大明哥也整理一套目前市面最常见的热点面试题。微信搜[大明哥聊 Java]或扫描下方二维码关注大明哥的原创公众号[大明哥聊 Java] ，回复【面试题】 即可免费领取。

阅读全文
文章标签： RocketMQ