您好,登錄后才能下訂單哦!
RocketMQ采用內(nèi)存和磁盤存儲來存儲消息。那現(xiàn)在來分析一下消息存儲的流程
在Broker啟動的時候會拉起相關(guān)服務(wù)
流程如下:
流程圖引用網(wǎng)址
http://blog.csdn.net/akfly/article/details/53447000
由于是Broker來存儲消息,那么消息入口的代碼應(yīng)該是在Broker里面,而Broker的入口是BrokerStartup,以及重要的BrokerController。
具體流程可以參考Broker啟動源代碼分析。
Broker啟動流程
以發(fā)送消息為例
Broker啟動的時候,會注冊一個SendMessageProcesser來響應(yīng)netty的發(fā)送消息請求,如下:
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);
}
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);
}
}
}
繼續(xù)看sendMessage..
private RemotingCommand sendMessage(final ChannelHandlerContext ctx,
...
PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
}
調(diào)用MessageStore.putMessage(msgInner)
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。