MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

RocketMQ 消息过滤机制与应用场景

2022-10-014.9k 阅读

RocketMQ 消息过滤机制概述

在分布式系统中，消息队列扮演着至关重要的角色，它负责在不同的组件之间传递消息。RocketMQ 作为一款高性能、高可靠的分布式消息队列，其消息过滤机制为开发者提供了灵活处理消息的能力。消息过滤机制允许消费者根据特定的条件筛选出感兴趣的消息，而不是处理所有发送到队列中的消息。这不仅提高了系统的效率，还减少了不必要的资源消耗。

RocketMQ 支持两种主要的消息过滤方式：基于 Tag 的过滤和基于 SQL92 表达式的过滤。这两种方式在不同的应用场景下各有优劣，开发者可以根据实际需求选择合适的过滤方式。

基于 Tag 的过滤

Tag 的概念

Tag 是 RocketMQ 中用于对消息进行分类的一种简单方式。在发送消息时，生产者可以为消息指定一个或多个 Tag，而消费者在订阅消息时，可以通过指定 Tag 来表明只接收特定 Tag 的消息。Tag 可以理解为消息的一种简单标识，通过它可以快速地对消息进行分类和筛选。

代码示例 - 生产者发送带 Tag 的消息

下面是使用 Java 语言编写的生产者发送带 Tag 消息的示例代码：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class TagProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("TagProducerGroup");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 创建消息实例，指定主题、Tag 和消息体
            Message msg = new Message("TagTopic",
                    "TagA",
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 发送消息
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }

        // 关闭生产者
        producer.shutdown();
    }
}

在上述代码中，我们创建了一个生产者，并向名为 TagTopic 的主题发送了 10 条消息，每条消息都带有 TagA 标签。

代码示例 - 消费者接收带 Tag 的消息

接下来是消费者接收带特定 Tag 消息的代码示例：

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 TagConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TagConsumerGroup");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 设置消费策略，从消息队列的头部开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 订阅主题和 Tag
        consumer.subscribe("TagTopic", "TagA");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        System.out.println(new String(msg.getBody(), "UTF-8"));
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

在这个消费者代码中，我们通过 consumer.subscribe("TagTopic", "TagA") 方法订阅了 TagTopic 主题下带有 TagA 标签的消息。只有符合这个 Tag 条件的消息才会被该消费者接收和处理。

Tag 过滤的优缺点

优点：

  1. 简单易用：Tag 的设置和使用非常简单，生产者只需在发送消息时指定 Tag，消费者在订阅时指定相应的 Tag 即可实现消息过滤。
  2. 性能较高：基于 Tag 的过滤在 Broker 端实现，过滤逻辑相对简单，因此性能开销较小。

缺点：

  1. 过滤条件单一：只能根据 Tag 进行筛选，无法满足复杂的过滤需求。例如，如果需要根据消息体中的某个字段值进行过滤，Tag 方式就无法实现。

基于 SQL92 表达式的过滤

SQL92 表达式过滤原理

RocketMQ 支持使用 SQL92 标准的部分语法来进行消息过滤。这种方式允许消费者根据消息的属性和消息体内容构建复杂的过滤条件。在发送消息时，生产者可以为消息设置自定义属性，然后消费者通过 SQL92 表达式对这些属性进行筛选。SQL92 表达式过滤在 Broker 端执行，能够在消息到达消费者之前就筛选出符合条件的消息。

代码示例 - 生产者发送带属性的消息

以下是生产者发送带有自定义属性消息的代码示例：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class SqlProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("SqlProducerGroup");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 创建消息实例，指定主题和消息体
            Message msg = new Message("SqlTopic",
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 设置消息属性
            msg.putUserProperty("age", String.valueOf(i));
            msg.putUserProperty("gender", i % 2 == 0? "male" : "female");
            // 发送消息
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }

        // 关闭生产者
        producer.shutdown();
    }
}

在上述代码中，我们为每条消息设置了 age 和 gender 两个自定义属性。

代码示例 - 消费者使用 SQL92 表达式过滤消息

下面是消费者使用 SQL92 表达式过滤消息的代码：

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 SqlConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("SqlConsumerGroup");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 设置消费策略，从消息队列的头部开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 订阅主题并设置 SQL92 过滤表达式
        consumer.subscribe("SqlTopic", "age > 5 AND gender = 'female'");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        System.out.println(new String(msg.getBody(), "UTF-8"));
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

在这个消费者代码中，通过 consumer.subscribe("SqlTopic", "age > 5 AND gender = 'female'") 订阅了 SqlTopic 主题，并设置了 SQL92 过滤表达式，只有满足 age > 5 且 gender = 'female' 条件的消息才会被该消费者接收。

SQL92 表达式过滤的优缺点

