RocketMQ 如何实现消息的过滤？ – 编程技术之美-IT之美

发表于2023年10月27日2023年6月10日作者 IT之美

RocketMQ提供多种方式实现消息过滤:

Tag过滤:生产者在发送消息时,可以指定Tag进行标记,消费者在订阅Topic时可以指定Tag进行过滤订阅,从而实现Tag级别的过滤。

生产者发送指定Tag消息:DefaultMQProducer#send(Message msg, String tag)
消费者过滤订阅:DefaultMQPushConsumer#subscribe(String topic, String expression)

表达式示例: ||TagA||TagB 表示订阅TagA或TagB的消息

SQL过滤:消费者使用消费者端过滤配置进行SQL过滤,只消费满足条件的消息。

配置SQL过滤:DefaultMQPushConsumer#subscribe(String topic, MessageSelector messageSelector)

MessageSelector示例:

messageSelector = "id between 0 and 3 and age > 30"

自定义过滤器:开发者可以通过实现Filter接口,并设置到消费者来自定义消息过滤逻辑。

实现自定义Filter:

public class MyFilter implements Filter {
    @Override
    public boolean match(MessageExt msg) {
        // 消息过滤逻辑
        return msg.getTopic().equals("TopicTest");
    }
}

设置自定义Filter到消费者:defaultMQPushConsumer.setFilter(new MyFilter());