當前位置：首頁 > 科技 > 軟件

深入理解 RocketMQ 廣播消費

來源：責編：時間：2023-09-28 10:09:12 357觀看

導讀這篇文章我們聊聊廣播消費，因為廣播消費在某些場景下真的有奇效。筆者會從基礎概念、實現機制、實戰案例三個方面一一展開，希望能幫助到大家。1 基礎概念RocketMQ 支持兩種消息模式：集群消費（ Clustering ）和廣播消費（ Broa

這篇文章我們聊聊廣播消費，因為廣播消費在某些場景下真的有奇效。筆者會從基礎概念、實現機制、實戰案例三個方面一一展開，希望能幫助到大家。

1 基礎概念

RocketMQ 支持兩種消息模式：集群消費（ Clustering ）和廣播消費（ Broadcasting ）。

集群消費：

同一 Topic 下的一條消息只會被同一消費組中的一個消費者消費。也就是說，消息被負載均衡到了同一個消費組的多個消費者實例上。

圖片

廣播消費：

當使用廣播消費模式時，每條消息推送給集群內所有的消費者，保證消息至少被每個消費者消費一次。

圖片

2 源碼解析

首先下圖展示了廣播消費的代碼示例。

public class PushConsumer {    public static final String CONSUMER_GROUP = "myconsumerGroup";    public static final String DEFAULT_NAMESRVADDR = "localhost:9876";    public static final String TOPIC = "mytest";    public static final String SUB_EXPRESSION = "TagA || TagC || TagD";    public static void main(String[] args) throws InterruptedException, MQClientException {        // 定義 DefaultPushConsumer         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);        // 定義名字服務地址        consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);        // 定義消費讀取位點        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);        // 定義消費模式        consumer.setMessageModel(MessageModel.BROADCASTING);        // 訂閱主題信息        consumer.subscribe(TOPIC, SUB_EXPRESSION);        // 訂閱消息監聽器        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {            try {                for (MessageExt messageExt : msgs) {                    System.out.println(new String(messageExt.getBody()));                }            }catch (Exception e) {                e.printStackTrace();            }            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;        });        consumer.start();        System.out.printf("Broadcast Consumer Started.%n");    }}

和集群消費不同的點在于下面的代碼：

consumer.setMessageModel(MessageModel.BROADCASTING);

接下來，我們從源碼角度來看看廣播消費和集群消費有哪些差異點？

首先進入 DefaultMQPushConsumerImpl 類的 start 方法 , 分析啟動流程中他們兩者的差異點：

圖片

▍ 差異點1：拷貝訂閱關系

private void copySubscription() throws MQClientException {    try {       Map<String, String> sub = this.defaultMQPushConsumer.getSubscription();       if (sub != null) {          for (final Map.Entry<String, String> entry : sub.entrySet()) {              final String topic = entry.getKey();              final String subString = entry.getValue();              SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(topic, subString);                this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);            }        }       if (null == this.messageListenerInner) {          this.messageListenerInner = this.defaultMQPushConsumer.getMessageListener();       }       // 注意下面的代碼 , 集群模式下自動訂閱重試主題        switch (this.defaultMQPushConsumer.getMessageModel()) {           case BROADCASTING:               break;           case CLUSTERING:                final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());                SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(retryTopic, SubscriptionData.SUB_ALL);                this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);                break;            default:                break;        }    } catch (Exception e) {        throw new MQClientException("subscription exception", e);    }}

在集群模式下，會自動訂閱重試隊列，而廣播模式下，并沒有這段代碼。也就是說廣播模式下，不支持消息重試。

▍ 差異點2：本地進度存儲

switch (this.defaultMQPushConsumer.getMessageModel()) {    case BROADCASTING:        this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());        break;    case CLUSTERING:        this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());        break;    default:        break;}this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);

我們可以看到消費進度存儲的對象是：LocalFileOffsetStore , 進度文件存儲在如下的主目錄 /{用戶主目錄}/.rocketmq_offsets。

public final static String LOCAL_OFFSET_STORE_DIR = System.getProperty(    "rocketmq.client.localOffsetStoreDir",    System.getProperty("user.home") + File.separator + ".rocketmq_offsets");

