RocketMQ消息过滤是如何实现的？

三种消息过滤方式

RocketMQ的过滤方式主要分为如下2种，其中类模式过滤的方式会在RocketMQ 5.0.0版本中被移除，所以不进行详细分析；

根据Tag过滤

public class TagFilterProducer {

    public static final String RPODUCER_GROUP_NAME = "tagFilterProducerGroup";
    public static final String TOPIC_NAME = "tagFilterTopic";

    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(RPODUCER_GROUP_NAME);
        producer.setNamesrvAddr("192.168.199.138:9876");
        producer.setSendMsgTimeout(6000);
        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC"};

        for (int i = 0; i < 12; i++) {
            Message message = new Message(TOPIC_NAME, tags[i % tags.length], ("hello rocketmq " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(message);
            System.out.println(sendResult);
        }

        producer.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Producer端总共发送了12条数据。TagA，TagB，TagC的数据各4条。

public class TagFilterConsumer {

    public static final String CONSUMER_GROUP_NAME = "tagFilterConsumerGroup";

    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP_NAME);
        consumer.setNamesrvAddr("192.168.199.138:9876");
        consumer.subscribe(TagFilterProducer.TOPIC_NAME, "TagA || TagC");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.printf("%s receive new message %s", Thread.currentThread().getName(), list);
                System.out.println();
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Consumer Started");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Consumer端订阅了TagA和TagC的数据，最终只接收到8条数据，TagA和TagC的数据各4条

根据SQL92语法过滤

在发送消息的时候，可以给消息设置很多扩展属性，SQL92语法过滤就是根据这些扩展属性来过滤消息的（其实tag也是扩展属性的一种）

rocketmq默认是不支持sql92语法过滤的，需要在broker.conf中加入如下配置

enablePropertyFilter = true
filterSupportRetry = true
1
2

public class SqlFilterProducer {

    public static final String RPODUCER_GROUP_NAME = "sqlFilterProducerGroup";
    public static final String TOPIC_NAME = "sqlFilterTopic";

    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(RPODUCER_GROUP_NAME);
        producer.setNamesrvAddr("192.168.199.138:9876");
        producer.setSendMsgTimeout(6000);
        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC"};

        for (int i = 0; i < 12; i++) {
            Message message = new Message(TOPIC_NAME, tags[i % tags.length], ("hello rocketmq " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            message.putUserProperty("a", String.valueOf(i));
            SendResult sendResult = producer.send(message);
            System.out.println(sendResult);
        }

        producer.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

public class SqlFilterConsumer {

    public static final String CONSUMER_GROUP_NAME = "sqlFilterConsumerGroup";

    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP_NAME);
        consumer.setNamesrvAddr("192.168.199.138:9876");
        consumer.subscribe(SqlFilterProducer.TOPIC_NAME,
                MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))"
                + "and (a is not null and a between 0 and 3)"));

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.printf("%s receive new message %s \n", Thread.currentThread().getName(), list);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Consumer Started");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

生产者发送了12条数据，消费者消费了2条数据（tag为TagA或TagB，属性a的值在0到3之间）

原理解析

因为通过tag过滤的方式最常用，因此只分析tag过滤的实现过程。

如果让你实现通过tag过滤你会在Consumer端进行过滤还是在Broker端进行过滤？

可能大多数人都会选择在Broker端进行过滤，因为在Consumer端进行过滤，会有无用的消息被发送到Consumer端，造成带宽浪费

其实RocketMQ根据Tag过滤消息的方式需要Consumer端和Broker端一起来完成

1. Broker端收到消息后，先根据ConsumerQueue中的tagCods（tag的hashcode）进行过滤，过滤完成后从CommitLog取出对应的消息值，然后返回给Consumer; (节省宽带浪费))
2. Consumer收到消息后再根据tag的具体值进行过滤，然后再开始消费消息（因为有可能2个tag的内容不同，但是hashcode相同，所以还需要根据内容再过滤一次）
   
   可能你会说那目前这种方式可能会有大量不匹配的消息发送到客户端！

首先tag的hashcode冲突的概率比较低。其次，你完全可以在设置tagName的时候避免和之前tagName的hashcode产生冲突;

一个消费组订阅一个主题下的不同tag，为什么会会丢消息？

1. 消息队列的负载均衡，一个消费端只会分配到一个或者多个队列(负载均衡)，在同一个时间，一个队列只能分配到一个消费者；
2. 同一个tag会分布在不同的队列中，但消费者C1分配到的队列为q0,q1，q0,q1中有taga和tagb的消息，但tagb的消息会被消费者C1过滤，但这部分消息，却不会被C2消费，造成了消息丢失。