概述

官方推荐的是一个系统就用一个topic,然后系统下的类目就用tag来区分,比如说订单系统,整个订单就放到一个topic下,是下单还是支付还是退款等等发消息的时候通过不同的tag来区分.

broker再推送消息给consumer的时候就提前通过tag来判断是否推送到这个consumer了,如果你这个consumer没配置这个tag,那么broker就不会把这个tag推送给你.这样的好处就可以减少网络IO.

SQL92语法

RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。

  • 数值比较，比如：>，>=，<，<=，BETWEEN，=；
  • 字符比较，比如：=，<>，IN；
  • IS NULL 或者 IS NOT NULL；
  • 逻辑符号 AND，OR，NOT；

常量支持类型为：

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

使用注意!

只有推模式的消费者可以使用SQL过滤。拉模式是用不了的。

代码案例

生产者

  1. package org.apache.rocketmq.example.filter;
  3. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  4. import org.apache.rocketmq.client.producer.SendResult;
  5. import org.apache.rocketmq.common.message.Message;
  6. import org.apache.rocketmq.remoting.common.RemotingHelper;
  8. public class TagFilterProducer {
  10.     public static void main(String[] args) throws Exception {
  12.         DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
  13.         producer.setNamesrvAddr("zjj101:9876;zjj102:9876;zjj103:9876");
  14.         producer.start();
  15.     //发送的时候设置tag
  16.         String[] tags = new String[]{"TagA", "TagB", "TagC"};
  18.         for (int i = 0; i < 15; i++) {
  19.             Message msg = new Message("TagFilterTest",
  20.                     tags[i % tags.length],
  21.                     "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
  23.             SendResult sendResult = producer.send(msg);
  24.             System.out.printf("%s%n", sendResult);
  25.         }
  27.         producer.shutdown();
  28.     }
  29. }

消费者

  1. package org.apache.rocketmq.example.filter;
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  6. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  7. import org.apache.rocketmq.client.exception.MQClientException;
  8. import org.apache.rocketmq.common.message.MessageExt;
  10. import java.io.IOException;
  11. import java.util.List;
  13. public class TagFilterConsumer {
  15.     public static void main(String[] args) throws InterruptedException, MQClientException, IOException {
  17.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
  18.         consumer.setNamesrvAddr("zjj101:9876;zjj102:9876;zjj103:9876");
  19.         /*  TagA || TagC  ,意思是 只有 TagA 或者 TagC 的消息会被接收 */
  20.         consumer.subscribe("TagFilterTest", "TagA || TagC");
  22.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  24.             @Override
  25.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
  26.                                                             ConsumeConcurrentlyContext context) {
  27.                 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
  28.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  29.             }
  30.         });
  32.         consumer.start();
  34.         System.out.printf("Consumer Started.%n");
  35.     }
  36. }

sql过滤

“TagA || TagC” 的写法一个消息只能有一个tag,如果想用多个tag来过滤的话,”TagA || TagC”这样的写法就做不到了.所以就使用sql过滤.

代码演示

Linux配置

在broker.properties那里配置,配置完了需要重启Broker集群

  1. #是否支持根据属性过滤。如果使用基于标准的sql92模式过滤消息则改参数必须设置为true。 
  2. enablePropertyFilter=true

生产者

  1. package org.apache.rocketmq.example.filter;
  3. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  4. import org.apache.rocketmq.client.producer.SendResult;
  5. import org.apache.rocketmq.common.message.Message;
  6. import org.apache.rocketmq.remoting.common.RemotingHelper;
  8. public class SqlFilterProducer {
  10.     public static void main(String[] args) throws Exception {
  12.         DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
  13.         producer.setNamesrvAddr("zjj101:9876;zjj102:9876;zjj103:9876");
  14.         producer.start();
  16.         String[] tags = new String[]{"TagA", "TagB", "TagC"};
  18.         for (int i = 0; i < 15; i++) {
  19.             // 发送消息的时候指定tag
  20.             Message msg = new Message("SqlFilterTest",
  21.                     tags[i % tags.length],
  22.                     ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
  23.             );
  24.             // 设置用户属性
  25.             msg.putUserProperty("a", String.valueOf(i));
  27.             SendResult sendResult = producer.send(msg);
  28.             System.out.printf("%s%n", sendResult);
  29.         }
  31.         producer.shutdown();
  32.     }
  33. }

消费者

  1. package org.apache.rocketmq.example.filter;
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.MessageSelector;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  6. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  7. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  8. import org.apache.rocketmq.common.message.MessageExt;
  10. import java.util.List;
  12. public class SqlFilterConsumer {
  14.     public static void main(String[] args) throws Exception {
  16.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
  17.         consumer.setNamesrvAddr("zjj101:9876;zjj102:9876;zjj103:9876");
  19.         //sql选择器
  20.         /*下面的sql选择器是 tags不为空,并且tags 是 'TagA' 或者'TagB' ,
  21.         并且MQ的UserProperty里面的a属性不能是null 并且 a属性必须在 0~3范围 , 包括0和3
  22.         */
  23.         consumer.subscribe("SqlFilterTest",
  24.                 MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
  25.                         "and (a is not null and a between 0 and 3)"));
  27.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  29.             @Override
  30.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
  31.                                                             ConsumeConcurrentlyContext context) {
  32.                 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
  33.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  34.             }
  35.         });
  37.         consumer.start();
  38.         System.out.printf("Consumer Started.%n");
  39.     }
  40. }

查看效果

先执行消费者, 再执行生产者

控制台消息我就不贴了, 符合sql表达式的消息会被consumer接收到,不符合的就不会被接收到.