broker消息接收，假设接收的是一个普通消息(即没有事务)，此处分析也只分析master上动作逻辑，不涉及ha。
1. 如何找到消息接收处理入口
可以通过broker的监听端口10911顺藤摸瓜式的找到 NettyClientConfig.setListenPort-->BrokerStartup-->BrokerController-->NettyRemotingServer
com.alibaba.rocketmq.remoting.netty.NettyDecoder com.alibaba.rocketmq.remoting.netty.NettyRemotingServer.NettyServerHandler.channelRead0(ChannelHandlerContext, RemotingCommand)
也可以顺着broker的启动脚本找到
BrokerStartup-->BrokerController-->NettyRemotingServer
因为rocketmq连接层默认使用netty开发，如果熟悉netty的话，
可以直接查找ChannelInitializer或者pipeline().addLast等
2. 调试NettyDecoder
com.alibaba.rocketmq.remoting.netty.NettyDecoder.decode(ChannelHandlerContext, ByteBuf)
会调用RemotingCommand.decode(byteBuffer)组装RemotingCommand
在com.alibaba.rocketmq.remoting.protocol.RemotingCommand.decode(ByteBuffer) 中断点 排除如下code
cmd.getCode() != 0 && cmd.getCode() != 34 && cmd.getCode() != 15
&& cmd.getCode() != 38 && cmd.getCode() != 11
用producer的client发一条消息到broker
发现RemotingCommand.decode解析出来的RemotingCommand的code为310 310对应RequestCode.SEND_MESSAGE_V2 发送消息
此时处理线程是：Thread [NettyServerWorkerThread_2]
RemotingCommand主要描述了操作code、opaque号、body即消息体。
3. 调试NettyServerHandler
对com.alibaba.rocketmq.remoting.netty.NettyRemotingServer.NettyServerHandler.channelRead0(ChannelHandlerContext, RemotingCommand)断点
断点条件属性是msg.getCode() == 310
此时处理线程是：Thread [NettyServerWorkerThread_2]
4. NettyRemotingAbstract
com.alibaba.rocketmq.remoting.netty.NettyRemotingAbstract.processRequestCommand(ChannelHandlerContext, RemotingCommand)
方法负责分发给相应的NettyRequestProcessor实例
负责处理的是com.alibaba.rocketmq.broker.processor.SendMessageProcessor@17fb0278
处理线程是：Thread [SendMessageThread_1]，跟上面的处理已经不在一个线程上了，异步
5. SendMessageProcessor
SendMessageProcessor.processRequest(ChannelHandlerContext, RemotingCommand)介入处理
构建SendMessageRequestHeader
进入SendMessageProcessor.sendMessage(ChannelHandlerContext, RemotingCommand, SendMessageContext, SendMessageRequestHeader)
int queueIdInt = requestHeader.getQueueId(); 作用是什么 什么时候确定的id
构建MessageExtBrokerInner实例 tags转tagsCode 就是tags的String的hashcode
PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
交给MessageStore进行put(store,进行存放)
6. DefaultMessageStore
com.alibaba.rocketmq.store.DefaultMessageStore.putMessage(MessageExtBrokerInner) 消息topic的长度校验 不能超过127
消息properties转成字符串后，长度不能超过32767
下面交个CommitLog进行putMessage
7. CommitLog
com.alibaba.rocketmq.store.CommitLog.putMessage(MessageExtBrokerInner)
设置存储时间
设置消息体CRC校验值
交给MapedFile进行存储mapedFile.appendMessage(msg, this.appendMessageCallback)
8. MapedFile
mapedFile.appendMessage(msg, this.appendMessageCallback)
获取当前文件已经写到什么位置了
文件已经做了map map到mappedByteBuffer 讲当前的mappedByteBuffer 割出来并设置position为当前写位置
交由callback做append
callback是CommitLog$DefaultAppendMessageCallback
传给callback的参数有
- 1. 这个文件的offset，意思是：一个broker要存储很多消息，那么一个文件肯定不够存，当存到第二个文件的时候，这个文件里的第一条消息相对于整个broker中的消息有个offset，此值是文件名。
- 2. 从store文件映射的buffer中割出来的byteBuffer
- 3. 文件的剩余空间
- 4. 消息体
8.1 callback中append逻辑：
计算wroteOffset 即整个broker中的offset， 代码注释中称之为物理offset，计算逻辑也很简单即上面参数1+这条消息在这个文件中的position
计算消息id，逻辑使用主机物理IP加上一步的wroteOffset计算，用了ByteBuffer处理，比较高效，待完善验证用例。
获取这个topic的这个队列的offset(一个topic写多个队列)，这个offset有啥用？估计是用于快速查找
将消息bean对象用msgStoreItemMemory进行以约定消息体的形式进行byte的put
将msgStoreItemMemory put到上面传进来的参数2的buffer 即写入磁盘
构建append结果，没啥逻辑，主要是一些信息 AppendMessageResult
将消息在这个队列中的offset加1
更新MapedFile的wrotePostion，就是将原来的wrotePostion加上这次写入的字节数
此时订阅断已经收到消息，（应该更早，待确认，待分析消息如何送到consume的）
8.2 回CommitLog--：
构建 PutMessageResult对象。
8.3 处理同异步刷盘逻辑
同步刷盘 用GroupCommitService， 其中使用了多条消息一起刷的设计，并且设计了刷盘如果超时的异常场景的反馈
异步刷盘 用CommitLog$FlushRealTimeService（是一个线程，此时会叫醒他）
8.4 处理主从双机同异步同步逻辑
同步形式同步从节点 交由HAService 实时同步，并等待同步结果。
异步不用管。
9. 收集统计数据
耗时等等