写点什么

RocketMQ 源码 -broker 消息接收流程 (写入 commitLog)

作者:小小怪下士
  • 2023-03-25
    湖南
  • 本文字数:17174 字

    阅读完需:约 56 分钟

从本文开始,我们来分析 rocketMq 消息接收、分发以及投递流程。


RocketMq 消息处理整个流程如下:



  1. 消息接收:消息接收是指接收 producer 的消息,处理类是 SendMessageProcessor,将消息写入到 commigLog 文件后,接收流程处理完毕;

  2. 消息分发:broker 处理消息分发的类是 ReputMessageService,它会启动一个线程,不断地将 commitLong 分到到对应的 consumerQueue,这一步操作会写两个文件:consumerQueue 与 indexFile,写入后,消息分发流程处理 完毕;

  3. 消息投递:消息投递是指将消息发往 consumer 的流程,consumer 会发起获取消息的请求,broker 收到请求后,调用 PullMessageProcessor 类处理,从 consumerQueue 文件获取消息,返回给 consumer 后,投递流程处理完毕。


以上就是 rocketMq 处理消息的流程了,接下来我们就从源码来看相关流程的实现。

1. remotingServer 的启动流程

在正式分析接收与投递流程前,我们来了解下 remotingServer 的启动。


remotingServer 是一个 netty 服务,他开启了一个端口用来处理 producer 与 consumer 的网络请求。


remotingServer 是在 BrokerController#start 中启动的,代码如下:


    public void start() throws Exception {        // 启动各组件        ...
if (this.remotingServer != null) { this.remotingServer.start(); }
... }
复制代码


继续查看 remotingServer 的启动流程,进入 NettyRemotingServer#start 方法:


public void start() {    ...
ServerBootstrap childHandler = this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector) ... .childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ch.pipeline() .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler) .addLast(defaultEventExecutorGroup, encoder, new NettyDecoder(), new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()), connectionManageHandler, // 处理业务请求的handler serverHandler ); } });
...
}
复制代码


这就是一个标准的 netty 服务启动流程了,套路与 nameServer 的启动是一样的。关于 netty 的相关内容,这里我们仅关注 pipeline 上的 channelHandler,在 netty 中,处理读写请求的操作为一个个 ChannelHandler,remotingServer 中处理读写请求的 ChanelHandler 为 NettyServerHandler,代码如下:


 @ChannelHandler.Sharableclass NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
@Override protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { processMessageReceived(ctx, msg); }}
复制代码


这块的操作与 nameServer 对外提供的服务极相似(就是同一个类),最终调用的是 NettyRemotingAbstract#processRequestCommand 方法:


 public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {     // 根据 code 从 processorTable 获取 Pair    final Pair<NettyRequestProcessor, ExecutorService> matched         = this.processorTable.get(cmd.getCode());    // 找不到默认值        final Pair<NettyRequestProcessor, ExecutorService> pair =          null == matched ? this.defaultRequestProcessor : matched;
...
// 从 pair 中拿到 Processor 进行处理 NettyRequestProcessor processor = pair.getObject1(); // 处理请求 RemotingCommand response = processor.processRequest(ctx, cmd);
.... }
复制代码


如果进入源码去看,会发现这个方法非常长,这里省略了异步处理、异常处理及返回值构造等,仅列出了关键步骤:


  1. 根据 code 从 processorTable 拿到对应的 Pair

  2. 从 Pair 里获取 Processor


最终处理请求的就是 Processor 了。

2. Processor 的注册

从上面的分析中可知, Processor 是处理消息的关键,它是从 processorTable 中获取的,这个 processorTable 是啥呢?


