当前位置：首页 > news >正文

rocketmq5--（三）--broker发送消息给消费者

news 来源：原创 2025/7/21 14:37:04

202412/30回过头来记录一下：之前一直找不到在哪里把消息丢给消费者（2024/11/23），找了很久都没找到，就放弃了，然后发现，我靠原来是在poplongpollingservice里。。。。还是乱打断点，偶然找到的。。。这属于运气好，好的，接下来处理consumer流程

RocketMQ主要有两种消费模式：CLUSTERING集群消费（默认）和BROADCASTING广播消费。在CLUSTERING模式下，一个ConsumerGroup中的Consumer实例根据队列分配策略算法为Consumer分配队列，平均分摊（默认）消费消息。例如，如果Topic是Test的消息发送到该主题的不同队列中，发送了有100条消息，其中一个ConsumerGroup有3个Consumer实例，那么根据队列分配算法，每个队列都会有消费者，每个消费者实例只消费自己队列上的数据，消费完的消息不能被其他消费实例消费。在BROADCASTING模式下，消息会被所有在该Topic的ConsumerGroup中的Consumer实例共同消费。消费模式在创建Consumer时指定，不同的消费模式其内部机制也不同，消息的消费方式、记录消费进度、消息的消费状态等也都各不相同。

对于长时间没有 ACK 的消息，Broker 端并非毫无办法。Pop 消费引入了消息不可见时间（invisibleTime）的机制。当 Pop 出一条消息后，这条消息对所有消费者不可见，即进入不可见时间，当它超过该时刻还没有被 ACK，Broker 将会把它放入 Pop 专门的重试 Topic（这个过程称为 Revive），这条消息重新可以被消费。https://my.oschina.net/u/4249738/blog/5609999

rocketmq5有三种消费者：默认是pop

    CONSUME_ACTIVELY("PULL")    --CONSUME_PASSIVELY("PUSH")   --CONSUME_POP("POP")          -- 对应PopLongPollingService

顺序写随机读：

发消息时，Productor的消息确实是顺序写CommitLog
订阅消息时，Consumer也是顺序读ConsumeQueue，然而根据其中的起始物理位置偏移量offset读取消息真实内容却是随机读CommitLog。

https://zhouj000.github.io/2021/05/10/rocketmq-source1/

pullRequestHoldService

ReputMessageService

