6、Broker消息处理流程（六）

墨蓝 2024-02-05 12:12 39阅读 0赞

前面分析完Broker启动会启动RemotingServer服务同时会注册Processor处理器，接着分析Producer进行消息的发送，当Producer发送完消息后就得到Broker去接收Producer发送的消息了。  
Producer发送给Broker消息时候，发送的请求code为SEND\_MESSAGE(这里在上一章节有过分析)，根据消息发送过来的Code，这时会调用NettyRemotingAbstract的processRequestCommand方法，该方法里面会根据消息传输的Code来取出对应的Processor，进入Processor系列类的SendMessageProcessor的asyncProcessRequest方法（前面这一部分之前都有过分析，接下来我们一起看看后面的操作，正好也将之前的知识串在一起更有利于理解和记忆）

public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,
                                                                      RemotingCommand request) throws RemotingCommandException {
        
        final SendMessageContext mqtraceContext;
        switch (request.getCode()) {
        
            // 消息重回队列
            case RequestCode.CONSUMER_SEND_MSG_BACK:
                return this.asyncConsumerSendMsgBack(ctx, request);
            default:
                // 解析消息头
                SendMessageRequestHeader requestHeader = parseRequestHeader(request);
                if (requestHeader == null) {
        
                    return CompletableFuture.completedFuture(null);
                }
                // 构建上下文，并调用处理前钩子函数
                mqtraceContext = buildMsgContext(ctx, requestHeader);
                this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
                // 判断批量消息还是单条消息
                if (requestHeader.isBatch()) {
        
                    return this.asyncSendBatchMessage(ctx, request, mqtraceContext, requestHeader);
                } else {
        
                    return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);
                }
        }
    }

首先解析消息头构建上下文，处理消息发送前钩子函数，最后异步处理消息请求，如果是批量消息调用asyncSendBatchMessage方法，如果是单条消息调用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 msgInner = new MessageExtBrokerInner();
            msgInner.setTopic(requestHeader.getTopic());
            msgInner.setQueueId(queueIdInt);
    
    
            if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) {
        
                return CompletableFuture.completedFuture(response);
            }
    
            msgInner.setBody(body);
            msgInner.setFlag(requestHeader.getFlag());
            Map<String, String> origProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
            MessageAccessor.setProperties(msgInner, origProps);
            // 时间
            msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
          // 远程地址
            msgInner.setBornHost(ctx.channel().remoteAddress());
            // 主机
            msgInner.setStoreHost(this.getStoreHost());
            // 重试次数
            msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
            String clusterName = this.brokerController.getBrokerConfig().getBrokerClusterName();
            // ...省略
    
    
            CompletableFuture<PutMessageResult> putMessageResult = null;
            String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
            // 事务消息
            if (Boolean.parseBoolean(transFlag)) {
        
                if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
        
                    response.setCode(ResponseCode.NO_PERMISSION);
                    response.setRemark(
                            "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                                    + "] sending transaction message is forbidden");
                    return CompletableFuture.completedFuture(response);
                }
                // 事务消息的状态（后面再分析）
                putMessageResult = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);
            } else {
        
                // 消息存储
                putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
            }
            // 生成结果返回
            return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);
        }

构建MessageExtBrokerInner对象，设置相关属性执行asyncPutMessage方法存储消息并将结果返回客户端。

创建响应，验证以及自动创建topic

// 准备响应，验证以及自动创建topic
    private RemotingCommand preSend(ChannelHandlerContext ctx, RemotingCommand request,
                                    SendMessageRequestHeader requestHeader) {
        
        // 准备响应
        final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
        // 设置唯一id
        response.setOpaque(request.getOpaque());
    
        response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
        response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
    
        log.debug("Receive SendMessage request command {}", request);
        // 获取broker处理请求服务的起始时间
        final long startTimestamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
    
        if (this.brokerController.getMessageStore().now() < startTimestamp) {
        
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimestamp)));
            return response;
        }
        response.setCode(-1);
        // 验证topic以及自动创建逻辑
        super.msgCheck(ctx, requestHeader, response);
        return response;
    }