進度文件是 /mqClientId/{consumerGroupName}/offsets.json 。

this.storePath = LOCAL_OFFSET_STORE_DIR + File.separator + this.mQClientFactory.getClientId() + File.separator + this.groupName + File.separator + "offsets.json";

筆者創建了一個主題 mytest , 包含4個隊列，進度文件內容如下：

圖片

消費者啟動后，我們可以將整個流程簡化如下圖，并繼續整理差異點：

圖片

▍ 差異點3：負載均衡消費該主題的所有 MessageQueue

進入負載均衡抽象類 RebalanceImpl 的rebalanceByTopic方法。

private void rebalanceByTopic(final String topic, final boolean isOrder) {    switch (messageModel) {        case BROADCASTING: {            Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);            if (mqSet != null) {                boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);                // 省略代碼            } else {                log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);            }            break;        }        case CLUSTERING: {            Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);            List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);            // 省略代碼            if (mqSet != null && cidAll != null) {                List<MessageQueue> mqAll = new ArrayList<MessageQueue>();                mqAll.addAll(mqSet);                Collections.sort(mqAll);                Collections.sort(cidAll);                AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;                List<MessageQueue> allocateResult = null;                try {                     allocateResult = strategy.allocate(                            this.consumerGroup,                            this.mQClientFactory.getClientId(),                            mqAll,                            cidAll);                    } catch (Throwable e) {                        // 省略日志打印代碼                        return;                    }                Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();                if (allocateResult != null) {                    allocateResultSet.addAll(allocateResult);                }                boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);                //省略代碼            }            break;        }        default:            break;    }}

從上面代碼我們可以看到消息模式為廣播消費模式時，消費者會消費該主題下所有的隊列，這一點也可以從本地的進度文件 offsets.json 得到印證。

▍ 差異點4：不支持順序消息

我們知道消費消息順序服務會向 Borker 申請鎖。消費者根據分配的隊列 messageQueue ，向 Borker 申請鎖，如果申請成功，則會拉取消息，如果失敗，則定時任務每隔 20 秒會重新嘗試。

if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {        @Override        public void run() {            try {                ConsumeMessageOrderlyService.this.lockMQPeriodically();            } catch (Throwable e) {                log.error("scheduleAtFixedRate lockMQPeriodically exception", e);            }        }    }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);}

但是從上面的代碼，我們發現只有在集群消費的時候才會定時申請鎖，這樣就會導致廣播消費時，無法為負載均衡的隊列申請鎖，導致拉取消息服務一直無法獲取消息數據。

筆者修改消費例子，在消息模式為廣播模式的場景下，將消費模式從并發消費修改為順序消費。

consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {    try {        for (MessageExt messageExt : msgs) {            System.out.println(new String(messageExt.getBody()));        }    }catch (Exception e) {        e.printStackTrace();    }    return ConsumeOrderlyStatus.SUCCESS;});

圖片

推送服務是一個 TCP 服務（自定義協議），同時也是一個消費者服務，消息模式是廣播消費。

司機打開司機端 APP 后，APP 會通過負載均衡和推送服務創建長連接，推送服務會保存 TCP 連接引用（比如司機編號和 TCP channel 的引用）。

派單服務是生產者，將派單數據發送到 MetaQ , 每個推送服務都會消費到該消息，推送服務判斷本地內存中是否存在該司機的 TCP channel ，若存在，則通過 TCP 連接將數據推送給司機端。

肯定有同學會問：假如網絡原因，推送失敗怎么處理？有兩個要點：

司機端 APP 定時主動拉取派單信息；
當推送服務沒有收到司機端的 ACK 時，也會一定時限內再次推送，達到閾值后，不再推送。

3.2 緩存同步

高并發場景下，很多應用使用本地緩存，提升系統性能。

本地緩存可以是 HashMap 、ConcurrentHashMap ，也可以是緩存框架 Guava Cache 或者 Caffeine cache 。

圖片

如上圖，應用A啟動后，作為一個 RocketMQ 消費者，消息模式設置為廣播消費。為了提升接口性能，每個應用節點都會將字典表加載到本地緩存里。