优点：

  1. 过滤条件丰富：可以使用 SQL92 表达式构建复杂的过滤条件，例如比较大小、逻辑与、逻辑或等，能够满足多样化的过滤需求。
  2. 灵活性高：基于消息属性进行过滤，不受限于简单的 Tag 分类，能够根据业务需求灵活定制过滤规则。

缺点：

  1. 性能开销较大：相比 Tag 过滤，SQL92 表达式的解析和执行需要更多的计算资源，在高并发场景下可能会对 Broker 性能产生一定影响。
  2. 使用相对复杂：需要开发者熟悉 SQL92 语法，编写正确的过滤表达式，对开发者的技能要求相对较高。

RocketMQ 消息过滤的应用场景

业务数据分流

在电商系统中，订单消息可能包含不同类型的订单，如普通订单、团购订单、秒杀订单等。通过为订单消息设置不同的 Tag 或自定义属性，利用 RocketMQ 的消息过滤机制，可以将不同类型的订单消息分发给不同的业务模块进行处理。例如，将团购订单消息发送到专门处理团购业务的模块，而普通订单消息发送到常规订单处理模块。这样可以实现业务数据的分流，提高系统的处理效率和可维护性。

数据筛选与聚合

在大数据分析场景中，系统可能会产生大量的日志消息。通过在日志消息中设置诸如日志级别（如 INFO、WARN、ERROR）、业务模块等属性，使用 SQL92 表达式过滤，可以筛选出特定级别的日志消息或特定业务模块的日志消息进行聚合分析。例如，只筛选出 ERROR 级别的日志消息进行详细分析，以快速定位系统中的问题。

个性化推送

在消息推送系统中，用户可能有不同的兴趣偏好。当用户订阅消息时，可以根据用户的偏好设置相应的过滤条件。例如，用户 A 对体育新闻感兴趣，用户 B 对科技新闻感兴趣。在发送新闻消息时，为每条新闻消息设置诸如新闻类别（体育、科技等）的属性，通过 RocketMQ 的消息过滤机制，只有符合用户兴趣类别的消息才会推送给相应的用户，实现个性化推送。

异步任务处理优化

在一个包含多个异步任务的系统中，不同的任务可能有不同的优先级或处理条件。通过为任务消息设置优先级、任务类型等属性，使用消息过滤机制，可以让具有不同处理能力的消费者节点只接收和处理符合自身条件的任务消息。例如，处理能力较强的节点可以接收优先级较高的任务消息，而处理能力相对较弱的节点处理优先级较低的任务消息，从而优化整个异步任务处理流程。

消息过滤机制在不同架构中的考量

单体架构

在单体架构中，应用程序的所有功能模块都集中在一个进程中。虽然 RocketMQ 的消息过滤机制在单体架构中同样可用，但由于系统规模相对较小，消息处理逻辑相对简单，基于 Tag 的过滤可能就足以满足需求。例如，一个小型的电商单体应用，通过 Tag 对订单消息进行简单分类，将不同类型的订单消息发送到不同的业务逻辑模块进行处理。使用 Tag 过滤既简单又高效，不会引入过多的复杂性。

微服务架构

在微服务架构中，系统由多个独立的微服务组成，每个微服务负责特定的业务功能。不同的微服务可能只对部分消息感兴趣。此时，SQL92 表达式过滤就显得尤为重要。例如，在一个电商微服务架构中，库存微服务可能只关心与库存变动相关的订单消息，通过设置复杂的 SQL92 过滤表达式，如订单商品类别为特定类别且库存操作类型为减库存等条件，库存微服务可以精确地获取到需要处理的消息，避免处理大量无关消息，提高微服务间的协作效率和系统整体性能。

分布式架构

分布式架构下，系统可能跨越多个数据中心或地域。消息过滤机制不仅要考虑过滤条件的准确性，还要考虑网络传输和性能问题。在这种情况下，需要综合使用 Tag 过滤和 SQL92 表达式过滤。对于一些简单的、通用的消息分类，可以使用 Tag 过滤，减少网络传输和 Broker 的处理压力。而对于复杂的、特定业务需求的过滤，则使用 SQL92 表达式过滤。例如，在一个全球分布式的电商系统中，对于不同地区的订单消息，可以先通过 Tag 进行地区分类，然后每个地区的处理节点再根据本地业务需求使用 SQL92 表达式对消息进行进一步过滤。

消息过滤机制的性能优化

合理选择过滤方式

根据实际业务需求，合理选择基于 Tag 的过滤或基于 SQL92 表达式的过滤。如果业务需求简单，只是对消息进行简单分类，如按照消息类型进行区分，使用 Tag 过滤是一个不错的选择，因为其性能开销较小。而当业务需求复杂，需要根据多个属性进行复杂条件筛选时，虽然 SQL92 表达式过滤能满足需求，但要充分考虑其性能影响，在性能要求极高的场景下，可能需要对业务逻辑进行优化，尽量简化过滤条件。

减少不必要的属性设置

