• RocketMQ中Broker接收消息流程代码解析

    在RocketMQ中，Broker通过SendMessageProcessor来接收和处理producer发送过来的消息。

    1. private RemotingCommand sendMessage(final ChannelHandlerContext ctx,
    2.                                         final RemotingCommand request,
    3.                                         final SendMessageContext sendMessageContext,
    4.                                         final SendMessageRequestHeader requestHeader) throws RemotingCommandException {
    5.  
    6.         //1、构建Response的Header
    7.         final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
    8.         final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader();
    9.  
    10.         response.setOpaque(request.getOpaque());
    11.  
    12.         response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
    13.         response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
    14.  
    15.         log.debug("receive SendMessage request command, {}", request);
    16.         //2、判断当前时间broker是否提供服务，不提供则返回code为SYSTEM_ERROR的response
    17.         final long startTimstamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
    18.         if (this.brokerController.getMessageStore().now() < startTimstamp) {//broker还没开始提供接收消息服务
    19.             response.setCode(ResponseCode.SYSTEM_ERROR);
    20.             response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimstamp)));
    21.             return response;
    22.         }
    23.  
    24.         response.setCode(-1);
    25.         //3、检查topic和queue，如果不存在且broker设置中允许自动创建，则自动创建
    26.         super.msgCheck(ctx, requestHeader, response); 
    27.         if (response.getCode() != -1) {
    28.             return response;
    29.         }
    30.  
    31.         final byte[] body = request.getBody();
    32.         
    33.         int queueIdInt = requestHeader.getQueueId();
    34.         //4、获取topic的配置
    35.         TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
    36.         //5、如果消息中的queueId小于0，则随机选取一个queue
    37.         if (queueIdInt < 0) {
    38.             queueIdInt = Math.abs(this.random.nextInt() % 99999999) % topicConfig.getWriteQueueNums();
    39.         }
    40.         //6、重新封装request中的message成MessageExtBrokerInner
    41.         MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
    42.         msgInner.setTopic(requestHeader.getTopic());
    43.         msgInner.setQueueId(queueIdInt);
    44.         //7、对于RETRY消息，1)判断是否consumer还存在
    45.         //  2)如果超过最大重发次数，尝试创建DLQ，并将topic设置成DeadQueue,消息将被存入死信队列
    46.         if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) {
    47.             return response;
    48.         }
    49.  
    50.         msgInner.setBody(body);
    51.         msgInner.setFlag(requestHeader.getFlag());
    52.         MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
    53.         msgInner.setPropertiesString(requestHeader.getProperties());
    54.         msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
    55.         msgInner.setBornHost(ctx.channel().remoteAddress());
    56.         msgInner.setStoreHost(this.getStoreHost());
    57.         msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
    58.         PutMessageResult putMessageResult = null;
    59.         Map oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
    60.         String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
    61.         if (traFlag != null && Boolean.parseBoolean(traFlag)) {
    62.             //如果是事务消息
    63.             if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
    64.                 response.setCode(ResponseCode.NO_PERMISSION);
    65.                 response.setRemark(
    66.                     "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
    67.                         + "] sending transaction message is forbidden");
    68.                 return response;
    69.             }
    70.             putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
    71.         } else {
    72.             //8、调用MessageStore接口存储消息
    73.             putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
    74.         }
    75.         //9、根据putResult设置repsonse状态，更新broker统计信息，成功则回复producer，更新context上下文
    76.         return handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt);
    77.     }

    在步骤3中，会对topic进行检查，如果topic不存在，且设置成自动创建topic，就会在Broker上自动创建topic。

    Broker最后调用MessageStore来存储数据。

    MessageStore保存消息

    DefaultMessageStore#putMessage
    1. public PutMessageResult putMessage(MessageExtBrokerInner msg) {
    2.         if (this.shutdown) {
    3.             log.warn("message store has shutdown, so putMessage is forbidden");
    4.             return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    5.         }
    6.  
    7.         // 从节点不允许写入
    8.         if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
    9.             long value = this.printTimes.getAndIncrement();
    10.             if ((value % 50000) == 0) {
    11.                 log.warn("message store is slave mode, so putMessage is forbidden ");
    12.             }
    13.  
    14.             return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    15.         }
    16.  
    17.         // store是否允许写入
    18.         if (!this.runningFlags.isWriteable()) {
    19.             long value = this.printTimes.getAndIncrement();
    20.             if ((value % 50000) == 0) {
    21.                 log.warn("message store is not writeable, so putMessage is forbidden " + this.runningFlags.getFlagBits());
    22.             }
    23.  
    24.             return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    25.         } else {
    26.             this.printTimes.set(0);
    27.         }
    28.  
    29.         // topic过长
    30.         if (msg.getTopic().length() > Byte.MAX_VALUE) {
    31.             log.warn("putMessage message topic length too long " + msg.getTopic().length());
    32.             return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
    33.         }
    34.  
    35.         // 消息附加属性过长
    36.         if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
    37.             log.warn("putMessage message properties length too long " + msg.getPropertiesString().length());
    38.             return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null);
    39.         }
    40.  
    41.         if (this.isOSPageCacheBusy()) {
    42.             return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null);
    43.         }
    44.  
    45.         long beginTime = this.getSystemClock().now();
    46.         // 添加消息到commitLog
    47.         PutMessageResult result = this.commitLog.putMessage(msg);
    48.  
    49.         long eclipseTime = this.getSystemClock().now() - beginTime;
    50.         if (eclipseTime > 500) {
    51.             log.warn("putMessage not in lock eclipse time(ms)={}, bodyLength={}", eclipseTime, msg.getBody().length);
    52.         }
    53.         this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime);
    54.  
    55.         if (null == result || !result.isOk()) {
    56.             this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
    57.         }
    58.  
    59.         return result;
    60.     }

    putMessage会做一下检查，然后调用CommitLog的putMessage方法来写入消息。

    CommitLog保存消息

    1. public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
    2.         // 1、Set the storage time
    3.         msg.setStoreTimestamp(System.currentTimeMillis());
    4.         // 2、Set the message body BODY CRC (consider the most appropriate setting
    5.         // on the client)
    6.         msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
    7.         // Back to Results
    8.         AppendMessageResult result = null;
    9.  
    10.         StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
    11.  
    12.         String topic = msg.getTopic();
    13.         int queueId = msg.getQueueId();
    14.  
    15.         final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
    16.         if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
    17.             || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
    18.             //非事务消息
    19.             // Delay Delivery
    20.             if (msg.getDelayTimeLevel() > 0) {
    21.                 //3、延时投放消息，变更topic
    22.                 if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
    23.                     msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
    24.                 }
    25.  
    26.                 topic = ScheduleMessageService.SCHEDULE_TOPIC;
    27.                 queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
    28.  
    29.                 // Backup real topic, queueId
    30.                 MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
    31.                 MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
    32.                 msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
    33.  
    34.                 msg.setTopic(topic);
    35.                 msg.setQueueId(queueId);
    36.             }
    37.         }
    38.  
    39.         long eclipseTimeInLock = 0;
    40.         MappedFile unlockMappedFile = null;
    41.         //4、获取当前正在写入文件
    42.         MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
    43.         //5、获取写message的锁
    44.         putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
    45.         try {
    46.             long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
    47.             this.beginTimeInLock = beginLockTimestamp;//记录lock time
    48.  
    49.             // Here settings are stored timestamp, in order to ensure an orderly
    50.             // global
    51.             msg.setStoreTimestamp(beginLockTimestamp);
    52.             //6、新建一个mapp file如果文件不存在或者文件已经写满
    53.             if (null == mappedFile || mappedFile.isFull()) {
    54.                 mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
    55.             }
    56.             if (null == mappedFile) {
    57.                 //文件创建失败，则返回错误
    58.                 log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
    59.                 beginTimeInLock = 0;
    60.                 return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
    61.             }
    62.             //7、消息写文件
    63.             result = mappedFile.appendMessage(msg, this.appendMessageCallback);
    64.             switch (result.getStatus()) {
    65.                 case PUT_OK:
    66.                     break;
    67.                 case END_OF_FILE:
    68.                     //8、如果文件已满，则新建一个文件继续
    69.                     unlockMappedFile = mappedFile;
    70.                     // Create a new file, re-write the message
    71.                     mappedFile = this.mappedFileQueue.getLastMappedFile(0);
    72.                     if (null == mappedFile) {
    73.                         log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
    74.                         beginTimeInLock = 0;
    75.                         return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
    76.                     }
    77.                     result = mappedFile.appendMessage(msg, this.appendMessageCallback);
    78.                     break;
    79.                 case MESSAGE_SIZE_EXCEEDED:
    80.                 case PROPERTIES_SIZE_EXCEEDED:
    81.                     beginTimeInLock = 0;
    82.                     return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
    83.                 case UNKNOWN_ERROR:
    84.                     beginTimeInLock = 0;
    85.                     return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
    86.                 default:
    87.                     beginTimeInLock = 0;
    88.                     return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
    89.             }
    90.  
    91.             eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
    92.             beginTimeInLock = 0;
    93.         } finally {
    94.             //9、释放第5)步中获取到的锁
    95.             putMessageLock.unlock();
    96.         }
    97.  
    98.         if (eclipseTimeInLock > 500) {//写消息时间过长
    99.             log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipseTimeInLock, msg.getBody().length, result);
    100.         }
    101.         //unlock已经写满的文件，释放内存锁
    102.         if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
    103.             this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
    104.         }
    105.  
    106.         PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
    107.  
    108.         // Statistics
    109.         storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
    110.         storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
    111.         //10、flush数据到磁盘，分同步和异步
    112.         handleDiskFlush(result, putMessageResult, msg);
    113.         //11、如果broker设置成SYNC_MASTER，则等SLAVE接收到数据后才返回(接收到数据是指offset延后没有超过制定的字节数)
    114.         handleHA(result, putMessageResult, msg);
    115.  
    116.         return putMessageResult;
    117.     }
    • 第3步，判断是否是Consumer发来的Retry消息，如果是则修改topic为 SCHEDULE_TOPIC_XXXX，根据延时时长计算queueId。将原始的topic和queueId放到消息的properties字段中。这样这个消息只会被重发的Schedule任务读到。
    • 第4步，前面讲MessageStore的时候讲过，CommitLog是由连续的MappedFile的列表组成的。在同一时间，只有最后一个MappedFile有写入，因为之前的文件都已经写满了，所以这里是取最后一个。
    • 第6步，如果commitLog是第一次启动，或者文件size已经达到maxSize，则新建一个文件
    • 第7步，将消息内容写入MappedFile，这里会传一个callback的参数，真正的写入是在callback中实现的，后面我们再看这个实现
    • 第8步，在上一步中的appendMessage()接口中，如果文件剩余的空间已经不足以写下这条消息，则会用一个EOF消息补齐文件，然后返回一个EOF错误。在收到这个错误时，会新建一个文件，然后重写一次。
    • 第10步，用户可以使用MessageStore的FlushDiskType参数来控制数据flush到磁盘的方式，如果参数值SYNC_FLUSH，则每次写完消息都会做一次flush，完成才会返回结果。如果是ASYNC_FLUSH，只会唤醒flushCommitLogService，由它异步的去检查是否要做flush。
    • 第9步，Broker的主从数据同步也可以有两种方式，如果是SYNC_MASTER，则Master保存消息后，需要将消息同步给slave后才会返回结果。如果ASYNC_MASTER，这里不会做任何操作，由HAService的后台线程做数据同步。

    MappedFile消息写入

    1. public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {
    2.         assert messageExt != null;
    3.         assert cb != null;
    4.         //1、获取当前的write position
    5.         int currentPos = this.wrotePosition.get();
    6.  
    7.         if (currentPos < this.fileSize) {
    8.             //2、生成buffer切片
    9.             ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
    10.             byteBuffer.position(currentPos);
    11.             AppendMessageResult result = null;
    12.             if (messageExt instanceof MessageExtBrokerInner) {
    13.                 //3、写单条消息到byteBuffer
    14.                 result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);
    15.             } else if (messageExt instanceof MessageExtBatch) {
    16.                 //3、批量消息到byteBuffer
    17.                 result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
    18.             } else {
    19.                 return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
    20.             }
    21.             //4、更新write position，到最新值
    22.             this.wrotePosition.addAndGet(result.getWroteBytes());
    23.             this.storeTimestamp = result.getStoreTimestamp();
    24.             return result;
    25.         }
    26.         log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
    27.         return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
    28.     }

    消息写入首先是获取ByteBuffer，从第2步中可以发现有个判断，这里是因为MappedFile写数据到文件有两种实现方式：

    • 从FileChannel获取直接内存映射，收到消息后，将数据写入到这块内存中，内存和物理文件的数据交互由操作系统负责
    • CommitLog启动的时候初始化一块内存池(通过ByteBuffer申请的堆外内存)，消息数据首先写入内存池中，然后后台有个线程定时将内存池中的数据commit到FileChannel中。这种方式只有MessageStore是ASYNC模式时才能开启。代码中if判断writeBuffer不为空的情况就是使用的这种写入方式。

    第3步，最终回调的Callback类将数据写入buffer中，消息的序列化也是在callback里面完成的。

  • 相关阅读:
    【智慧排水】智慧排水监测系统助力城市抗洪排涝建设
    (个人杂记)第九章 串口中断3
    学习python第7天
    MySQL如何一劳永逸的永久支持输入中文
    如何解决golang开发中遇到的报错：checksum mismatch downloaded
    实现注册手机号用户
    springcloudalibaba架构(1):如何实现服务调用Ribbon和Feign
    用最少的代码模拟gRPC四种消息交换模式
    ChatGPT：Spring Boot和Maven——Java应用开发的关键工具和区别
    精品Python商铺摊位租赁管理系统
  • 原文地址：https://blog.csdn.net/bao2901203013/article/details/126202374