當字典表數據變更時，可以通過業務系統發送一條消息到 RocketMQ ，每個應用節點都會消費消息，刷新本地緩存。

4 總結

集群消費和廣播消費模式下，各功能的支持情況如下：

功能	集群消費	廣播消費
順序消息	支持	不支持
重置消費位點	支持	不支持
消息重試	支持	不支持
消費進度	服務端維護	客戶端維護

廣播消費主要用于兩種場景：消息推送和緩存同步。

參考資料：

https://www.51cto.com/article/714277.html
https://ost.51cto.com/posts/21100

本文鏈接：http://www.www897cc.com/showinfo-26-11888-0.html深入理解 RocketMQ 廣播消費

聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com

上一篇：消息隊列技術選型：這七種消息場景一定要考慮！

下一篇：從零開發可視化大屏制作平臺

標簽：

熱門焦點

六大權益！華為8月服務日開啟：手機免費貼膜、維修免人工費

8月5日消息，一年一度的華為開發者大會2023（Together）日前在松山湖拉開帷幕，與此同時，華為8月服務日也式開啟，到店可享六大專屬權益。華為用戶可在華為商城Ap
十個可以手動編寫的 JavaScript 數組 API

JavaScript 中有很多API，使用得當，會很方便，省力不少。你知道它的原理嗎? 今天這篇文章，我們將對它們進行一次小總結。現在開始吧。1.forEach()forEach()用于遍歷數組接收一參
一年經驗在二線城市面試后端的經驗分享

忠告這篇文章只適合2年內工作經驗、甚至沒有工作經驗的朋友閱讀。如果你是2年以上工作經驗，請果斷劃走，對你沒啥幫助~主人公這篇文章內容來自「升職加薪」星球星友的投稿，坐
在線圖片編輯器，支持PSD解析、AI摳圖等

自從我上次分享一個人開發仿造稿定設計的圖片編輯器到現在，不知不覺已過去一年時間了，期間我經歷了裁員失業、面試找工作碰壁，寒冬下一直沒有很好地履行計劃.....這些就放在日
“又被陳思誠騙了”

作者｜張思齊出品｜眾面（ID：ZhongMian_ZM）如今的國產懸疑電影，成了陳思誠的天下。最近大爆電影《消失的她》票房突破30億斷層奪魁暑期檔，陳思誠再度風頭無兩。你可以說陳思誠的
7月4日見！iQOO 11S官宣：“雞血版”驍龍8 Gen2+200W快充加持

上半年已接近尾聲，截至目前各大品牌旗下的頂級旗艦都已悉數亮相，而下半年即將推出的頂級旗艦已經成為了數碼圈爆料的主流，其中就包括全新的iQOO 11S系
由于成本持續增加，筆記本產品價格預計將明顯上漲

根據知情人士透露，由于材料、物流等成本持續增加，筆記本產品價格預計將在2021年下半年有明顯上漲。進入6月下旬以來，全球半導體芯片缺貨情況加劇，顯卡、處理器
外交部：美方應停止在網絡安全問題上不負責任地指責他國

　中國外交部今天（16日）舉行例行記者會。會上，有記者問，美國情報官員稱，他們正在阻攔來自中國以及其他國家的黑客獲取相關科研成果。中方對此有何評論？對此

日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

深入理解 RocketMQ 廣播消費

1 基礎概念

2 源碼解析

3.2 緩存同步

4 總結

六大權益！華為8月服務日開啟：手機免費貼膜、維修免人工費

十個可以手動編寫的 JavaScript 數組 API

一年經驗在二線城市面試后端的經驗分享

在線圖片編輯器，支持PSD解析、AI摳圖等

重估百度丨“晚熟”的百度云，能等到春天嗎？

“又被陳思誠騙了”

7月4日見！iQOO 11S官宣：“雞血版”驍龍8 Gen2+200W快充加持

由于成本持續增加，筆記本產品價格預計將明顯上漲

外交部：美方應停止在網絡安全問題上不負責任地指責他國

最新推薦

猜你喜歡

熱門推薦

相關資訊