this.brokerController.getBrokerConfig().isAutoCreateTopicEnable()用来判断是否支持自动创建topic，根据权限来判断如果是不支持自动创建就将权限设置为可读可写不可继承，后面我们去判断是否可以去继承，如果能继承就说明支持自动创建，这是就会new一个TopicConfig，这样就通过autoCreateTopicEnable自动来控制是否能够自动创建topic，同时也会调用registerBrokerAll方法注册到Broker路由信息里面，当然官方建议我们还是不要开启这个配置因为它没有做到压力的分摊。

存盘 asyncPutMessage方法  
根据topic查询对应的路由信息即broker。

public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
           
            msg.setStoreTimestamp(System.currentTimeMillis());
            msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
            AppendMessageResult result = null;
            StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
            String topic = msg.getTopic();
            final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
            if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
                    || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
        
                // Delay Delivery
                // 延迟消息转到系统Topic(后面在分析)
                if (msg.getDelayTimeLevel() > 0) {
        // ...省略}
            }
          // 发送消息地址
            InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();
            if (bornSocketAddress.getAddress() instanceof Inet6Address) {
        
                msg.setBornHostV6Flag();
            }
          // 存储消息地址
            InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();
            if (storeSocketAddress.getAddress() instanceof Inet6Address) {
        
                msg.setStoreHostAddressV6Flag();
            }
            // 更新消息大小
            PutMessageThreadLocal putMessageThreadLocal = this.putMessageThreadLocal.get();
            updateMaxMessageSize(putMessageThreadLocal);
            if (!multiDispatch.isMultiDispatchMsg(msg)) {
        
                PutMessageResult encodeResult = putMessageThreadLocal.getEncoder().encode(msg);
                if (encodeResult != null) {
        
                    return CompletableFuture.completedFuture(encodeResult);
                }
                msg.setEncodedBuff(putMessageThreadLocal.getEncoder().getEncoderBuffer());
            }
            PutMessageContext putMessageContext = new PutMessageContext(generateKey(putMessageThreadLocal.getKeyBuilder(), msg));
    
            long elapsedTimeInLock = 0;
            MappedFile unlockMappedFile = null;
            // 写入CommitLog文件前加锁,保证文件操作并发安全
            putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
            try {
        
                // 获取最后一个mapperFile
                MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
                long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
                this.beginTimeInLock = beginLockTimestamp;
    
                msg.setStoreTimestamp(beginLockTimestamp);
              // 如果不存在或者满了就创建一个
                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());
                    return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null));
                }
                // 实际写入CommitLog，在后面追加
                result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
                switch (result.getStatus()) {
        
                        // 添加成功直接break
                    case PUT_OK:
                        break;
                        // 表示当前文件存放不下，只保存了一部分
                    case END_OF_FILE:
                        unlockMappedFile = mappedFile;
                        // 创建一个新的文件
                        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());
                            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result));
                        }
                        // 继续追加进去
                        result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
                        break;
                    case MESSAGE_SIZE_EXCEEDED:
                    case PROPERTIES_SIZE_EXCEEDED:
                        return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));
                    case UNKNOWN_ERROR:
                        return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
                    default:
                        return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
                }
              // 锁的时间
                elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
            } finally {
        
                beginTimeInLock = 0;
                putMessageLock.unlock();
            }
    
            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()).add(1);
            storeStatsService.getSinglePutMessageTopicSizeTotal(topic).add(result.getWroteBytes());
            // 提交刷盘申请
            CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
            // 提交主从复制申请
            CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);
            return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
        
                if (flushStatus != PutMessageStatus.PUT_OK) {
        
                    putMessageResult.setPutMessageStatus(flushStatus);
                }
                if (replicaStatus != PutMessageStatus.PUT_OK) {
        
                    putMessageResult.setPutMessageStatus(replicaStatus);
                }
                return putMessageResult;
            });
        }

首先它会去处理延时消息这里我不做过细的分析，后面针对各种消息在来具体分析，接着就将消息进行编码然后加锁并写入消息以获取最后文件进行追加的方式来将消息内存文件里面，最后进行刷盘以及通知主从同步的操作。