源码阅读建议：因为rocketmq中有很多topic，而同一套代码流程可能同时处理我们的生产者消费者，也可能处理系统的相关消息，比如heartbeat，所以为了打断点的时候能确保是我们的代码触发的，所以建议如下修改源码，这样一旦运行到断点处就能确定是我们的代码触发的：

 @Overridepublic GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,final int maxMsgNums, final int maxTotalMsgSize, final MessageFilter messageFilter) {if (this.shutdown) {LOGGER.warn("message store has shutdown, so getMessage is forbidden");return null;}......if (topic.equals("testz")){                    #testz是我们自己的代码中使用的topicSystem.out.println("i found you:testz");   #随便来两条语句，然后在这里打断点int a=1;                                   #这样就能确保运行到这个断点的时候topic是我们自己的topic#而不会被heartbeat等系统消息所干扰}......

消费逻辑：

client sdk中consumer有两种消息拉取模式：broker push和consumer pull 
20241230 补充：rocketmq5还有pop方式

三个where:

commitWhere表示消费者已经消费到哪了，也是另开一个线程去flush即在CommitRealTimeService.run中刷到磁盘
flushWehre表示数据已经刷盘到哪里了，flushWhere的数据也叫confirmedOffset
妈的，一大堆offset、where，都绕晕了

ReputService：

ReputService是DefaultMessageStore的一个内部类

ReputService.runwhile !this.isStopped:ReputService.doReputif ReputService.reputFromOffset < CommitLog.getMinOffset:  #commitLog的minOffset表示目前存在的消息的最小offset#因为offset是递增的，所以这就表示凡是offset小于minOffset#的消息都已经被删除了，也就是对应的文件expired了ReputService.reputFromOffset = CommitLog.getMinOffset    #所以此时需要重置reputOffset为minOffset#即第一条有效地commitLogwhile doNext:                                            ok=isCommitLogAvailable:                               #检测commitLog的offset是否有变动，new=ReputService.getReputEndOffset                   #首先获取commitLog的位置#有两个offset：maxOffset和confirmedOffset#maxOffset表示commitLog的当前位置#并且这一块数据可能还没有flush到磁盘，所以是不稳定的#confirmedOffset则是已经刷新到磁盘的数据的最大offset#所以如果开启读未提交则返回maxOffset，反之返回confirmedOffsetMessageStoreConfig.isReadUnCommitted?CommitLog.getMaxOffset():CommitLog.getConfirmOffset()return cur<new                                       #一旦检测到commitLog的offset大于记录的reputOffset#就说明有更新即本次需要doReput而不是sleepCommitLog.getData(reputFromOffset)                     #从commitLog获取新增的数据#所有数据都是存放在commitLog，consumerQueue存的是索引没有存原始数据CommitLog.checkMessageAndReturnSize                     #检查消息是否ok比如检查crc对不对，以及获取消息大小        if reputFromOffset + size > getReputEndOffset():        #如果当前reputOffset+msgSize超过了offsetdoNext = false                                        #就表示这不是一条完整的消息即所有消息都处理完了，可以结束本次循环break if dispatchRequest.isSuccess():                         #如果这条消息okif size > 0:                                          #如果size大于0ReputService.doDispatch                             #分发消息，默认有4个dispatch，分别是：#CommitLogDispatcherCalcBitMap：#DefaultMessageStore.CommitLogDispatcherBuildConsumeQueue、#DefaultMessageStore.CommitLogDispatcherBuildIndex、#CommitLogDispatcherCompaction1: 计算consumerQueue对应的bitmapFilter         CommitLogDispatcherCalcBitMap.dispatch            #1:CommitLogDispatcherCalcBitMap计算consumerQueue对应的bitmapFilter  ...略...                                        #估计是用来加快查找的，和indexFile类似#不过默认是不启用的2：把offset写入consumerQueue文件CommitLogDispatcherBuildConsumeQueue.dispatch     #2:把offset写入consumerQueue文件switch (tranType):                            case MessageSysFlag.TRANSACTION_NOT_TYPE:     #如果是非事务消息即普通消息case MessageSysFlag.TRANSACTION_COMMIT_TYPE:  #或者是COMMIT消息即已完成的事务消息#即prepare或者rollback的消息我们就不需要写入consumerQueueDefaultMessageStore.putMessagePositionInfo  #执行分发ConsumerQueueStore.putMessagePositionInfoWrapper #ConsumerQueueStoreInterface可以有多种实现：default和rocksdbConsuemrQueueStore.findOrCreateConsumeQueue    #分发时首先查找consuemrQueue对应的文件#会把msg的offset写入consumerQueue对应的文件#consumerQueue是构建在commitLog之上的逻辑队列#consumerQueue中只保存msg的offset#读取一个消息的时候要先从consuemrQueue对应的文件中#读出offset，然后拿着这个offset再去commitlog中读取消息体ConcurrentMap<Integer, ConsumeQueueInterface> map = consumeQueueTable.get(topic) #因为有很多topic而且每个topic都有多个队列#所以用一个map<topicName,topicQueuInfo>加速查找topic的queue信息#然后又用一个map<queueId,queueStore>来存放该topic的每个queue信息#所以首先根据topic获取map<topicName,topicQueuInfo>ConsumeQueueInterface logic=map.get(queueId) #然后根据queueId获取queueStoreif logic != null:                            #如果存在对应的信息则直接返回  return logic;else:newLogic = new ConsumeQueue(               #如果不存在对应的queustore，则直接创建一个新的topic,                                   #这是consuemrQueue对应的topic名queueId,                                 #对应的queueidgetStorePathConsumeQueue                 #consumerQueue路径$qpath=$storePathRootDir\\consumerqueue#完整文件路径为filePath:#$qpath\\$topicName\\$queueid\\$startOffsetthis.messageStoreConfig.getMappedFileSizeConsumeQueue(),this.messageStore);map.putIfAbsent(queueId, newLogic)         #新创建的信息放到map<queueId,queueStore>中return newLogicAbstractConsumerQueueStore.putMessagePositionInfoWrapper          #把msg的offset追加到consumerQueue文件末尾ConsumerQueueStore.putMessagePositionInfoWrapperRunningFlags.isCQWriteable()                                  #判断当前consumerQueue是否可写result=ConsuemrQueue.putMessagePositionInfo                   #把offset写入consumerQueue文件if MessageStoreConfig.isPutConsumeQueueDataByFileChannel(): #默认是fileDefaultMappedFile.appendMessageUsingFileChannel           #写入磁盘......---->flush线程会定时刷盘consumerQueue的数据刷盘并更新onfirmOffsetFlushConsumeQueueService.run                          #这里是另一个异步刷盘线程while !this.isStopped:                              #死循环FlushConsumeQueueService.doFlush                  #刷盘ConsumeQueueStore.flush                         #执行刷盘ConsumerStore.getLifeCycle                    #先获取consumerQueue对象ConsumerStore.findOrCreateConsumeQueueConsumerQueue.flush                           #然后刷盘和更新flushWhereDefaultMappedFile.flush                     #刷盘MappedFileQueue.setFlushedWhere             #更新flushWhereif messageStoreConfig.isEnableCompaction:       #如果开启了compactioncompactionStore.flush(flushConsumeQueueLeastPages) #则也要处理compactionStoreelse:DefaultMappedFile.appendMessage< - - - 回到reputService.run线程if !notifyMessageArriveInBatch:DefaultMessageStore.notifyMessageArriveIfNecessaryNotifyMessageArrivingListener.arrivingPullRequestHoldService.notifyMessageArriving...暂时略...PopMessageProcessor.notifyMessageArriving...暂时略...NotificationProcessor.notifyMessageArriving...暂时略...if !notifyMessageArriveInBatch:ReputService.notifyMessageArriveIfNecessaryelse if size == 0:                                    #如果消息体size为0，说明下一个offset是在下一个mappedFile文件中了#（因为commitlog是一个逻辑文件，实际由多个offset连续的文件组成）reputFromOffset=CommitLog.rollNextFile              #则直接更新reputFromOffset来跳过这条消息ReputService.finishCommitLogDispatch                    #空函数，什么也不做

PopLongPollingService：

这是单独的线程负责查询消息并返回给consumer，我真的郁闷死,原来rocketmq5是有3种消费方式，而默认的是pop。。。。我一直以为是pull，挂不得我就说pullMessageProcessor中怎么只有heartBeat消息，怎么就没有对应的testz主题的消息。。。。一度使我放弃。。。。还是业务不精搞的鬼。。。

PopLongPollingService.runfor (Map.Entry<String, ConcurrentSkipListSet<PopRequest>> entry : pollingMap.entrySet())PopLongPollingService.wakeupAbstractExecutorService.submitresp=PopMessageProcessor.processRequest    #核心函数RemotingCommand.decodeCommandCustomHeader    #解析消费者请求的请求头ConsumerManager.compensateBasicConsumerInfo( #如果指定的消费者的信息不存在，则创建并保存到map中requestHeader.getConsumerGroup()           #消费者组名ConsumeType.CONSUME_POP,                   #消费者类型为POPMessageModel.CLUSTERING                    #消费方式为集群即所有消费者均摊)isTimeOut:=PopMessageRequestHeader.isTimeoutTooMuch    #判断是否超时即太久没有popreturn System.currentTimeMillis() - bornTime - pollTime > 500; #默认是500msif isTimeOut:                                          #如果超时就返回失败response.setRemark("the broker[%s] pop message is timeout too much")return PermName.isReadable                          #检查该broker是否允许消费者读取，默认是可读可写的if requestHeader.getMaxMsgNums() > 32:       #一次最多只允许broker pop 32条消息给消费者response.setCode(ResponseCode.SYSTEM_ERROR);response.setRemark("the broker[%s] pop message's num is greater than 32")return response;if !MessageStoreConfig.isTimerWheelEnable():    #检查该broker是否允许pop消息response.setCode(ResponseCode.SYSTEM_ERROR);response.setRemark("the broker[%s] pop message is forbidden because timerWheelEnable is false")return response;TopicConfigManager.selectTopicConfig           #从map中获取topic对应的config对象...略去一系列对topic的检查，比如是否可读，queueid对应的读队列是否存在等...  SubscriptionGroupManager.findSubscriptionGroupConfig  #获取消费者组信息BrokerController.getBrokerConfig               #获取broker配置信息   if requestHeader.getExp() != null && !String.isEmpty(): #获取表达式，一般是TAG表达式，如果TAG不为空则处理tag#这里的逻辑是：如果请求中设置了tag，就使用请求中的tag和表达式#如果请求中没有设置tag在，则默认使用*通配符匹配所有消息即不过滤任何消息subscriptionData = FilterAPI.build            #构建订阅信息ConsumerManager.compensateSubscribeData       #保存订阅信息retryTopic = KeyBuilder.buildPopRetryTopic    #构建retryTopic名字，每个topic都有一个对应的retryTopicretrySubscriptionData = FilterAPI.build       #构建retry订阅信息ConsumerManager.compensateSubscribeData       #保存retrytopic上的订阅信息consummerFilterData=nil                       #consumerFilterData内部含有布隆过滤器产生的数据，方便过滤消息if !ExpressionType.isTagType:                 #如果不是tag类型   consumerFilterData = ConsumerFilterManager.build #则创建consumerFilter  messageFilter = new ExpressionMessageFilter   #根据consumerFilter创建msgFilter#如果用了tag则consumerFilter会为nilelse:                                           #如果请求中没有设置tag，则默认是*，也就是会匹配全部消息#else和if里面的操作几乎一模一样subscriptionData = FilterAPI.build(requestHeader.getTopic(), "*", ExpressionType.TAG) ConsumerManager.compensateSubscribeData       #保存订阅信息retryTopic = KeyBuilder.buildPopRetryTopic    #构建retryTopic名字，每个topic都有一个对应的retryTopicretrySubscriptionData = FilterAPI.build(retryTopic, "*", ExpressionType.TAG) #构建retry订阅信息ConsumerManager.compensateSubscribeData       #保存retrytopic上的订阅信息int randomQ = random.nextInt(100)               #Pop模式中每次消费者请求都会概率性重试retry队列 #就是判断randomQ< 阈值，小于则需要重试int reviveQid;                                  #reviveQueueId：pop模式会把每次对应的checkPointMessage#保存到reviveTopic的reviveQueueId对应的队列中if requestHeader.isOrder:                       #如果请求时顺序消息（不太懂）reviveQid = KeyBuilder.POP_ORDER_REVIVE_QUEUE #则reviveQid为指定queueidelse:                                           #否则轮询使用不同的reviveQueuereviveQid = ckMessageNumber.getAndIncrement() %BrokerConfig.getReviveQueueNumneedRetry = randomQ < brokerConfig.getPopFromRetryProbability() #判断本次是否消费重试队列，比例可调if (brokerConfig.isEnableRetryTopicV2() &&      #如果开启了v2版本的pop retrybrokerConfig.isRetrieveMessageFromPopRetryTopicV1())  #并且也开启了v2版本的pop retry{                          needRetryV1 = randomQ % 2 == 0;               #那么就%2来表示本次使用哪个retry策略，不过默认是v1开启，v2关闭}if needRetry && !requestHeader.isOrder():        #如果需要消费重试队列并且请求不是顺序的，那么就消费重试topicif needRetryV1:                                #v1和v2版本的retrytopic的名字构建策略不一样，这里如果是v1retryTopic = KeyBuilder.buildPopRetryTopicV1 #则构建v1版本的retryTopic名字#v1和v2的区别只有POP_RETRY_SEPARATOR_V1为_#POP_RETRY_SEPARATOP_V2为+return MixAll.RETRY_GROUP_TOPIC_PREFIX + cid + POP_RETRY_SEPARATOR_V1 + topic  PopMessageProcessor.popMsgFromTopic(retryTopic, reviveQide) #从retryTopic中获取消息else:                                          #先判断是否开启v2,开启则创建v2 retryTopic,反之则创建v1 retryTopicretryTopic = KeyBuilder.buildPopRetryTopic(requestHeader.getTopic()) PopMessageProcessor.popMsgFromTopic(retryTopic, reviveQid)if requestHeader.getQueueId() < 0:               #如果请求没有指定要消费的qid，则可消费所有的队列PopMessageProcessor.popMsgFromTopic(topic,reviveQid) #popMsgFromTopic有多个重载函数，这里是reviveQid版本#就是每个队列都消费一批数据for i = 0; i < topicConfig.getReadQueueNums(); i++:queueId = (randomQ + i) % topicConfig.getReadQueueNums() #不是每次都从第一个队列开始，而是随机从第n个队列开始PopMessageProcessor.popMsgFromQueue(queueid,reviveQid)lockKey =topic+PopAckConstants.SPLIT+consumerGroup+PopAckConstants.SPLIT+queueId#创建lockkey，举个例子mytopic@myconsumer@4PopMessageProcessor.getPopOffset(reviveQid,queueid) #获取消息对应的offset，从这里开始popoffset = ConsumerOffsetManager.queryOffset(consumerGroup,topic,queueid) #先获取消费者进度if offset < 0:offset = PopMessageProcessor.getInitOffset #从已存储的且还未被清除的消息开始if checkResetOffset:                              #再检查是否其他线程已经执行了resetOffset命令#如果执行了则表明offset已经失效，需要重新设置offsetPopMessageProcessor.resetPopOffset(topic, group, queueId)  #查询并根据resetOffset设置offsetlockKey = topic + PopAckConstants.SPLIT + group + PopAckConstants.SPLIT + queueId;ConsumerOffsetManager.queryThenEraseResetOffset #查询resetOffsetMap是否存在对应的resetOffsetif resetOffset != null:                        #如果存在，则说明有人执行了resetoffsetConsumerOrderInfoManager.clearBlock(topic, group, queueId) #所以这里先清除掉该队列对应的orderInfo(不太懂)PopBufferMergeService.clearOffsetQueue(lockKey)    #popbufservice中清除该队列的信息ConsumerOffsetManager.commitOffset(group, topic, queueId, resetOffset) #重新设置该队列的offset#备注：offset就是commitOffsetif resetOffset!=nil:                            #如果设置了resetOffset，则直接返回resetOffsetreturn resetOffsetbufferOffset=PopBufferMergeService.getLatestOffset(lockKey) #看popbuf中的offset到哪了if bufferOffset < 0:                            #如果popbuf中不存在该队列return offset                                #则返回offsetelse:return Math.max(bufferOffset, offset)        #反之看哪个进度更大，就返回哪个offset                                                                                                                                                                             if !QueueLockManager.tryLock(lockKey):    #在消费一个队列的时候会先加锁，保证一个队列同一时刻只有一个消费者在消费return                                  # if isPopShouldStop:                       #判断出于inflight的消息大小是否超过了阈值，即已经pop但是还没有ackreturn                                  #是则直接返回if isOrder:                               #如果是顺序消息...待补充，因为现在还不了解啥是顺序消息...long finalOffset = offsetDefaultMessageStore.getMessageAsync        #异步获取消息......DefaultMessageStore.getMessageDefaultMessageStore.findConsumeQueue              #根据queueid获取消费队列对象  ConsumerQueueStore.findOrCreateConsumeQueue     #如果不存在会自动创建bufferConsumeQueue=ConsumeQueue.iterateFrom(nextBeginOffset, maxMsgNums)#创建迭代器，用于遍历消费者队列#一次最多允许拉取maxMsgNums条消息while (bufferConsumeQueue.hasNext()           #接下来就是遍历迭代器拉取消息了&& nextBeginOffset < maxOffset)           #maxOffset表示队列的消息的最大offset{......if messageFilter != null && !messageFilter.isMatchedByConsumeQueue: continue                                  #如果该消息被过滤掉了 就过滤掉该消息。。。废话文学   CommitLog.getMessage(offsetPy, sizePy)      #从commitLog文件读取消息  GetMessageResult.addMessage                 #保存到响应缓冲区}if diskFallRecorded:                          #记录当前队列的消费进度落后commitLog多少long fallBehind = maxOffsetPy - maxPhyOffsetPullingBrokerStatsManager.recordDiskFallBehindSize(group, topic, queueId, fallBehind) diff = maxOffsetPy - maxPhyOffsetPulling      #记算当前队列的消费进度落后commitLog多少memory =(StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE #计算可访问的内存比例* (MessageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));GetMessageResult.setSuggestPullingFromSlave(diff > memory) #如果落后程度超过了可访问内存区间，#则在response中告诉client从slave broker读取#这里的逻辑就是：master一般只想用40%的内存区域保存前40%的数据#你这个消费者在这个消费队列的进度太落后了，比如还只读到50%的数据#这样master就需要保存50%~40%这个区间内的数据#所以此时master就建议去slave读，估计是为了减少master内存压力if isOrder:...待补充...else:PopMessageProcessor.appendCheckPoint      #保存本次pop消息对应的checkPoint消息PopBufferMergeService.addCk             #是保存在内存中PopBufferMergeService.putOffsetQueue   PopBufferMergeService.commitOffsets.putIfAbsentPopBufferMergeService.buffer.put(pointWrapper.getMergeKey(), pointWrapper)queueLockManager.unLock(lockKey)            #解锁队列else:PopMessageProcessor.popMsgFromTopic(topic,queueid,reviveQid) #这里是queueid版本接直接处理该队列的消息if resp!=nil:NettyRemotingAbstract.writeResponse(request.getChannel())     #写回响应20241230 17:10 不想搞了，下班了。。。身体被掏空。。。

PopMessageProcessor.processRequest

相关文章：