Spring?Boot?整合RocketMq實現(xiàn)消息過濾功能

更新時間：2022年06月07日 14:29:43 作者：劍圣無痕

這篇文章主要介紹了Spring?Boot?整合RocketMq實現(xiàn)消息過濾,本文講解了RocketMQ實現(xiàn)消息過濾，針對不同的業(yè)務場景選擇合適的方案即可,需要的朋友可以參考下

簡介

消息過濾是指消費者一端在消費消息時，對消息進行選擇性過濾，只消費符合過濾條件的消息。 RocketMQ的消息過濾機制大致分為兩種：標簽過濾和類過濾。其中標簽過濾又分為Tag過濾和SQL92過濾。

根據(jù)TAG過濾消息

消息發(fā)送端只能設置一個tag，消息接收端可以設置多個tag。

生產(chǎn)者

 public void sendTagMessage()
   {
       String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
       for(int i=0;i<10;i++)
       {
           String tag = tags[i % tags.length];
           logger.info("sendTagMessage tag is :{}",tag);
           String msg = "hello, 這是第" + (i + 1) + "條消息";
           org.springframework.messaging.Message<String> msg1 = MessageBuilder.withPayload(msg).build(); 
           rocketMQTemplate.convertAndSend("test-tag-rocketmq" + ":" + tag, msg1);
       }
   }

說明：示例中循環(huán)發(fā)送了10條消息，每條消息設置了一個tag發(fā)送過濾消息的格式為：topic:tag的形式，注意發(fā)送端只能設定一個tag。

消費者

@Component
@RocketMQMessageListener(consumerGroup="test-tagrocketmq-group",topic="test-tag-rocketmq",selectorExpression="TagA || TagC || TagD",selectorType=SelectorType.TAG, messageModel = MessageModel.CLUSTERING)
public class TagConsumer implements RocketMQListener<Object>
{
    private Logger logger =LoggerFactory.getLogger(getClass());
    @Override
    public void onMessage(Object o)
    {
        String msg=JSON.toJSONString(o);
        logger.info("send TagA || TagC || TagD  succss content is:{}", msg);
    }
}

說明：

selectorType：指定消息通過的tag的方式，默認為SelectorType.TAG
messageModel：指定消息的消費模式，默認為MessageModel.CLUSTERING模式每條消息只能由一個消費者消費，而MessageModel.BROADCASTING模式為廣播模式，所有訂閱者都能消費。
selectorExpression :指定那些Tag消息能夠被消費，多個采用||分割。

測試結果

從結果我可以看出第2條為TAGC、第7條為TAGC、第8條為TAGD，第3條為TAGD，第5條為TAGA，第0條為TAGA，而消費端監(jiān)聽的TAG為TAGA、TAGC、TAGD所以對于不符合條件的消息進行了過濾。

根據(jù)SQL表達式過濾消息

SQL表達式方式可以根據(jù)發(fā)送消息時輸入的屬性進行一些計算。

RocketMQ的SQL表達式語法只定義了一些基本的語法功能。

數(shù)字比較，如>，>=，<，<=，BETWEEN，=；
字符比較，如：=，<>，IN；IS NULL or IS NOT NULL;
邏輯運算符：AND, OR, NOT;
常量類型:
數(shù)值，如：123, 3.1415；
字符, 如：‘abc’, 必須使用單引號;
NULL，特殊常量
Boolean, TRUE or FALSE;

生產(chǎn)者

   public void sendSQLMessage()
   {
       String msg = "hello, 這是第1條消息";
       org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(msg).build() ;
       Map<String, Object> headers = new HashMap<>() ;
       headers.put("i", 5) ;
       rocketMQTemplate.convertAndSend("test-sql-rocketmq", message, headers);
   }

說明：傳遞了參數(shù)為5進行條件判斷。

消費者

@Component
@RocketMQMessageListener(consumerGroup="test-sqlrocketmq-group",topic="test-sql-rocketmq",selectorExpression = "i=5",selectorType=SelectorType.SQL92, messageModel = MessageModel.CLUSTERING)
public class SQLConsumer implements RocketMQListener<MessageExt>
{
    private Logger logger =LoggerFactory.getLogger(getClass());
    @Override
    public void onMessage(MessageExt message)
    {
        String msg=new String(message.getBody());
        String paramStr=JSON.toJSONString(message.getProperties());
        //消息內容
        logger.info("send succss content is:{}", msg);
        //消息參數(shù)
        logger.info("send mssage parma is:{}", paramStr);
    }
}

說明：

selectorType：指定消息通過的tag的方式，默認為SelectorType.CLUSTERING
messageModel：指定消息的消費模式，默認為MessageModel.CLUSTERING模式每條消息只能由一個消費者消費，而MessageModel.BROADCASTING模式為廣播模式，所有訂閱者都能消費。
selectorExpression : 采用rocketMQ支持的表達式。例如i=5