消息过滤

RocketMQ支持在消费者端对消息进行过滤。
包括:

SQL过滤:使用SQL92过滤消息内容
Tag过滤:根据Tag来过滤消息
Regex过滤:使用正则表达式过滤消息
-etc

政策类

RocketMQ定义了MessageFilter接口:

public interface MessageFilter {

  boolean isMatched(Message msg);

}

具体策略

提供不同的实现:

public class SQL92Filter implements MessageFilter{  

  // 使用SQL来实现isMatched方法  
}

public class TagFilter implements MessageFilter{

  // 根据消息Tag实现isMatched 
} 

public class RegexFilter implements MessageFilter{

  // 使用正则实现isMatched方法  
}

使用策略

注册过滤器到消费者中:

consumer.registerMessageFilter(new SQL92Filter());

Message msg = consumer.receive();

// 会使用注册的过滤器 isMatched 来决定接收消息

也可以动态注册:

MessageFilter filter = getMessageFilter(); 

consumer.registerMessageFilter(filter);

优点

通过策略模式实现消息过滤可以:

根据需求使用不同的过滤策略
新添加过滤策略不影响已有代码
客户端代码与具体策略解耦

符合策略模式的原则:

定义一个策略接口
多个具体策略实现
用户不依赖具体的策略实现。

总的来说,RocketMQ利用策略模式实现了不同的消息过滤策略。客户端代码与具体策略解耦。