聊聊carrera的RocketMQProduceOffsetFetcher

序

本文主要研究一下carrera的RocketMQProduceOffsetFetcher

RocketMQProduceOffsetFetcher

DDMQ/carrera-monitor/src/main/java/com/xiaojukeji/carrera/monitor/lag/offset/RocketMQProduceOffsetFetcher.java

public class RocketMQProduceOffsetFetcher {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMQProduceOffsetFetcher.class);

    private DefaultMQAdminExt defaultMQAdminExt;

    private DefaultMQPullConsumer defaultMQPullConsumer;

    private String namesrvAddr;

    public RocketMQProduceOffsetFetcher(String namesrvAddr) {
        this.defaultMQAdminExt = new DefaultMQAdminExt();
        defaultMQAdminExt.setNamesrvAddr(namesrvAddr);
        defaultMQAdminExt.setInstanceName("admin-" + Long.toString(System.currentTimeMillis()));

        this.defaultMQPullConsumer = new DefaultMQPullConsumer(MixAll.TOOLS_CONSUMER_GROUP, null);
        defaultMQPullConsumer.setInstanceName("admin-" + Long.toString(System.currentTimeMillis()));
        defaultMQPullConsumer.setNamesrvAddr(namesrvAddr);
        this.namesrvAddr = namesrvAddr;
    }

    public String getNamesrvAddr() {
        return namesrvAddr;
    }

    public void start() throws MQClientException {
        defaultMQAdminExt.start();
        defaultMQPullConsumer.start();
        defaultMQPullConsumer.getDefaultMQPullConsumerImpl().getPullAPIWrapper().setConnectBrokerByUser(true);
    }

    public void shutdown() {
        defaultMQAdminExt.shutdown();
        defaultMQPullConsumer.shutdown();
    }

    public ConsumeStats getConsumeStats(String group, String topic) throws Exception {
        return defaultMQAdminExt.examineConsumeStats(group, topic);
    }

    public TopicStatsTable getProduceStats(String topic) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        return defaultMQAdminExt.examineTopicStats(topic);
    }

    public PullResult queryMsgByOffset(MessageQueue mq, long offset) throws Exception {
        return defaultMQPullConsumer.pull(mq, "*", offset, 1);
    }
}

• RocketMQProduceOffsetFetcher的构造器接收namesrvAddr，然后创建DefaultMQAdminExt及DefaultMQPullConsumer
• 其start方法会执行defaultMQAdminExt.start()、defaultMQPullConsumer.start()及defaultMQPullConsumer.getDefaultMQPullConsumerImpl().getPullAPIWrapper().setConnectBrokerByUser(true)；其shutdown执行defaultMQAdminExt.shutdown()及defaultMQPullConsumer.shutdown()
• 其getConsumeStats方法执行的是defaultMQAdminExt.examineConsumeStats(group, topic)；其getProduceStats方法执行的是defaultMQAdminExt.examineTopicStats(topic)；其queryMsgByOffset方法执行的是defaultMQPullConsumer.pull(mq, "*", offset, 1)

DefaultMQAdminExt

DDMQ/rocketmq/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java

public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
    private final DefaultMQAdminExtImpl defaultMQAdminExtImpl;
    private String adminExtGroup = "admin_ext_group";
    private String createTopicKey = MixAll.DEFAULT_TOPIC;
    private long timeoutMillis = 5000;

    //......

    @Override
    public ConsumeStats examineConsumeStats(String consumerGroup,
        String topic) throws RemotingException, MQClientException,
        InterruptedException, MQBrokerException {
        return defaultMQAdminExtImpl.examineConsumeStats(consumerGroup, topic);
    }

    @Override
    public TopicStatsTable examineTopicStats(
        String topic) throws RemotingException, MQClientException, InterruptedException,
        MQBrokerException {
        return defaultMQAdminExtImpl.examineTopicStats(topic);
    }

    //......
}

• examineConsumeStats及examineTopicStats都委托给了defaultMQAdminExtImpl

DefaultMQAdminExtImpl

DDMQ/rocketmq/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java

public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
    private final Logger log = ClientLogger.getLog();
    private final DefaultMQAdminExt defaultMQAdminExt;
    private ServiceState serviceState = ServiceState.CREATE_JUST;
    private MQClientInstance mqClientInstance;
    private RPCHook rpcHook;
    private long timeoutMillis = 20000;
    private Random random = new Random();

    //......

    @Override
    public ConsumeStats examineConsumeStats(String consumerGroup,
        String topic) throws RemotingException, MQClientException,
        InterruptedException, MQBrokerException {
        String queryTopic = topic == null ? MixAll.getRetryTopic(consumerGroup) : topic;
        TopicRouteData topicRouteData = this.examineTopicRouteInfo(queryTopic);
        ConsumeStats result = new ConsumeStats();

        for (BrokerData bd : topicRouteData.getBrokerDatas()) {
            String addr = bd.selectBrokerAddr();
            if (addr != null) {
                ConsumeStats consumeStats =
                    this.mqClientInstance.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, topic, timeoutMillis * 3);
                result.getOffsetTable().putAll(consumeStats.getOffsetTable());
                double value = result.getConsumeTps() + consumeStats.getConsumeTps();
                result.setConsumeTps(value);
            }
        }

        if (result.getOffsetTable().isEmpty()) {
            throw new MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
                "Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message");
        }

        return result;
    }

    @Override
    public TopicStatsTable examineTopicStats(
        String topic) throws RemotingException, MQClientException, InterruptedException,
        MQBrokerException {
        TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
        TopicStatsTable topicStatsTable = new TopicStatsTable();

        for (BrokerData bd : topicRouteData.getBrokerDatas()) {
            String addr = bd.selectBrokerAddr();
            if (addr != null) {
                TopicStatsTable tst = this.mqClientInstance.getMQClientAPIImpl().getTopicStatsInfo(addr, topic, timeoutMillis);
                topicStatsTable.getOffsetTable().putAll(tst.getOffsetTable());
            }
        }

        if (topicStatsTable.getOffsetTable().isEmpty()) {
            throw new MQClientException("Not found the topic stats info", null);
        }

        return topicStatsTable;
    }

    //......
}

• examineConsumeStats方法通过examineTopicRouteInfo(queryTopic)方法获取topicRouteData，然后通过topicRouteData.getBrokerDatas()获取brokerAddr，之后通过mqClientInstance.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, topic, timeoutMillis * 3)获取consumeStats；examineTopicStats方法也是先通过examineTopicRouteInfo(topic)方法获取topicRouteData，然后通过topicRouteData.getBrokerDatas()获取brokerAddr，之后通过mqClientInstance.getMQClientAPIImpl().getTopicStatsInfo(addr, topic, timeoutMillis)获取topicStatsTable

小结

• RocketMQProduceOffsetFetcher的构造器接收namesrvAddr，然后创建DefaultMQAdminExt及DefaultMQPullConsumer
• 其start方法会执行defaultMQAdminExt.start()、defaultMQPullConsumer.start()及defaultMQPullConsumer.getDefaultMQPullConsumerImpl().getPullAPIWrapper().setConnectBrokerByUser(true)；其shutdown执行defaultMQAdminExt.shutdown()及defaultMQPullConsumer.shutdown()
• 其getConsumeStats方法执行的是defaultMQAdminExt.examineConsumeStats(group, topic)；其getProduceStats方法执行的是defaultMQAdminExt.examineTopicStats(topic)；其queryMsgByOffset方法执行的是defaultMQPullConsumer.pull(mq, "*", offset, 1)

doc

• RocketMQProduceOffsetFetcher