RocketMQ源码解析七（Consumer拉取消息流程）

比眉伴天荒 2022-09-12 07:47 261阅读 0赞

> **RocketMQ版本4.6.0，记录自己看源码的过程**

这里基于PUSH模式和集群模式来分析消息拉取机制。  
在Consumer启动的时候，会开启一个线程，专门用来拉取消息。

/** * 拉取消息服务 */
    public class PullMessageService extends ServiceThread { 
        
        /** * 拉取请求阻塞队列 */
        private final LinkedBlockingQueue<PullRequest> pullRequestQueue = new LinkedBlockingQueue<PullRequest>();
    
        private void pullMessage(final PullRequest pullRequest) { 
            // 根据消费组名获取消费者内部实现类
            final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
            if (consumer != null) { 
                // 强制转换成DefaultMQPushConsumerImpl
                DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
                // 委托给具体的实现类实现
                impl.pullMessage(pullRequest);
            } else { 
                log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
            }
        }
    
        @Override
        public void run() { 
            log.info(this.getServiceName() + " service started");
    
            while (!this.isStopped()) { 
                try { 
                    // 从pullRequestQueue中获取消息拉取任务，如果队列为空，则阻塞；
                    // 一个消费这分配到几个消息队列，则pullRequestQueue就有可能有几个PullRequest，轮询拉取
                    PullRequest pullRequest = this.pullRequestQueue.take();
                    // 去拉取消息
                    this.pullMessage(pullRequest);
                } catch (InterruptedException ignored) { 
                } catch (Exception e) { 
                    log.error("Pull Message Service Run Method exception", e);
                }
            }
    
            log.info(this.getServiceName() + " service end");
        }
    
    }

会有两个地方将PullRequest对象放入拉取请求阻塞队列，第一次是重平衡后会为新分配的消息队列创建PullRequest对象放入该阻塞队列，后续是执行完拉取任务更新下一次拉取的位置后又将该对象重新放入该队列，这样可以不断的拉取。  
调用具体模式的消费者拉取方法：  
**DefaultMQPushConsumerImpl**

/** * 根据拉取任务拉取消息 */
    public void pullMessage(final PullRequest pullRequest) { 
        final ProcessQueue processQueue = pullRequest.getProcessQueue();
        if (processQueue.isDropped()) { 
            log.info("the pull request[{}] is dropped.", pullRequest.toString());
            return;
        }
    
        // 设置拉取时间
        pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());
    
        // 省略限流操作。。。
    
        final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
        if (null == subscriptionData) { 
            this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
            log.warn("find the consumer's subscription failed, {}", pullRequest);
            return;
        }
    
        final long beginTimestamp = System.currentTimeMillis();
    
        // 省略拉取结果的回调函数，后面再看
    
        boolean commitOffsetEnable = false;
        long commitOffsetValue = 0L;
        if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) { 
            // 当前内存中的消费进度
            commitOffsetValue = this.offsetStore.readOffset(pullRequest.getMessageQueue(), ReadOffsetType.READ_FROM_MEMORY);
            if (commitOffsetValue > 0) { 
                commitOffsetEnable = true;
            }
        }
    
        String subExpression = null;
        boolean classFilter = false;
        // topic的订阅数据
        SubscriptionData sd = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
        if (sd != null) { 
            if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) { 
                subExpression = sd.getSubString();
            }
    
            classFilter = sd.isClassFilterMode();
        }
    
        // 构建系统标志
        int sysFlag = PullSysFlag.buildSysFlag(
            commitOffsetEnable, // commitOffset
            true, // suspend
            subExpression != null, // subscription
            classFilter // class filter
        );
        try { 
            // 去broker拉取消息
            this.pullAPIWrapper.pullKernelImpl(
                pullRequest.getMessageQueue(),
                subExpression,
                subscriptionData.getExpressionType(),
                subscriptionData.getSubVersion(),
                pullRequest.getNextOffset(),
                this.defaultMQPushConsumer.getPullBatchSize(),
                sysFlag,
                commitOffsetValue,
                BROKER_SUSPEND_MAX_TIME_MILLIS,
                CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
                CommunicationMode.ASYNC,
                pullCallback
            );
        } catch (Exception e) { 
            log.error("pullKernelImpl exception", e);
            this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
        }
    }

调用消息拉取组件去拉取消息  
**PullAPIWrapper**

