云消息队列 RocketMQ 版：消息过滤

应用场景

云消息队列 RocketMQ 版作为发布订阅模型的消息中间件广泛应用于上下游业务集成场景。在实际业务场景中，同一个主题下的消息往往会被多个不同的下游业务方处理，各下游的处理逻辑不同，只关注自身逻辑需要的消息子集。

使用云消息队列 RocketMQ 版的消息过滤功能，可以帮助消费者更高效地过滤自己需要的消息集合，避免大量无效消息投递给消费者，降低下游系统处理压力。

云消息队列 RocketMQ 版主要解决的单个业务域即同一个主题内不同消息子集的过滤问题，一般是基于同一业务下更具体的分类进行过滤匹配。如果是需要对不同业务域的消息进行拆分，建议使用不同主题处理不同业务域的消息。

功能概述

消息过滤定义

过滤的含义指的是将符合条件的消息投递给消费者，而不是将匹配到的消息过滤掉。

云消息队列 RocketMQ 版的消息过滤功能通过生产者和消费者对消息的属性、标签进行定义，并在云消息队列 RocketMQ 版服务端根据过滤条件进行筛选匹配，将符合条件的消息投递给消费者进行消费。

消息过滤原理

消息过滤分类

具体的使用方式及示例，请参见下文的Tag标签过滤和SQL属性过滤。

订阅关系一致性

过滤表达式属于订阅关系的一部分，云消息队列 RocketMQ 版的领域模型规定，同一消费者分组内的多个消费者的订阅关系包括过滤表达式，必须保持一致，否则可能会导致部分消息消费不到。更多信息，请参见订阅关系（Subscription）。

Tag标签过滤

Tag标签过滤方式是云消息队列 RocketMQ 版提供的基础消息过滤能力，基于生产者为消息设置的Tag标签进行匹配。生产者在发送消息时，设置消息的Tag标签，消费者需指定已有的Tag标签来进行匹配订阅。

场景示例

Tag标签设置

• Tag由生产者发送消息时设置，每条消息允许设置一个Tag标签。

• Tag使用可见字符，建议长度不超过128字符。

• Tag匹配时大小写敏感，例如，TAG A和tag a属于两个标签。

Tag标签过滤规则

使用示例

• 发送消息，设置Tag标签。

  Message message = messageBuilder.setTopic("topic")
                  //设置消息索引键，可根据关键字精确查找某条消息。
                  .setKeys("messageKey")
                  //设置消息Tag，用于消费端根据指定Tag过滤消息。
                  //该示例表示消息的Tag设置为“TagA”。
                  .setTag("TagA")
                  //消息体。
                  .setBody("messageBody".getBytes())
                  .build();
                      

• 订阅消息，匹配单个Tag标签。

  String topic = "Your Topic";
  //只订阅消息标签为“TagA”的消息。
  FilterExpression filterExpression = new FilterExpression("TagA", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

• 订阅消息，匹配多个Tag标签。

  String topic = "Your Topic";
  //只订阅消息标签为“TagA”、“TagB”或“TagC”的消息。
  FilterExpression filterExpression = new FilterExpression("TagA||TagB||TagC", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

• 订阅消息，匹配Topic中的所有消息，不进行过滤。

  String topic = "Your Topic";
  //使用Tag标签过滤消息，订阅所有消息。
  FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

SQL属性过滤

SQL属性过滤是云消息队列 RocketMQ 版提供的高级消息过滤方式，通过生产者为消息设置的属性（Key）及属性值（Value）进行匹配。生产者在发送消息时可设置多个属性，消费者订阅时可设置SQL语法的过滤表达式过滤多个属性。

场景示例

消息属性设置

生产者发送消息时可以自定义消息属性，每个属性都是一个自定义的键值对（Key-Value）。

每条消息支持设置多个属性。

SQL属性过滤规则

SQL属性过滤使用SQL92语法作为过滤规则表达式，语法规范如下：

使用示例

• 发送消息，同时设置消息Tag标签和自定义属性。

  Message message = messageBuilder.setTopic("topic")
                  //设置消息索引键，可根据关键字精确查找某条消息。
                  .setKeys("messageKey")
                  //设置消息Tag，用于消费端根据指定Tag过滤消息。
                  //该示例表示消息的Tag设置为“messageTag”。
                  .setTag("messageTag")
                  //消息也可以设置自定义的分类属性，例如环境标签、地域、逻辑分支。
                  //该示例表示为消息自定义一个属性，该属性为地域，属性值为杭州。
                  .addProperty("Region", "Hangzhou")
                  //消息体。
                  .setBody("messageBody".getBytes())
                  .build();

• 订阅消息，根据单个自定义属性匹配消息。

  String topic = "topic";
  //只订阅地域属性为杭州的消息。
  FilterExpression filterExpression = new FilterExpression("Region IS NOT NULL AND Region='Hangzhou'", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

• 订阅消息，同时根据多个自定义属性匹配消息。

  String topic = "topic";
  //只订阅地域属性为杭州且价格属性大于30的消息。
  FilterExpression filterExpression = new FilterExpression("Region IS NOT NULL AND price IS NOT NULL AND Region = 'Hangzhou' AND price > 30", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

• 订阅消息，匹配Topic中的所有消息，不进行过滤。

  String topic = "topic";
  //订阅所有消息。
  FilterExpression filterExpression = new FilterExpression("True", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

使用建议

合理划分主题和Tag标签

消息过滤常见问题

多个消费者订阅同一个Topic下的不同Tag，出现消息丢失情况。

可能原因：若多个消费者是通过同一个消费者分组（Group ID）订阅的指定Topic，则所有消费者的过滤条件即订阅的Tag要一致，否则会出现订阅关系不一致，导致部分消息丢失。

消费者设置了过滤条件，消息消费数量是按照过滤前还是过滤后计算呢？

按照过滤后的消息数量计算。

消费者在线无消费消息，但Group有堆积。

采用SQL/TAG消费过滤的方式，未被过滤条件命中的消息会计算为消息堆积，堆积量的计算如下所示。

• SQL消费方式：堆积量 = 已就绪的消息量 + 处理中的消息量 - 未被SQL命中的消息数量。

• TAG消费方式：堆积量 =（已就绪的消息量 + 处理中的消息量）* TAG标签消息百分比。

说明

TAG标签消息百分比 = 采样内TAG标签消息量 / 采样消息总量

相关文档

完整的消息收发示例代码，请参见SDK参考概述。