fifty-two。 Source code interpretation-RocketMQ message writing mechanism

I. preface

RocketMQ uses memory and disk storage to store messages. Now let's analyze the flow of message storage

II. code flow

Related services are pulled up when Broker starts

The process is as follows:

Flowchart reference URL

http://blog.csdn.net/akfly/article/details/53447000

III. code flow

Since Broker stores messages, the code of message entry should be in Broker, and Broker entry is BrokerStartup, and important BrokerController.

Refer to Broker to start source code analysis for the specific process.

Broker Start Process

Take sending a message as an example

1. Broker Start Register Send Message Processor

When Broker starts, it registers a SendMessageProcessor to respond to netty's send message request, as follows:

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. Message Processor processes messages sent by sender 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); } }}

Continue reading sendMessage..

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

Call MessageStore.putMessage(msgInner)