/** * 拉取消息 * * @param mq 拉取的队列 * @param subExpression 消息过滤表达式 * @param expressionType 消息表达式类型 * @param subVersion 版本 * @param offset 消息拉取偏移量 * @param maxNums 本次拉取的最大消息条数 * @param sysFlag 系统标记 * @param commitOffset 当前MessageQueue的消费进度 * @param brokerSuspendMaxTimeMillis 消息拉取过程允许broker挂起的时间，默认为15s * @param timeoutMillis 消息拉取超时时间 * @param communicationMode 消息拉取模式，默认为异步拉取 * @param pullCallback 消息拉取成功后的回调函数 */
    public PullResult pullKernelImpl(
        final MessageQueue mq,
        final String subExpression,
        final String expressionType,
        final long subVersion,
        final long offset,
        final int maxNums,
        final int sysFlag,
        final long commitOffset,
        final long brokerSuspendMaxTimeMillis,
        final long timeoutMillis,
        final CommunicationMode communicationMode,
        final PullCallback pullCallback
    ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
        // 先根据brokerName和brokerId得到broker地址
        FindBrokerResult findBrokerResult =
            this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
                this.recalculatePullFromWhichNode(mq), false);
        // 没找到则先从NameServer拉取路由信息，再重新获取broker地址
        if (null == findBrokerResult) { 
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
            findBrokerResult =
                this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
                    this.recalculatePullFromWhichNode(mq), false);
        }
    
        if (findBrokerResult != null) { 
            { 
                // check version
                if (!ExpressionType.isTagType(expressionType)
                    && findBrokerResult.getBrokerVersion() < MQVersion.Version.V4_1_0_SNAPSHOT.ordinal()) { 
                    throw new MQClientException("The broker[" + mq.getBrokerName() + ", "
                        + findBrokerResult.getBrokerVersion() + "] does not upgrade to support for filter message by " + expressionType, null);
                }
            }
            int sysFlagInner = sysFlag;
    
            if (findBrokerResult.isSlave()) { 
                sysFlagInner = PullSysFlag.clearCommitOffsetFlag(sysFlagInner);
            }
    
            // 构建拉取消息请求数据
            PullMessageRequestHeader requestHeader = new PullMessageRequestHeader();
            requestHeader.setConsumerGroup(this.consumerGroup);
            requestHeader.setTopic(mq.getTopic());
            requestHeader.setQueueId(mq.getQueueId());
            requestHeader.setQueueOffset(offset);
            requestHeader.setMaxMsgNums(maxNums);
            requestHeader.setSysFlag(sysFlagInner);
            requestHeader.setCommitOffset(commitOffset);
            requestHeader.setSuspendTimeoutMillis(brokerSuspendMaxTimeMillis);
            requestHeader.setSubscription(subExpression);
            requestHeader.setSubVersion(subVersion);
            requestHeader.setExpressionType(expressionType);
    
            String brokerAddr = findBrokerResult.getBrokerAddr();
            if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) { 
                brokerAddr = computPullFromWhichFilterServer(mq.getTopic(), brokerAddr);
            }
    
            // 去拉取消息，默认是异步方式，这里是返回null的
            PullResult pullResult = this.mQClientFactory.getMQClientAPIImpl().pullMessage(
                brokerAddr,
                requestHeader,
                timeoutMillis,
                communicationMode,
                pullCallback);
    
            return pullResult;
        }
    
        throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
    }

最终调到MQClientAPIImpl的异步拉取消息方法上

private void pullMessageAsync(
        final String addr,
        final RemotingCommand request,
        final long timeoutMillis,
        final PullCallback pullCallback
    ) throws RemotingException, InterruptedException { 
        this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() { 
            @Override
            public void operationComplete(ResponseFuture responseFuture) { 
                // 拉取消息的响应结果
                RemotingCommand response = responseFuture.getResponseCommand();
                if (response != null) { 
                    try { 
                        // 处理响应，将响应封装为PullResultExt
                        PullResult pullResult = MQClientAPIImpl.this.processPullResponse(response);
                        assert pullResult != null;
                        // 调用传进来的回调函数
                        pullCallback.onSuccess(pullResult);
                    } catch (Exception e) { 
                        pullCallback.onException(e);
                    }
                } else { 
                    if (!responseFuture.isSendRequestOK()) { 
                        pullCallback.onException(new MQClientException("send request failed to " + addr + ". Request: " + request, responseFuture.getCause()));
                    } else if (responseFuture.isTimeout()) { 
                        pullCallback.onException(new MQClientException("wait response from " + addr + " timeout :" + responseFuture.getTimeoutMillis() + "ms" + ". Request: " + request,
                            responseFuture.getCause()));
                    } else { 
                        pullCallback.onException(new MQClientException("unknown reason. addr: " + addr + ", timeoutMillis: " + timeoutMillis + ". Request: " + request, responseFuture.getCause()));
                    }
                }
            }
        });
    }

