一. 前言

RocketMQ采用內存和磁盤存儲來存儲消息。那現在來分析一下消息存儲的流程

成都創新互聯公司主要從事網站設計、網站建設、網頁設計、企業做網站、公司建網站等業務。立足成都服務白堿灘,10余年網站建設經驗,價格優惠、服務專業,歡迎來電咨詢建站服務:028-86922220

二. 代碼流程

在Broker啟動的時候會拉起相關服務
流程如下：

流程圖引用網址
http://blog.csdn.net/akfly/article/details/53447000

三. 代碼流程

由于是Broker來存儲消息，那么消息入口的代碼應該是在Broker里面，而Broker的入口是BrokerStartup，以及重要的BrokerController。
具體流程可以參考Broker啟動源代碼分析。

Broker啟動流程

以發送消息為例

1. Broker啟動注冊發送消息處理器

Broker啟動的時候，會注冊一個SendMessageProcesser來響應netty的發送消息請求，如下：

public void registerProcessor() {
        /**
         * SendMessageProcessor
         */
        SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
        sendProcessor.registerSendMessageHook(sendMessageHookList);
        sendProcessor.registerConsumeMessageHook(consumeMessageHookList);

        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
}

2. 消息處理器處理發送者發送過來的消息

public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {

    @Override
    public RemotingCommand proce***equest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        SendMessageContext mqtraceContext;
                ...
        switch (request.getCode()) {
            response = this.sendMessage(ctx, request, mqtraceContext, requestHeader);
        }
    }
}       

繼續看sendMessage..

private RemotingCommand sendMessage(final ChannelHandlerContext ctx,
        ...
        PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
}               

調用MessageStore.putMessage(msgInner)

當前文章：52.源代碼解讀-RocketMQ消息寫入機制
文章起源：http://www.js-pz168.com/article48/pojohp.html

成都網站建設公司_創新互聯，為您提供動態網站、網站收錄、軟件開發、網站排名、微信公眾號、營銷型網站建設

廣告

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創新互聯