踩坑记：rocketmq-console 消费TPS为0，但消息积压数却在降低是个什么“鬼”-深蓝源码网

1、背景

当消息积压后，消费端将其代码进行优化后，重启消费端服务器，从rocketmq-console上发现TPS为0。如图所示。

乍一看，第一时间得出应用还未恢复，就开始去查看相关的启动日志,通常查看的是应用服务器的 /home/baseuser/logs/rockemqlogs/rocketmq_client.logs，碰巧又看到如下的错误日志：

RebalanceService - [BUG] ConsumerGroup: consumergroup_1 The consumerId: consumer-client-id-clusterA-192.168.x.x@ not in cidAll: [consumer-client-id-clusterA-192.168.x.x@]

上面的日志显示在队列负载时候，当前节点竟然不属于 consumergroup_1 消费组的活跃连接，导致一大片的报错：

-11-02 19:29:17 WARN NettyClientPublicExecutor_1 - execute the pull request exception
org.apache.rocketmq.client.exception.MQBrokerException: CODE: 25  DESC: the consumer's subscription not latest
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.processPullResponse(MQClientAPIImpl.java:639)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.access$200(MQClientAPIImpl.java:156)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl$2.operationComplete(MQClientAPIImpl.java:592)
    at org.apache.rocketmq.remoting.netty.ResponseFuture.executeInvokeCallback(ResponseFuture.java:51)
    at org.apache.rocketmq.remoting.netty.NettyRemotingAbstract$2.run(NettyRemotingAbstract.java:275)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

乍一看确实是 rocketmq 相关的问题，导致上述 消费TPS 为0，经过半个小时的日志分析，发现这是RocketMQ 这是一种正常现象，最终会自动恢复，这里我留一个伏笔，将在我的知识星球中与广大星友讨论，经过日志分析得出 rocketmq 没问题，故后面去查看消息积压，发现消息积压明显在减少，那这就奇了怪了，咋消息积压在快速减少，但为啥消费TPS还是为0呢？

接下来将该问题进行探讨。

温馨提示：在问题分析部分，作者没有直接给出答案，而是一步一步探寻答案，因此会通过追踪源码来寻求答案，如果大家想急于答案，可以跳过问题分析，直接查看本文末尾的问题解答部分。 通过本文的阅读，您将获得如下信息： 1、RocketMQ 消费TPS的收集与计算逻辑。 2、RocketMQ 监控指标的设计思路。 3、RocketMQ 主从同步，消费者从主服务器拉取还是从从服务器拉取的判断逻辑。

2、问题分析

2.1 rocketmq-console 数据获获取逻辑探讨

要解开消费TPS 显示为０的问题，我们首先要来看一下 rocketmq-console 这个页面的展示逻辑，即通过阅读 rocketmq-console的源码来解开其采集逻辑。

得知，【消费者】界面查询各个消费组的基本信息的接口为 /consumer/groupList.query，那接下来，我们首先从源码的角度来分析该接口的实现逻辑。其入口如下：

org.apache.rocketmq.console.controller.ConsumerController#list
@RequestMapping(value = "/groupList.query")
@ResponseBody
public Object list() {
    return consumerService.queryGroupList();
}
就是调用消费服务处理类的 queryGroupList 方法，其实现代码如下：
ConsumerServiceImpl＃queryGroupList
public List<GroupConsumeInfo> queryGroupList() {
    Set<String> consumerGroupSet = Sets.newHashSet();
    try {
        ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();　　// @1
        for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {   // @2
            SubscriptionGroupWrapper subscriptionGroupWrapper = mqAdminExt.getAllSubscriptionGroup(brokerData.selectBrokerAddr(), 3000L);  // @3
            consumerGroupSet.addAll(subscriptionGroupWrapper.getSubscriptionGroupTable().keySet());                                                                 
        }
    } catch (Exception err) {
        throw Throwables.propagate(err);
    }
    List<GroupConsumeInfo> groupConsumeInfoList = Lists.newArrayList();
    for (String consumerGroup : consumerGroupSet) {                                                // @4
        groupConsumeInfoList.add(queryGroup(consumerGroup));                              
    }
    Collections.sort(groupConsumeInfoList);
    return groupConsumeInfoList;
}