在成功拉取到消息后，会调用前面设置好的回调函数去消费消息，这里看下回调函数长啥样

PullCallback pullCallback = new PullCallback() { 
    
        /** * 拉取成功 * * @param pullResult 拉取结果 */
        @Override
        public void onSuccess(PullResult pullResult) { 
            if (pullResult != null) { 
                pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                    subscriptionData);
    
                switch (pullResult.getPullStatus()) { 
                    case FOUND:
                        long prevRequestOffset = pullRequest.getNextOffset(); // 当前拉取请求的队列偏移量
                        // 更新下次的拉取偏移量，以便接着用该pullRequest拉取消息
                        pullRequest.setNextOffset(pullResult.getNextBeginOffset());
                        long pullRT = System.currentTimeMillis() - beginTimestamp;
                        DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),
                            pullRequest.getMessageQueue().getTopic(), pullRT);
    
                        long firstMsgOffset = Long.MAX_VALUE;
                        // 如果拉取到的消息列表为空，则立即将pullRequest重新放回队列中，使可以进行下一次的拉取任务
                        if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) { 
                            DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                        } else { 
                            // 本次拉取的消息中第一条消息的队列偏移量
                            firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();
    
                            DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),
                                pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());
    
                            // 将消息列表存入processQueue
                            boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
                            // 将消息异步提交到consumeMessageService中的线程池供消费者消费
                            DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                                pullResult.getMsgFoundList(),
                                processQueue,
                                pullRequest.getMessageQueue(),
                                dispatchToConsume);
    
                            // 之后再将pullRequest重新放到队列中去拉取下一批消息，可以看到拉取到消息后将消息提交给消费者
                            // 消费，不管消费结果如何，接着去拉取下一批消息
                            if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) { 
                                DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
                                    DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
                            } else { 
                                DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                            }
                        }
    
                        if (pullResult.getNextBeginOffset() < prevRequestOffset
                            || firstMsgOffset < prevRequestOffset) { 
                            log.warn(
                                "[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",
                                pullResult.getNextBeginOffset(),
                                firstMsgOffset,
                                prevRequestOffset);
                        }
    
                        break;
                    // 没有拉取到消息，则将拉取请求重新放回阻塞队列中 
                    case NO_NEW_MSG:
                        pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    
                        DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
    
                        DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                        break;
                    case NO_MATCHED_MSG:
                        pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    
                        DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
    
                        DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                        break;
                    case OFFSET_ILLEGAL:
                        log.warn("the pull request offset illegal, {} {}",
                            pullRequest.toString(), pullResult.toString());
                        pullRequest.setNextOffset(pullResult.getNextBeginOffset());
    
                        pullRequest.getProcessQueue().setDropped(true);
                        DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() { 
    
                            @Override
                            public void run() { 
                                try { 
                                    DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(),
                                        pullRequest.getNextOffset(), false);
    
                                    DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());
    
                                    DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());
    
                                    log.warn("fix the pull request offset, {}", pullRequest);
                                } catch (Throwable e) { 
                                    log.error("executeTaskLater Exception", e);
                                }
                            }
                        }, 10000);
                        break;
                    default:
                        break;
                }
            }
        }
    
        @Override
        public void onException(Throwable e) { 
            if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { 
                log.warn("execute the pull request exception", e);
            }
    
            DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
        }
    };

可以看到，拉取到消息后，会先将消息存入processQueue中，ProcessQueue是MessageQueue在消费端的重现、快照。PullMessageService从消息服务器默认每次拉取32条消息，按消息的队列偏移量顺序存放在该队列中，PullMessageService然后将消息提交到消费者消费线程池，消息成功消费后从该队列中移除当消费速率过慢时，该队列中的消息可能会越来越多。

