┌─────────────────────────────────────────────────────────────────┐
│                      消息过滤机制                                 │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   Producer ──► [msg1, msg2, msg3, msg4] ──► Broker              │
│                                                        │         │
│                                                        ▼         │
│   ┌─────────────────────────────────────────────────────────┐   │
│   │                    消息过滤                               │   │
│   │                                                         │   │
│   │   ┌───────────┐    ┌───────────┐    ┌───────────┐     │   │
│   │   │ Tag 过滤  │    │SQL92 过滤 │    │ 类过滤    │     │   │
│   │   └───────────┘    └───────────┘    └───────────┘     │   │
│   │                                                         │   │
│   └─────────────────────────────────────────────────────────┘   │
│                          │                                       │
│                          ▼                                       │
│   Consumer ◄─── 只接收过滤后的消息                               │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

/**
 * Tag 过滤
 */
public class TagFilterProducer {

    public void sendWithTags() {
        // 发送不同 Tag 的消息
        producer.send(new Message("ORDER_TOPIC", "order:create", 
            "订单1创建".getBytes()));
        
        producer.send(new Message("ORDER_TOPIC", "order:pay", 
            "订单2支付".getBytes()));
        
        producer.send(new Message("ORDER_TOPIC", "order:ship", 
            "订单3发货".getBytes()));
    }
}

public class TagFilterConsumer {

    public void subscribeByTags() {
        // 单个 Tag
        consumer.subscribe("ORDER_TOPIC", "order:create");
        
        // 多个 Tag (OR)
        consumer.subscribe("ORDER_TOPIC", "order:create || order:pay");
        
        // 所有 Tag
        consumer.subscribe("ORDER_TOPIC", "*");
    }
}

/**
 * SQL92 过滤表达式
 */
public class SQLFilterExpression {

    // 比较运算符: =, !=, >, <, >=, <=
    // 逻辑运算符: AND, OR, NOT
    // IN, BETWEEN, LIKE, IS NULL

    // 示例: amount > 100 AND city = '北京'
}

// 消费者订阅
@RocketMQMessageListener(
    topic = "ORDER_TOPIC",
    selectorType = SelectorType.SQL92,
    selectorExpression = "amount > 100 AND level = 'vip'"
)
public class SQLFilterConsumer implements RocketMQListener<Order> {
}

# broker.conf
enablePropertyFilter = true
waitTimeMomentInPropertyFilterQueue = 5000
maxMsgsInBatch = 64