本文編寫，參考：https://my.oschina.net/bieber/blog/725646

producer Send()的Message最終將由broker處理，處理類為：SendMessageProcessor ,處理方法：processRequet.

public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {

    private List<ConsumeMessageHook> consumeMessageHookList;

    public SendMessageProcessor
 
(final BrokerController brokerController) {
        super(brokerController);
    }

    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {}
上述方法，並不是直接處理訊息，而是交由MessageStore處理，相關程式碼如下：

MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
msgInner.setTopic(requestHeader.getTopic());
msgInner.setQueueId(queueIdInt);
//......
PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner)
 
;

然而MessageStore也不直接持久化訊息，轉交給 CommitLog

long beginTime = this.getSystemClock().now();
PutMessageResult result = this.commitLog.putMessages(messageExtBatch);

從MappedFileQueue中取出最新的一條：

MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
 //寫訊息

result = mappedFile.appendMessages(messageExtBatch, this.appendMessageCallback);
 

  //持久化到磁碟,最終通過FileChannel持久化到檔案
handleDiskFlush(result, putMessageResult, messageExtBatch);

handleHA(result, putMessageResult, messageExtBatch);


2.cousumer 從broker讀訊息。消費者從broker讀取訊息經由PullMessageProcessor類處理的，processRequest()方法處理請求：

RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend)

經過一系列的判斷處理，之後交由 MessageStore：

final GetMessageResult getMessageResult =
    this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
        requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), messageFilter);
讀取訊息。
之後交由commitLog，讀出訊息，

SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);

可以看到是先從ConsumerQueue中獲取訊息索引，然後再從commitlog中讀取訊息內容。這些內容也是在儲存訊息的時候寫入的。