public class ProcessQueue { 
        public final static long REBALANCE_LOCK_MAX_LIVE_TIME =
            Long.parseLong(System.getProperty("rocketmq.client.rebalance.lockMaxLiveTime", "30000"));
        public final static long REBALANCE_LOCK_INTERVAL = Long.parseLong(System.getProperty("rocketmq.client.rebalance.lockInterval", "20000"));
        private final static long PULL_MAX_IDLE_TIME = Long.parseLong(System.getProperty("rocketmq.client.pull.pullMaxIdleTime", "120000"));
        private final InternalLogger log = ClientLogger.getLog();
        /** * 读写锁，控制对msgTreeMap和consumingMsgOrderlyTreeMap的并发访问 */
        private final ReadWriteLock lockTreeMap = new ReentrantReadWriteLock();
        /** * 消息存储容器，key是偏移量，MessageExt是消息实体 */
        private final TreeMap<Long, MessageExt> msgTreeMap = new TreeMap<Long, MessageExt>();
        /** * 消息总数 */
        private final AtomicLong msgCount = new AtomicLong();
        private final AtomicLong msgSize = new AtomicLong();
        /** * 消费锁，只有获取到该锁，才可以调用监听方法开始消费 */
        private final Lock lockConsume = new ReentrantLock();
        /** * 用于暂存要消费的顺序消息，key是偏移量 */
        private final TreeMap<Long, MessageExt> consumingMsgOrderlyTreeMap = new TreeMap<Long, MessageExt>();
        private final AtomicLong tryUnlockTimes = new AtomicLong(0);
        /** * 该队列中的最大偏移量 */
        private volatile long queueOffsetMax = 0L;
        private volatile boolean dropped = false;
        private volatile long lastPullTimestamp = System.currentTimeMillis();
        private volatile long lastConsumeTimestamp = System.currentTimeMillis();
        /** * 表示是否获得broker端的锁 */
        private volatile boolean locked = false;
        /** * 获得broker锁的时间 */
        private volatile long lastLockTimestamp = System.currentTimeMillis();
        private volatile boolean consuming = false;
        private volatile long msgAccCnt = 0;
    }

然后再将消息异步提交到consumeMessageService中的线程池供消费者消费  
**ConsumeMessageConcurrentlyService**

/** * 提交消费消息请求到线程池中消费 */
    @Override
    public void submitConsumeRequest(
        final List<MessageExt> msgs,
        final ProcessQueue processQueue,
        final MessageQueue messageQueue,
        final boolean dispatchToConsume) { 
        // 一次消费几条消息，默认为1条
        final int consumeBatchSize = this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
        // 如果一次拉取的消息数量<=并发消费数量，则一次性将这一批消息都放到consumeRequest中提交给消费线程池消费
        if (msgs.size() <= consumeBatchSize) { 
            ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
            try { 
                // 将消费任务提交给消费线程池消费
                this.consumeExecutor.submit(consumeRequest);
            } catch (RejectedExecutionException e) { 
                this.submitConsumeRequestLater(consumeRequest);
            }
        } else { 
            // 分批次提交，默认每次提交1条
            for (int total = 0; total < msgs.size(); ) { 
                List<MessageExt> msgThis = new ArrayList<MessageExt>(consumeBatchSize);
                for (int i = 0; i < consumeBatchSize; i++, total++) { 
                    if (total < msgs.size()) { 
                        msgThis.add(msgs.get(total));
                    } else { 
                        break;
                    }
                }
    
                ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
                try { 
                    this.consumeExecutor.submit(consumeRequest);
                } catch (RejectedExecutionException e) { 
                    for (; total < msgs.size(); total++) { 
                        msgThis.add(msgs.get(total));
                    }
    
                    this.submitConsumeRequestLater(consumeRequest);
                }
            }
        }
    }

根据consumeBatchSize参数分批次的提交给线程池执行，该消费线程池，使用的是无界队列，所以不会出现队列满后拒绝提交的情况。  
最后就不管消息消费的如何，重新将拉取请求放入到拉取请求阻塞队列，供拉取线程取出去拉取消息。  
![在这里插入图片描述][4681f0d48bc28b8eeaf7c81d0bd922b5.png_pic_center]

**参考资料**  
《儒猿技术窝——从 0 开始带你成为消息中间件实战高手》

[4681f0d48bc28b8eeaf7c81d0bd922b5.png_pic_center]: /images/20220828/18e8059a41f7448099f1fab2beeeb834.png