processorTable 是 NettyRemotingAbstract 成员变量,里面的内容是 BrokerController 在初始化时(执行 BrokerController#initialize 方法)注册的。之前在分析 BrokerController 的初始化流程时,就提到过 Processor 的提供操作,这里再回顾下:


BrokerController 的初始化方法 initialize 会调用 BrokerController#registerProcessor,Processor 的注册操作就在这个方法里:


public class BrokerController {
private final PullMessageProcessor pullMessageProcessor;
/** * 构造方法 */ public BrokerController(...) { // 处理 consumer 拉消息请求的 this.pullMessageProcessor = new PullMessageProcessor(this); }
/** * 注册操作 */ public void registerProcessor() { // SendMessageProcessor SendMessageProcessor sendProcessor = new SendMessageProcessor(this); sendProcessor.registerSendMessageHook(sendMessageHookList); sendProcessor.registerConsumeMessageHook(consumeMessageHookList); // 处理 Processor 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);
// PullMessageProcessor this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
// 省略其他许许多多的Processor注册 ...
}
...
复制代码


需要指明的是,sendProcessor 用来处理 producer 请求过来的消息,pullMessageProcessor 用来处理 consumer 拉取消息的请求。

3. 接收 producer 消息

了解完 remotingServer 的启动与 Processor 的注册内容后,接下来我们就可以分析接收 producer 消息的流程了。


producer 发送消息到 broker 时,发送的请求 code 为 SEND_MESSAGE(RocketMQ 源码 5-producer 同步发送和单向发送 第 1.4 小节),根据上面的分析,当消息过来时,会使用 NettyServerHandler 这个 ChannelHandler 来处理,之后会调用到 NettyRemotingAbstract#processRequestCommand 方法。


在 NettyRemotingAbstract#processRequestCommand 方法中,会根据消息的 code 获取对应的 Processor 来处理,从 Processor 的注册流程来看,处理该 SEND_MESSAGE 的 Processor 为 SendMessageProcessor,我们进入 SendMessageProcessor#processRequest 看看它的流程:


public RemotingCommand processRequest(ChannelHandlerContext ctx,        RemotingCommand request) throws RemotingCommandException {    RemotingCommand response = null;    try {        // broker处理接收消息        response = asyncProcessRequest(ctx, request).get();    } catch (InterruptedException | ExecutionException e) {        log.error("process SendMessage error, request : " + request.toString(), e);    }    return response;}
复制代码


没干啥事,一路跟下去,直接看普通消息的流程,进入 SendMessageProcessor#asyncSendMessage 方法:


private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx,         RemotingCommand request, SendMessageContext mqtraceContext,         SendMessageRequestHeader requestHeader) {    final RemotingCommand response = preSend(ctx, request, requestHeader);    final SendMessageResponseHeader responseHeader         = (SendMessageResponseHeader)response.readCustomHeader();
if (response.getCode() != -1) { return CompletableFuture.completedFuture(response); }
final byte[] body = request.getBody();
int queueIdInt = requestHeader.getQueueId(); TopicConfig topicConfig = this.brokerController.getTopicConfigManager() .selectTopicConfig(requestHeader.getTopic());
// 如果没指定队列,就随机指定一个队列 if (queueIdInt < 0) { queueIdInt = randomQueueId(topicConfig.getWriteQueueNums()); }
// 将消息包装为 MessageExtBrokerInner MessageExtBrokerInner msgInner = new MessageExtBrokerInner(); msgInner.setTopic(requestHeader.getTopic()); msgInner.setQueueId(queueIdInt);
// 省略处理 msgInner 的流程 ...
CompletableFuture<PutMessageResult> putMessageResult = null; Map<String, String> origProps = MessageDecoder .string2messageProperties(requestHeader.getProperties()); String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED); // 发送事务消息 if (transFlag != null && Boolean.parseBoolean(transFlag)) { ... // 发送事务消息 putMessageResult = this.brokerController.getTransactionalMessageService() .asyncPrepareMessage(msgInner); } else { // 发送普通消息 putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner); } return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);}
复制代码


这个方法是在准备消息的发送数据,所做的工作如下:


  1. 如果没指定队列,就随机指定一个队列,一般情况下不会给消息指定队列的,但如果要发送顺序消息,就需要指定队列了,这点后面再分析。

  2. 构造 MessageExtBrokerInner 对象,就是将 producer 上送的消息包装下,加上一些额外的信息,如消息标识 msgId、发送时间、topic、queue 等。

  3. 发送消息,这里只是分为两类:事务消息与普通消息,这里我们主要关注普通消息,事务消息后面再分析。


进入普通消息的发送方法 DefaultMessageStore#asyncPutMessage:


public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {    ...    // 保存到 commitLog    CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);    ...}
复制代码

3.1 commitLog 写入原理

一个 broker 逻辑上对应着一个 commitLog,你可以把它看作一个大文件,然后这个 broker 收到的所有消息都写到这个里面,但是物理上 ROCKET_HOME/commitlog/00000000000000000000 这个路径存储的,它是由若干个文件组成的,每个文件默认大小是 1G,然后每个文件都对应这个一个 MappedFile,00000000000000000000 就是第一个 MappedFile 对应的物理文件,每个文件的文件名就是在 commitLog 里面的一个其实 offset,第二个文件名就是 00000000001073741824,也就是上一个 MappedFile 文件起始 offset 加上每个文件的大小,这个 MappedFile 就是 RocketMQ 的黑科技,使用了内存映射技术来提高文件的访问速度与写入速度,然后都是采用追加写的方式提高写入速度。



我们直接看官方的描述(链接:github.com/apache/rock…):


rocketMq 消息存储架构图

消息存储架构图中主要有下面三个跟消息存储相关的文件构成。

(1) CommitLog:消息主体以及元数据的存储主体,存储 Producer 端写入的消息主体内容,消息内容不是定长的。单个文件大小默认 1G ,文件名长度为 20 位,左边补零,剩余为起始偏移量,比如 00000000000000000000 代表了第一个文件,起始偏移量为 0,文件大小为 1G=1073741824;当第一个文件写满了,第二个文件为 00000000001073741824,起始偏移量为 1073741824,以此类推。消息主要是顺序写入日志文件,当文件满了,写入下一个文件;

(2) ConsumeQueue:消息消费队列,引入的目的主要是提高消息消费的性能,由于 RocketMQ 是基于主题 topic 的订阅模式,消息消费是针对主题进行的,如果要遍历 commitlog 文件中根据 topic 检索消息是非常低效的。Consumer 即可根据 ConsumeQueue 来查找待消费的消息。其中,ConsumeQueue(逻辑消费队列)作为消费消息的索引,保存了指定 Topic 下的队列消息在 CommitLog 中的起始物理偏移量 offset,消息大小 size 和消息 Tag 的 HashCode 值。consumequeue 文件可以看成是基于 topic 的 commitlog 索引文件,故 consumequeue 文件夹的组织方式如下:topic/queue/file 三层组织结构,具体存储路径为:$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样 consumequeue 文件采取定长设计,每一个条目共 20 个字节,分别为 8 字节的 commitlog 物理偏移量、4 字节的消息长度、8 字节 tag hashcode,单个文件由 30W 个条目组成,可以像数组一样随机访问每一个条目,每个 ConsumeQueue 文件大小约 5.72M;

(3) IndexFile:IndexFile(索引文件)提供了一种可以通过 key 或时间区间来查询消息的方法。Index 文件的存储位置是:HOME\store\index{fileName},文件名 fileName 是以创建时的时间戳命名的,固定的单个 IndexFile 文件大小约为 400M,一个 IndexFile 可以保存 2000W 个索引,IndexFile 的底层存储设计为在文件系统中实现 HashMap 结构,故 rocketmq 的索引文件其底层实现为 hash 索引。

在上面的 RocketMQ 的消息存储整体架构图中可以看出,RocketMQ 采用的是混合型的存储结构,即为 Broker 单个实例下所有的队列共用一个日志数据文件(即为 CommitLog)来存储。RocketMQ 的混合型存储结构(多个 Topic 的消息实体内容都存储于一个 CommitLog 中)针对 Producer 和 Consumer 分别采用了数据和索引部分相分离的存储结构,Producer 发送消息至 Broker 端,然后 Broker 端使用同步或者异步的方式对消息刷盘持久化,保存至 CommitLog 中。只要消息被刷盘持久化至磁盘文件 CommitLog 中,那么 Producer 发送的消息就不会丢失。

正因为如此,Consumer 也就肯定有机会去消费这条消息。当无法拉取到消息后,可以等下一次消息拉取,同时服务端也支持长轮询模式,如果一个消息拉取请求未拉取到消息,Broker 允许等待 30s 的时间,只要这段时间内有新消息到达,将直接返回给消费端。这里,RocketMQ 的具体做法是,使用 Broker 端的后台服务线程—ReputMessageService 不停地分发请求并异步构建 ConsumeQueue(逻辑消费队列)和 IndexFile(索引文件)数据。

3.2 CommitLog#asyncPutMessage

继续进入 CommitLog#asyncPutMessage 方法,这个方法有点长,我们分部分:


public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {    // Set the storage time 设置存储时间    msg.setStoreTimestamp(System.currentTimeMillis());    // Set the message body BODY CRC (consider the most appropriate setting    // on the client)    // 设置crc    msg.setBodyCRC(UtilAll.crc32(msg.getBody()));    // Back to Results    AppendMessageResult result = null;
StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
String topic = msg.getTopic(); int queueId = msg.getQueueId();
// 获取事务状态 final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag()); // 事务 if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) { // Delay Delivery 延时消息的处理 if (msg.getDelayTimeLevel() > 0) { if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) { msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()); }
// 设置延迟队列 topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC; queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
// Backup real topic, queueId MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic()); MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId())); msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
msg.setTopic(topic); msg.setQueueId(queueId); } } ...
复制代码


这一部分其实就是从 msg 中获取一些信息,判断处理一下这个延时消息。


long elapsedTimeInLock = 0;MappedFile unlockMappedFile = null;// 获取最后一个 MappedFileMappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
// 获取写入锁putMessageLock.lock(); //spin or ReentrantLock ,depending on store configtry { long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now(); // 开始在锁里的时间 this.beginTimeInLock = beginLockTimestamp;
// Here settings are stored timestamp, in order to ensure an orderly // global // 设置写入的时间戳,确保它是有序的, msg.setStoreTimestamp(beginLockTimestamp);
// 判断MappedFile 是否是 null 或者是否是满了 if (null == mappedFile || mappedFile.isFull()) { mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise } if (null == mappedFile) { log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString()); beginTimeInLock = 0; return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null)); }
// todo 往mappedFile追加消息 result = mappedFile.appendMessage(msg, this.appendMessageCallback);
...
复制代码


这一部分就比较重要了,首先是从 mappedFileQueue 中获取最后一个 MappedFile,这个就是拿集合最后一个元素,因为都是有序的,最后一个元素就是最后一个 MappedFile,接着就是获取锁了,这个锁也是比较有讲究的,可以设置使用 ReentrantLock 也可以设置使用 cas,默认是使用 cas,接着就是设置 beginTimeInLock 这个变量了,这个变量我们在判断 os page cache 繁忙的时候说过,就是获取到锁的一个时间戳,在释放锁之前会重置成 0,接着就是判断 mappedFile 是不是 null 或者是不是满了,如果是的话就要新建一个了。


接着就是最最最重要的了 往 mappedFile 中追加消息,

mappedFile.appendMessage

/** * 将消息追加到MappedFile文件中 */public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {    assert messageExt != null;    assert cb != null;
// 获取MappedFile当前文件写指针 int currentPos = this.wrotePosition.get();
// 如果currentPos小于文件大小 if (currentPos < this.fileSize) { ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice(); byteBuffer.position(currentPos); AppendMessageResult result; // 单个消息 if (messageExt instanceof MessageExtBrokerInner) { // todo result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt); // 批量消息 } else if (messageExt instanceof MessageExtBatch) { result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt); } else { return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR); } this.wrotePosition.addAndGet(result.getWroteBytes()); this.storeTimestamp = result.getStoreTimestamp(); return result; } // 如果currentPos大于或等于文件大小,表明文件已写满,抛出异常 log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize); return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);}
复制代码


这里首先获取了一下这个 mappedFile 写到哪个位置了,它这个位置是从 0 开始的,然后判断一下当前位置与文件大小做对比,要是大于的就超了文件大小了,接着是获取 writerbuffer 因为这里是没有开启 transientStorePool 的,所以它是个空的,就会使用 mmapedByteBuffer,接着就是调用回调的 doAppend 追加消息了,我们看下它的参数, 第一个是开始 offset,这个 offset 是 commitlog 的一个 offset,举个例子,第一个 MappedFile 的开始 offset 是 0,然后一个 MappedFile 的大小是 1g,然后第二个 MappedFile 就得从 1073741824(1g)开始了,第二个参数是 bytebuffer,这个不用多说,第三个是这个 MappedFile 还空多少字节没用,第四个就是消息了。


我们来看下这个 doAppend 方法,这个也有点长,我们需要分开看下:


/** * // 只是将消息追加到内存中 * @param fileFromOffset 文件的第一个偏移量(就是MappedFile是从哪个地方开始的) */public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,    final MessageExtBrokerInner msgInner) {    // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
// PHY OFFSET long wroteOffset = fileFromOffset + byteBuffer.position();
int sysflag = msgInner.getSysFlag();
int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4; int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4; ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength); ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
this.resetByteBuffer(storeHostHolder, storeHostLength); // 创建全局唯一消息id String msgId; if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) { msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset); } else { msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset); }
// Record ConsumeQueue information keyBuilder.setLength(0); keyBuilder.append(msgInner.getTopic()); keyBuilder.append('-'); keyBuilder.append(msgInner.getQueueId()); String key = keyBuilder.toString(); // 获取该消息在消息队列的物理偏移量 Long queueOffset = CommitLog.this.topicQueueTable.get(key); if (null == queueOffset) { queueOffset = 0L; CommitLog.this.topicQueueTable.put(key, queueOffset); }
// Transaction messages that require special handling final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag()); switch (tranType) { // Prepared and Rollback message is not consumed, will not enter the // consumer queue case MessageSysFlag.TRANSACTION_PREPARED_TYPE: case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: queueOffset = 0L; break; case MessageSysFlag.TRANSACTION_NOT_TYPE: case MessageSysFlag.TRANSACTION_COMMIT_TYPE: default: break; }
复制代码


这一部分主要就是 计算了一下这个消息写在 commitlog 中的一个 offset,接着就是生成一个 msgId,然后根据 topic 与 queueId 从缓存中获取了一下这个 queueId 对应的一个 queue 的 offset,这个其实就是添加一个消息加 1,然后就是事务的东西了,如果有事务,然后还在准备阶段或者回滚阶段,就将 queue offset 设置成 0,再往下其实就是处理消息,然后写到 buffer 中了。


/** * Serialize message */final byte[] propertiesData =    msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;
if (propertiesLength > Short.MAX_VALUE) { log.warn("putMessage message properties length too long. length={}", propertiesData.length); return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);}
final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);final int topicLength = topicData.length;
final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
// todo 计算消息总长度final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
// Exceeds the maximum message if (msgLen > this.maxMessageSize) { // 最大消息长度 65536 CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength + ", maxMessageSize: " + this.maxMessageSize); return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);}...
复制代码


这里首先是获取了一下消息里面的 properties,将它转成字节数组,计算了一下长度,接着就是将 topic 转成字节数据,计算了一下长度,获取了一下 body 的长度,就是你往 Message 塞得内容长度,重点来,计算 消息的总长度,然后判断一下长度是否超长。其中 calMsgLength 如下:


    // 计算消息长度    protected static int calMsgLength(int sysFlag, int bodyLength, int topicLength, int propertiesLength) {        int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;        int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;        final int msgLen = 4 //TOTALSIZE 消息条目总长度,4字节            + 4 //MAGICCODE 魔数,4字节。固定值0xdaa320a7            + 4 //BODYCRC 消息体的crc校验码,4字节            + 4 //QUEUEID 消息消费队列ID,4字节            + 4 //FLAG 消息标记,RocketMQ对其不做处理,供应用程序使用, 默认4字节            + 8 //QUEUEOFFSET 消息在ConsumeQuene文件中的物理偏移量,8字节。            + 8 //PHYSICALOFFSET 消息在CommitLog文件中的物理偏移量,8字节            + 4 //SYSFLAG 消息系统标记,例如是否压缩、是否是事务消息 等,4字节            + 8 //BORNTIMESTAMP 消息生产者调用消息发送API的时间戳,8字 节            + bornhostLength //BORNHOST 消息发送者IP、端口号,8字节            + 8 //STORETIMESTAMP 消息存储时间戳,8字节            + storehostAddressLength //STOREHOSTADDRESS  Broker服务器IP+端口号,8字节            + 4 //RECONSUMETIMES 消息重试次数,4字节            + 8 //Prepared Transaction Offset 事务消息的物理偏移量,8 字节。            + 4  // 消息体长度,4字节            + (bodyLength > 0 ? bodyLength : 0) // BODY  消息体内容,长度为bodyLenth中存储的值            + 1 // 主题存储长度,1字节,表示主题名称不能超过255个字符。            + topicLength //TOPIC 主题,长度为TopicLength中存储的值            + 2  // 消息属性长度,2字节,表示消息属性长 度不能超过65536个字符。            + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength  消息属性,长度为PropertiesLength中存储的 值            + 0;        return msgLen;    }
复制代码


继续:


...// todo 消息长度+END_FILE_MIN_BLANK_LENGTH 大于commitLog的空闲空间,则返回END_OF_FILEif ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {    this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);    // 1 TOTALSIZE  4字节存储当前文件的剩余空间    this.msgStoreItemMemory.putInt(maxBlank);    // 2 MAGICCODE 4字节存储魔数    this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);    // 3 The remaining space may be any value    // Here the length of the specially set maxBlank    final long beginTimeMills = CommitLog.this.defaultMessageStore.now();    byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);    return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),        queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);}...
复制代码


判断剩下的空间能不能放开,如果放不开的话,就塞上一个结束的东西,8 个字节是正经的,剩下的随意,然后返回文件满了的状态。


...// Initialization of storage space// 初始化存储空间this.resetByteBuffer(msgStoreItemMemory, msgLen);// 1 TOTALSIZEthis.msgStoreItemMemory.putInt(msgLen);// 2 MAGICCODEthis.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);// 3 BODYCRCthis.msgStoreItemMemory.putInt(msgInner.getBodyCRC());// 4 QUEUEIDthis.msgStoreItemMemory.putInt(msgInner.getQueueId());// 5 FLAGthis.msgStoreItemMemory.putInt(msgInner.getFlag());// 6 QUEUEOFFSETthis.msgStoreItemMemory.putLong(queueOffset);// 7 PHYSICALOFFSETthis.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());// 8 SYSFLAGthis.msgStoreItemMemory.putInt(msgInner.getSysFlag());// 9 BORNTIMESTAMPthis.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());// 10 BORNHOSTthis.resetByteBuffer(bornHostHolder, bornHostLength);this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));// 11 STORETIMESTAMPthis.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());// 12 STOREHOSTADDRESSthis.resetByteBuffer(storeHostHolder, storeHostLength);this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));// 13 RECONSUMETIMESthis.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());// 14 Prepared Transaction Offsetthis.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());// 15 BODYthis.msgStoreItemMemory.putInt(bodyLength);if (bodyLength > 0)    this.msgStoreItemMemory.put(msgInner.getBody());// 16 TOPICthis.msgStoreItemMemory.put((byte) topicLength);this.msgStoreItemMemory.put(topicData);// 17 PROPERTIESthis.msgStoreItemMemory.putShort((short) propertiesLength);if (propertiesLength > 0)    this.msgStoreItemMemory.put(propertiesData);
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();// Write messages to the queue bufferbyteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);...
复制代码


这个就是封装消息了,最后将消息放到 byteBuffer 中。


...// 创建AppendMessageResult    AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,        msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
switch (tranType) { case MessageSysFlag.TRANSACTION_PREPARED_TYPE: case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: break; case MessageSysFlag.TRANSACTION_NOT_TYPE: case MessageSysFlag.TRANSACTION_COMMIT_TYPE: // The next update ConsumeQueue information // 更新消息队列的逻辑偏移量 CommitLog.this.topicQueueTable.put(key, ++queueOffset); break; default: break; } return result;}
复制代码


最后就是封装追加消息的结果是 put_ok,然后更新 queue offset ,其实就是+1。


接下来我们回过头来看下 appendMessagesInner 的后半部分,


...this.wrotePosition.addAndGet(result.getWroteBytes());this.storeTimestamp = result.getStoreTimestamp();return result;
复制代码


这里其实就是更新了一下 这个 MappedFile 写到哪个地方了,更新了下写入时间。


回到 commitLog 的 putMessage 方法:


...    // todo 往mappedFile追加消息    result = mappedFile.appendMessage(msg, this.appendMessageCallback);    switch (result.getStatus()) {        case PUT_OK:            break;        case END_OF_FILE:            unlockMappedFile = mappedFile;            // Create a new file, re-write the message            mappedFile = this.mappedFileQueue.getLastMappedFile(0);            if (null == mappedFile) {                // XXX: warn and notify me                log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());                beginTimeInLock = 0;                return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result));            }            result = mappedFile.appendMessage(msg, this.appendMessageCallback);            break;        case MESSAGE_SIZE_EXCEEDED:        case PROPERTIES_SIZE_EXCEEDED:            beginTimeInLock = 0;            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));        case UNKNOWN_ERROR:            beginTimeInLock = 0;            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));        default:            beginTimeInLock = 0;            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));    }
elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp; beginTimeInLock = 0;} finally { putMessageLock.unlock();}...
复制代码


这里追加完成了,就需要判断追加状态了,如果是那种 MappedFile 放不开消息的情况,它会重新获取一个 MappedFile,然后重新追加,在释放锁之前,它还会将 beginTimeInLock 这个字段重置为 0;


...if (elapsedTimeInLock > 500) {        log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);    }
if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) { this.defaultMessageStore.unlockMappedFile(unlockMappedFile); }
PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
// Statistics storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet(); storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
// todo 消息首先进入pagecache,然后执行刷盘操作, CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg); // todo 接着调用submitReplicaRequest方法将消息提交到HaService,进行数据复制 CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);
// todo 这里使用了ComplateFuture的thenCombine方法,将刷盘、复制当成一 // todo 个联合任务执行,这里设置消息追加的最终状态 return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> { if (flushStatus != PutMessageStatus.PUT_OK) { putMessageResult.setPutMessageStatus(flushStatus); } if (replicaStatus != PutMessageStatus.PUT_OK) { putMessageResult.setPutMessageStatus(replicaStatus); if (replicaStatus == PutMessageStatus.FLUSH_SLAVE_TIMEOUT) { log.error("do sync transfer other node, wait return, but failed, topic: {} tags: {} client address: {}", msg.getTopic(), msg.getTags(), msg.getBornHostNameString()); } } return putMessageResult; });}
复制代码


