@Override
        public void run() {
            try {
                do {
                    ConsumerResult result = messageConsumer.getMessage();
                    if (result.isSuccess()) {
                        List<Message> messageList = result.getMessageList();
                        if (messageList != null && !messageList.isEmpty()) {
                            int msgCnt = messageList.size();
                            TOTAL_COUNTER.addAndGet(msgCnt);
                            AtomicLong accCount =
                                    TOPIC_COUNT_MAP.get(result.getTopicName());
                            accCount.addAndGet(msgCnt);
                        }
                        messageConsumer.confirmConsume(result.getConfirmContext(), true);
                    } else {
                        if (!TErrCodeConstants.IGNORE_ERROR_SET.contains(result.getErrCode())) {
                            logger.info(
                                    "Receive messages errorCode is {}, Error message is {}",
                                    result.getErrCode(),
                                    result.getErrMsg());
                            if (messageConsumer.isShutdown()) {
                                break;
                            }
                        }
                    }
                    if (msgConsumeCnt >= 0) {
                        if (TOTAL_COUNTER.get() >= msgConsumeCnt) {
                            break;
                        }
                    }
                } while (true);
            } catch (TubeClientException e) {
                logger.error("Create consumer failed!", e);
            }
        }
Initializes the TubeMQ consumer client(s) with the specified requirements.
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliConsumer.java