RocketMQ源碼閱讀(七)-broker

broker是消息隊列的核心組建,承載了消息接收,存儲和轉(zhuǎn)發(fā)的職責(zé). 因此, broker需要具備各種基本功能和高階功能.
1.基本功能

  • 承載消息堆積的能力
    消息到達服務(wù)端如果不經(jīng)過任何處理就到接收者了, broker就失去了它的意義. 為了滿足我們錯峰/流控/最終可達等一系列需求, 把消息存儲下來, 然后選擇時機投遞就顯得是順理成章的了. 因此, broker必須具備強大的消息堆積能力.
  • 消費關(guān)系解偶
    其實就是具備單播(點對點)和廣播(一點對多點)功能.
    2.高階功能
  • 消息去重
    當(dāng)前的消息隊列還無法做到消息完全去重(除非允許消息丟失), 只能盡量減少消息重復(fù)的概率.
  • 順序消息
    消息有序指的是一類消息消費時, 能按照發(fā)送的順序來消費. 例如: 一個訂單產(chǎn)生了3條消息, 分別是訂單創(chuàng)建, 訂單付款, 訂單完成. 消費時, 要按照這個順序消費才能有意義. 但是同時訂單之間是可以并行消費的. RocketMQ可以嚴(yán)格的保證消息有序.

我們從三個方面分析broker, 分別為broker的啟動, 接收消息和消費消息.

broker的啟動

broker的啟動過程是一個比較復(fù)雜的過程, 其中涉及到很多broker模塊自身的初始化, 例如：路由信息設(shè)置, 消息落地, 消息推送, 以及和namesvr模塊通信部分的初始化.

broker啟動可以分為兩個步驟, initialize和start.

initialize

1.當(dāng)broker啟動命令得到外部傳入的初始化參數(shù)以后, 將參數(shù)注入對應(yīng)的config類當(dāng)中, 這些config類包括:

  • broker自身的配置：包括根目錄, namesrv地址, broker的IP和名稱, 消息隊列數(shù), 收發(fā)消息線程池數(shù)等參數(shù)
  • netty啟動配置：包括netty監(jiān)聽端口, 工作線程數(shù), 異步發(fā)送消息信號量數(shù)量等網(wǎng)絡(luò)配置等參數(shù).
  • 存儲層配置：包括存儲跟目錄, CommitLog配置, 持久化策略配置等參數(shù).
    這一步驟的具體代碼在BrokerStartup的start方法和createBrokerController方法中, 較為簡單,此處不貼出.

2.當(dāng)配置信息設(shè)置完畢后, broker會將這些參數(shù)傳入borkerController控制器當(dāng)中, 這個控制器會初始加載很多的管理器, 如下:

  • topicManager：用于管理broker中存儲的所有topic的配置
  • consumerOffsetManager：管理Consumer的消費進度
  • subscriptionGroupManager：用來管理訂閱組, 包括訂閱權(quán)限等
  • messageStore：用于broker層的消息落地存儲.

這部分代碼在BrokerController.initialize中, 代碼片段如下:

public boolean initialize() throws CloneNotSupportedException {
    boolean result = true;
    //加載topicConfigManager
    result = result && this.topicConfigManager.load();
    //加載consumerOffsetManager
    result = result && this.consumerOffsetManager.load();
    //加載subscriptionGroupManager
    result = result && this.subscriptionGroupManager.load();
    //加載messageStore
    if (result) {
        try {
            this.messageStore =
                new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
                    this.brokerConfig);
            this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);
            //load plugin
            MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);
            this.messageStore = MessageStoreFactory.build(context, this.messageStore);
        } catch (IOException e) {
            result = false;
            e.printStackTrace();
        }
    }

    result = result && this.messageStore.load();
    ...
}

下面看一下this.topicConfigManager.load()的代碼:

public boolean load() {
    String fileName = null;
    try {
        //讀取文件路徑
        fileName = this.configFilePath();
        //讀取文件內(nèi)容
        String jsonString = MixAll.file2String(fileName);
        if (null == jsonString || jsonString.length() == 0) {
            //讀取備份文件
            return this.loadBak();
        } else {
            //將文件內(nèi)容讀入到topicConfigManager.topicConfigTable中
            this.decode(jsonString);
            PLOG.info("load {} OK", fileName);
            return true;
        }
    } catch (Exception e) {
        PLOG.error("load " + fileName + " Failed, and try to load backup file", e);
        return this.loadBak();
    }
}

consumerOffsetManager和subscriptionGroupManager的加載類似, 三個類都繼承自ConfigManager.
接下去就是messageStore的加載, 用于消息的持久化, 關(guān)于這一部分, 可以參考前文.
3.當(dāng)上述的管理器全部加載完成以后, 控制器將開始進入下一步的初始化：

  • 啟動netty服務(wù)端, 包括處理消息的remotingServer和主從同步使用的fastRemotingServer.
  • 初始化多個線程池, 包括：sendMessageExecutor(處理發(fā)送消息), pullMessageExecutor(處理拉取消息), adminBrokerExecutor(管理Broker), clientManageExecutor(管理client), consumerManageExecutor(管理消費者)
  • 將上述的線程池注冊到netty消息處理器當(dāng)中
  • 啟動定時調(diào)度線程來做很多事情, 包括：一天上報一次broker的所有狀態(tài), 10秒持久化一次Consumer消費進度, 60分鐘清理一次無效的topic訂閱信息, 60秒獲取一次namesrv的地址信息

