新聞中心
在 RocketMQ 4.9.6 版本中,filtersrv 被移除了,這一變化是出于以下幾個(gè)目的和考慮:

我們提供的服務(wù)有:做網(wǎng)站、成都網(wǎng)站建設(shè)、微信公眾號開發(fā)、網(wǎng)站優(yōu)化、網(wǎng)站認(rèn)證、康縣ssl等。為1000多家企事業(yè)單位解決了網(wǎng)站和推廣的問題。提供周到的售前咨詢和貼心的售后服務(wù),是有科學(xué)管理、有技術(shù)的康縣網(wǎng)站制作公司
1、架構(gòu)優(yōu)化:RocketMQ 作為一個(gè)高性能、分布式的消息中間件,其架構(gòu)設(shè)計(jì)和實(shí)現(xiàn)一直在不斷演進(jìn)和優(yōu)化。filtersrv 的移除是其中一部分,旨在簡化架構(gòu),提高系統(tǒng)的可維護(hù)性和擴(kuò)展性。
2、功能整合:filtersrv 主要用于消息過濾,但在新版本中,這部分功能被整合到了其他組件中,Broker 或 NameServer,這樣的整合有助于減少組件數(shù)量,降低系統(tǒng)復(fù)雜性。
3、性能提升:通過移除 filtersrv,可以減少消息傳遞的環(huán)節(jié),從而降低延遲,提高整體性能。
4、資源節(jié)省:filtersrv 作為額外的服務(wù)進(jìn)程,會(huì)占用一定的系統(tǒng)資源,移除后,可以節(jié)省這部分資源,降低部署和維護(hù)成本。
5、易用性增強(qiáng):對于用戶來說,filtersrv 的存在可能增加了部署和維護(hù)的復(fù)雜性,移除后,用戶可以更加便捷地進(jìn)行部署和使用。
接下來,我們將詳細(xì)探討如何在移除 filtersrv 后,使用 RocketMQ 進(jìn)行消息過濾。
消息過濾機(jī)制
在 RocketMQ 中,消息過濾主要通過兩種方式實(shí)現(xiàn):
1、消費(fèi)端過濾:消費(fèi)者在訂閱主題時(shí),可以通過設(shè)置過濾條件(如 SQL 表達(dá)式),來選擇性地接收消息,這種方式適用于消費(fèi)者對特定類型或?qū)傩缘南⒏信d趣,而不想處理全部消息的場景。
2、生產(chǎn)者端過濾:生產(chǎn)者在發(fā)送消息時(shí),可以設(shè)置消息的屬性(如鍵值對),這些屬性可以被用于后續(xù)的消息篩選和處理。
示例代碼
以下是一個(gè)簡單的示例,展示了如何在移除 filtersrv 后,使用 RocketMQ 進(jìn)行消息過濾。
生產(chǎn)者端
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
public class Producer {
public static void main(String[] args) throws Exception {
// 創(chuàng)建生產(chǎn)者實(shí)例
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 設(shè)置 NameServer 地址
producer.setNamesrvAddr("127.0.0.1:9876");
// 啟動(dòng)生產(chǎn)者
producer.start();
// 創(chuàng)建消息實(shí)例
Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
// 設(shè)置消息屬性
msg.setKeys("KeyA");
// 發(fā)送消息
producer.send(msg);
// 關(guān)閉生產(chǎn)者
producer.shutdown();
}
}
消費(fèi)者端
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.message.MessageExt;
import java.util.List;
public class Consumer {
public static void main(String[] args) throws Exception {
// 創(chuàng)建消費(fèi)者實(shí)例
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
// 設(shè)置 NameServer 地址
consumer.setNamesrvAddr("127.0.0.1:9876");
// 訂閱主題
consumer.subscribe("TopicTest", "*");
// 注冊消息監(jiān)聽器
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 啟動(dòng)消費(fèi)者
consumer.start();
System.out.printf("Consumer Started.%n");
}
}
在上述示例中,生產(chǎn)者發(fā)送了一條帶有屬性 KeyA 的消息,消費(fèi)者可以基于這些屬性進(jìn)行過濾,只消費(fèi)滿足特定條件的消息。
標(biāo)題名稱:rocketmq4.9.6版本把filtersrv刪除是出于什么目的呢?
本文鏈接:http://fisionsoft.com.cn/article/copjiop.html


咨詢
建站咨詢