代码@1：获取集群的 broker 信息，主要是通过向 NameServer 发送 GET_BROKER_CLUSTER_INFO 请求，NameServer 返回集群包含的所有 broker 信息，包含从节点的信息，返回的格式如下：

"clusterInfo": {
    "brokerAddrTable": {
       "broker-a": {
           "cluster": "DefaultCluster",
            "brokerName": "broker-a",
            "brokerAddrs": {
                "0": "192.168.0.168:10911",
                "1": "192.168.0.169:10911"
            }
        },
        "broker-b": {
           "cluster": "DefaultCluster",
            "brokerName": "broker-b",
            "brokerAddrs": {
                "0": "192.168.0.170:10911",
                "1": "192.168.1.171:10911"
            }
        }
    },
    "clusterAddrTable": {
        "DefaultCluster": ["broker-a","broker-b"]
    }
}

代码@2：遍历集群中的 brokerAddrTable 数据结构，即存储了 broker 的地址信息的 Map 。

代码@3：分别向集群中的主节点(brokerData.selectBrokerAddr()) 获取所有的订阅关系（即消费组的订阅信息）。然后将所有的消费者组名称存入 consumerGroupSet。

代码@4：遍历代码@3收集到的消费组，调用 queryGroup 依次请求消费组的运行时信息，后面接下来详细分析。

接下来将重点分析 queryGroup方法的实现细节。

ConsumerServiceImpl#queryGroup

public GroupConsumeInfo queryGroup(String consumerGroup) {
    GroupConsumeInfo groupConsumeInfo = new GroupConsumeInfo();
    try {
        ConsumeStats consumeStats = null;
        try {
            consumeStats = mqAdminExt.examineConsumeStats(consumerGroup);  // @1
        } catch (Exception e) {
            logger.warn("examineConsumeStats exception, " + consumerGroup, e);
        }
        ConsumerConnection consumerConnection = null;
        try {
            consumerConnection = mqAdminExt.examineConsumerConnectionInfo(consumerGroup); 
        } catch (Exception e) {
            logger.warn("examineConsumerConnectionInfo exception, " + consumerGroup, e);
        }
        groupConsumeInfo.setGroup(consumerGroup);

        if (consumeStats != null) {
            groupConsumeInfo.setConsumeTps((int)consumeStats.getConsumeTps());    // @2
            groupConsumeInfo.setDiffTotal(consumeStats.computeTotalDiff());                   // @3
        }

        if (consumerConnection != null) {
            groupConsumeInfo.setCount(consumerConnection.getConnectionSet().size());
            groupConsumeInfo.setMessageModel(consumerConnection.getMessageModel());
            groupConsumeInfo.setConsumeType(consumerConnection.getConsumeType());
            groupConsumeInfo.setVersion(MQVersion.getVersionDesc(consumerConnection.computeMinVersion()));
        }
    } catch (Exception e) {
        logger.warn("examineConsumeStats or examineConsumerConnectionInfo exception, "
                + consumerGroup, e);
    }
    return groupConsumeInfo;
}

从上面@1，@2，@3这三处代码可以得知，rocketmq-console 相关界面上的消费TPS主要来自 examineConsumeStats 方法，该方法我就不再继续深入，我们只需找到该方法向 broker 发送的请求编码，然后根据该请求编码找到 broker 的处理逻辑即可，最后跟踪发送的请求编为：RequestCode.GET_CONSUME_STATS。

GET_CONSUME_STATS 命令在 broker 的处理逻辑如下：

AdminBrokerProcessor#getConsumeStats

private RemotingCommand getConsumeStats(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final GetConsumeStatsRequestHeader requestHeader =
            (GetConsumeStatsRequestHeader) request.decodeCommandCustomHeader(GetConsumeStatsRequestHeader.class);
        ConsumeStats consumeStats = new ConsumeStats();
        Set<String> topics = new HashSet<String>();
        if (UtilAll.isBlank(requestHeader.getTopic())) {
            topics = this.brokerController.getConsumerOffsetManager().whichTopicByConsumer(requestHeader.getConsumerGroup());
        } else {
            topics.add(requestHeader.getTopic());
        }
        for (String topic : topics) {   // @1
            TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topic);
            if (null == topicConfig) {  // @2
                log.warn("consumeStats, topic config not exist, {}", topic
						 技术沙龙 教程文章 热点综合