代碼如下:

public boolean initialize() throws CloneNotSupportedException {
    ......
    if (result) {
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
        NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
        fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
        this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);
        this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
            this.brokerConfig.getSendMessageThreadPoolNums(),
            this.brokerConfig.getSendMessageThreadPoolNums(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.sendThreadPoolQueue,
            new ThreadFactoryImpl("SendMessageThread_"));

        this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
            this.brokerConfig.getPullMessageThreadPoolNums(),
            this.brokerConfig.getPullMessageThreadPoolNums(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.pullThreadPoolQueue,
            new ThreadFactoryImpl("PullMessageThread_"));

        this.adminBrokerExecutor =
            Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
                "AdminBrokerThread_"));

        this.clientManageExecutor = new ThreadPoolExecutor(
            this.brokerConfig.getClientManageThreadPoolNums(),
            this.brokerConfig.getClientManageThreadPoolNums(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.clientManagerThreadPoolQueue,
            new ThreadFactoryImpl("ClientManageThread_"));

        this.consumerManageExecutor =
            Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
                "ConsumerManageThread_"));

        this.registerProcessor();

        // TODO remove in future
        final long initialDelay = UtilAll.computNextMorningTimeMillis() - System.currentTimeMillis();
        final long period = 1000 * 60 * 60 * 24;
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    BrokerController.this.getBrokerStats().record();
                } catch (Throwable e) {
                    log.error("schedule record error.", e);
                }
            }
        }, initialDelay, period, TimeUnit.MILLISECONDS);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    BrokerController.this.consumerOffsetManager.persist();
                } catch (Throwable e) {
                    log.error("schedule persist consumerOffset error.", e);
                }
            }
        }, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    //清理出問題的consumer
                    BrokerController.this.protectBroker();
                } catch (Exception e) {
                    log.error("protectBroker error.", e);
                }
            }
        }, 3, 3, TimeUnit.MINUTES);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    //打印生產(chǎn), 消費隊列信息
                    BrokerController.this.printWaterMark();
                } catch (Exception e) {
                    log.error("printWaterMark error.", e);
                }
            }
        }, 10, 1, TimeUnit.SECONDS);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                try {
                    log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes());
                } catch (Throwable e) {
                    log.error("schedule dispatchBehindBytes error.", e);
                }
            }
        }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);

        //更新namesrv的地址信息
        if (this.brokerConfig.getNamesrvAddr() != null) {
            this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
            log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
        } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    try {
                        BrokerController.this.brokerOuterAPI.fetchNameServerAddr();
                    } catch (Throwable e) {
                        log.error("ScheduledTask fetchNameServerAddr exception", e);
                    }
                }
            }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
        }

        //slave更新master地址, 從master同步消息數(shù)據(jù)
        if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
            if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {
                this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
                this.updateMasterHAServerAddrPeriodically = false;
            } else {
                this.updateMasterHAServerAddrPeriodically = true;
            }

            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    try {
                        BrokerController.this.slaveSynchronize.syncAll();
                    } catch (Throwable e) {
                        log.error("ScheduledTask syncAll slave exception", e);
                    }
                }
            }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
        } else {
            //master打印slave落后的信息
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    try {
                        BrokerController.this.printMasterAndSlaveDiff();
                    } catch (Throwable e) {
                        log.error("schedule printMasterAndSlaveDiff error.", e);
                    }
                }
            }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
        }
    }
    return result;
}

上述的全部過程中, 比較關(guān)鍵的操作是: 啟動netty服務(wù)端, 注冊多個消息處理器, 初始化線程池用來做消息的并發(fā)處理.

start

當(dāng)broker初始化了配置參數(shù)以后, 就可以開始啟動部分了, 啟動的部分會簡單一些：

  • 啟動剛剛初始化的各個管理器：topicManager, consumerOffsetManager, subscriptionGroupManager, messageStore
  • 開啟定時調(diào)度線程30秒向namesrv不斷上報自己的信息
public void start() throws Exception {
    if (this.messageStore != null) {
        this.messageStore.start();
    }

    if (this.remotingServer != null) {
        this.remotingServer.start();
    }

    if (this.fastRemotingServer != null) {
        this.fastRemotingServer.start();
    }

    if (this.brokerOuterAPI != null) {
        this.brokerOuterAPI.start();
    }

    if (this.pullRequestHoldService != null) {
        this.pullRequestHoldService.start();
    }

    if (this.clientHousekeepingService != null) {
        this.clientHousekeepingService.start();
    }

    if (this.filterServerManager != null) {
        this.filterServerManager.start();
    }

    this.registerBrokerAll(true, false);

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                //向nameserver上報信息
                BrokerController.this.registerBrokerAll(true, false);
            } catch (Throwable e) {
                log.error("registerBrokerAll Exception", e);
            }
        }
    }, 1000 * 10, 1000 * 30, TimeUnit.MILLISECONDS);

    if (this.brokerStatsManager != null) {
        this.brokerStatsManager.start();
    }

    if (this.brokerFastFailure != null) {
        this.brokerFastFailure.start();
    }
}

后文介紹消息的接收.

參考資料
1.rocketMQ的broker模塊

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容