在使用 SQL92 表达式过滤时，为消息设置过多不必要的属性会增加消息的大小和 Broker 的处理负担。因此，要根据实际过滤需求，只设置必要的属性。例如，在日志消息过滤场景中，如果只需要根据日志级别和业务模块进行过滤，就无需设置其他无关的属性，这样可以减少消息的冗余，提高系统性能。

优化 SQL92 表达式

编写高效的 SQL92 表达式对于性能提升至关重要。避免使用复杂的嵌套子查询和全表扫描式的表达式。例如，在使用 IN 操作符时，如果列表元素过多，可以考虑使用多个 OR 条件代替，以提高表达式的执行效率。同时，尽量使用索引属性进行过滤，RocketMQ 在处理 SQL92 表达式时，对某些类型的属性查询有优化，合理利用这些特性可以提升过滤性能。

负载均衡与资源分配

在分布式系统中，合理分配 Broker 节点的负载对于消息过滤性能也很关键。可以通过负载均衡算法，将消息过滤任务均匀分配到各个 Broker 节点上，避免某个节点因处理过多复杂的过滤任务而导致性能瓶颈。同时，根据 Broker 节点的硬件资源情况，动态调整其处理的消息过滤任务量，确保系统整体性能的稳定。

消息过滤机制与其他 RocketMQ 特性的结合

与事务消息的结合

在一些涉及到事务性操作的场景中，消息过滤机制可以与事务消息特性相结合。例如，在电商的订单支付场景中，首先发送一个事务消息，消息中包含订单相关的属性和消息体。在事务执行过程中，如果订单支付成功，提交事务消息，此时可以利用消息过滤机制，让订单处理系统根据消息中的属性（如支付金额、订单类型等）进行过滤，只处理符合条件的订单消息。这样可以确保事务性操作与消息过滤的协同工作，提高系统的一致性和可靠性。

与顺序消息的结合

对于顺序消息，消息过滤机制同样可以发挥作用。在一些业务场景中，需要按照特定顺序处理消息，并且只处理符合条件的消息。例如，在物流系统中，订单的状态更新消息需要按照顺序处理，同时只有状态为“已发货”的消息才需要进行下一步的配送处理。通过在发送顺序消息时设置相关属性，如订单状态属性，消费者在订阅时使用 SQL92 表达式过滤出状态为“已发货”的顺序消息进行处理，保证了消息处理的顺序性和准确性。

与批量消息的结合

RocketMQ 支持批量发送和接收消息，消息过滤机制也可以与批量消息特性协同工作。在批量发送消息时，可以为批量消息中的每条消息设置相同或不同的属性，然后消费者在接收批量消息时，通过消息过滤机制筛选出符合条件的消息进行处理。例如，在批量发送订单消息时，为每个订单消息设置订单金额属性，消费者在接收批量消息时，通过 SQL92 表达式过滤出订单金额大于一定值的消息进行特殊处理，提高批量消息处理的灵活性和效率。

实际应用中的注意事项

过滤条件的动态调整

在实际应用中，业务需求可能会发生变化，这就要求消息过滤条件能够动态调整。例如，在电商促销活动期间，可能需要临时调整订单消息的过滤条件，以处理特殊的促销订单。RocketMQ 本身并不直接支持动态修改过滤条件，但可以通过一些间接方式实现，如通过配置中心动态修改消费者订阅的过滤表达式，然后重启消费者或者通过热加载机制动态更新过滤逻辑。

兼容性与版本问题

在使用 RocketMQ 消息过滤机制时，要注意不同版本之间的兼容性。特别是在使用 SQL92 表达式过滤时，不同版本的 RocketMQ 对 SQL92 语法的支持程度可能略有差异。在进行版本升级或系统迁移时，要仔细检查过滤表达式是否仍然有效，避免因版本兼容性问题导致消息过滤失败。

异常处理

在消息过滤过程中，可能会出现各种异常情况，如 SQL92 表达式语法错误、属性不存在等。消费者应用需要有完善的异常处理机制，能够捕获并处理这些异常，避免因异常导致消息处理中断。例如，当 SQL92 表达式语法错误时，消费者应该记录错误日志，并尝试恢复或重新订阅正确的过滤条件。

测试与验证

在将消息过滤机制应用到生产环境之前，要进行充分的测试与验证。包括对不同过滤条件下消息的发送、接收和处理进行功能测试，以及在高并发场景下对过滤性能进行性能测试。通过测试，可以发现潜在的问题，如过滤条件不准确、性能瓶颈等，并及时进行优化，确保消息过滤机制在生产环境中稳定可靠地运行。

总之，RocketMQ 的消息过滤机制为后端开发提供了强大而灵活的消息处理能力。通过深入理解其原理、应用场景以及与其他特性的结合，开发者可以更好地利用这一机制优化分布式系统的消息处理流程，提高系统的性能、可靠性和可维护性。在实际应用中，要注意各种细节和注意事项，通过合理的设计和优化，充分发挥消息过滤机制的优势。