判断了一下耗时,如果是大于 500ms 的话,打印警告,封装 put 消息的结果,统计 store,可以看到后面调用了 2 个方法,一个是刷盘的,一个是同步消息的,我们这里要看下这个刷盘动作:


public CompletableFuture<PutMessageStatus> submitFlushRequest(AppendMessageResult result, MessageExt messageExt) {    // Synchronization flush 同步刷盘    if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {        final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;        if (messageExt.isWaitStoreMsgOK()) {            // 构建GroupCommitRequest同步任务并提交到GroupCommitRequest            GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),                    this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());            // 刷盘请求            service.putRequest(request);            return request.future();        } else {            service.wakeup();            return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);        }    }    // Asynchronous flush  异步刷盘 这个就是靠os    else {        if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {            flushCommitLogService.wakeup();        } else  {            commitLogService.wakeup();        }        return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);    }}
复制代码


如果 broker 配置的 SYNC_FLUSH 并且是个同步消息,这个时候就会创建一个刷盘请求,然后提交刷盘请求,这个时候会等着刷盘完成,默认就是 5s。


接着就是到存储器的 putMessage 方法的后半部分了:


...// todo 存储消息CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);
putResultFuture.thenAccept((result) -> { long elapsedTime = this.getSystemClock().now() - beginTime; if (elapsedTime > 500) { log.warn("putMessage not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length); } // 记录状态 this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);
if (null == result || !result.isOk()) { // 记录状态 this.storeStatsService.getPutMessageFailedTimes().incrementAndGet(); }});
return putResultFuture;
复制代码


commitlog 存入消息之后,咱们这块也就算是完成了,最后就是回到那个 processor,然后将 put 结果写入对应的 channel 给返回去,告诉消息生产者消息写入结果 。消息存储其实就是找对应的 MappedFile,按照一定的格式往文件里面写入,需要注意的是内存映射文件。


这里附一张消息存储字段存储顺序与字段长度的图:


4. 总结

本文主要分析了 broker 接收 producer 消息的流程,流程如下:


  1. 处理消息接收的底层服务为 netty,在 BrokerController#start 方法中启动

  2. netty 服务中,处理消息接收的 channelHandler 为 NettyServerHandler,最终会调用 SendMessageProcessor#processRequest 来处理消息接收

  3. 消息接收流程的最后,MappedFile#appendMessage(...)方法会将消息内容写入到 commitLog 文件中。

用户头像

还未添加个人签名 2022-09-04 加入

热衷于分享java技术,一起交流学习,探讨技术。 需要Java相关资料的可以+v:xiaoyanya_1

评论

发布
暂无评论
RocketMQ源码-broker 消息接收流程(写入commitLog)_Java_小小怪下士_InfoQ写作社区