发送与接收过滤消息

操作场景

本文以调用 Java SDK 为例介绍通过开源 SDK 实现过滤消息收发的操作过程。目前支持两种，分别是Tag 和 SQL 方式

前提条件

• 完成资源创建与准备（如果是全局顺序消息，需要创建单队列topic）
• 安装1.8或以上版本 JDK
• 安装2.5或以上版本 Maven
• 参考普通消息的发送和接收

TAG使用步骤

创建生产者消费者的基础代码不在赘述，可以参考普通消息的发送和接收即可。

• 对于生产消息，主要是在构造消息体的时候，带上TAG
• 对于消费消息，主要是订阅的时候，带上单个TAG 或者 * 或者多个TAG表达式

步骤1：生产消息

发送消息

发送代码和简单的消息没有区别 主要是在构造消息体的时候，带上TAG，仅允许一个。

int totalMessagesToSend = 5;
for (int i = 0; i < totalMessagesToSend; i++) {
    Message msg = new Message(TOPIC_NAME, "Tag1", "Hello RocketMQ.".getBytes(StandardCharsets.UTF_8));
    // 发送消息
    SendResult sendResult = producer.send(message);
    System.out.println("sendResult = " + sendResult);
}

步骤2：消费消息

订阅消息。

// 订阅topic 订阅所有的TAG
pushConsumer.subscribe(topic_name, "*");

//订阅指定的TAG
//pushConsumer.subscribe(TOPIC_NAME, "Tag1");  

//订阅多个TAG
//pushConsumer.subscribe(TOPIC_NAME, "Tag1||Tag2");   
      
// 注册回调实现类来处理从broker拉取回来的消息
pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
          // 消息处理逻辑
          System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
          // 标记该消息已经被成功消费, 根据消费情况，返回处理状态
          return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// 启动消费者实例
pushConsumer.start();

SQL 使用步骤

创建生产者消费者的基础代码不在赘述，可以参考普通消息的发送和接收即可。

• 对于生产消息，主要是在构造消息体的时候，带上用户自定义的properties
• 对于消费消息，主要是订阅的时候，带上对应的SQL表达式

步骤1：生产消息

发送代码和简单的消息没有区别 主要是在构造消息体的时候，带上自定义属性，允许多个。

int totalMessagesToSend = 5;
for (int i = 0; i < totalMessagesToSend; i++) {
    Message msg = new Message(TOPIC_NAME,"Hello RocketMQ.".getBytes(StandardCharsets.UTF_8));
    msg.putUserProperty("key1","value1");
    // 发送消息
    SendResult sendResult = producer.send(message);
    System.out.println("sendResult = " + sendResult);
}

步骤2：消费消息

对于消费消息，主要是订阅的时候，带上对应的SQL表达式，其他的和普通的没有区别

pushConsumer.subscribe(TOPIC_NAME, MessageSelector.bySql("True"));  

// 订阅topic 订阅单个key的sql，最常用
//pushConsumer.subscribe(TOPIC_NAME,    MessageSelector.bySql("key1 IS NOT NULL AND key1='value1'"));

//订阅多个属性
//pushConsumer.subscribe(TOPIC_NAME,    MessageSelector.bySql("key1 IS NOT NULL AND key2 IS NOT NULL  AND key1='value1' AND key2='value2'"));
      
// 注册回调实现类来处理从broker拉取回来的消息
pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
          // 消息处理逻辑
          System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
          // 标记该消息已经被成功消费, 根据消费情况，返回处理状态
          return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// 启动消费者实例
pushConsumer.start();