public void startFetchWorkers() throws TubeClientException {
        this.pushConsumer.getBaseConsumer().checkClientRunning();
        if (managerStatus.get() == 1) {
            logger.info("Duplicated call startFetchWorkers, MessageFetchManager has started !");
            return;
        }
        if (!managerStatus.compareAndSet(-1, 1)) {
            return;
        }
        StringBuilder sBuilder = new StringBuilder(256);
        logger.info("Starting Fetch Worker Pool !");
        this.fetchWorkerPool =
                new Thread[this.consumerConfig.getPushFetchThreadCnt()];
        logger.info(sBuilder
                .append("Prepare to start Fetch Worker Pool, total count:")
                .append(fetchWorkerPool.length).toString());
        sBuilder.delete(0, sBuilder.length());
        for (int i = 0; i < this.fetchWorkerPool.length; i++) {
            this.fetchWorkerPool[i] = new Thread(new FetchTaskWorker());
            this.fetchWorkerStatusMap.put(this.fetchWorkerPool[i].getId(), -1);
            this.fetchWorkerPool[i].setName(sBuilder.append("Fetch_Worker_")
                    .append(this.consumerConfig.getConsumerGroup())
                    .append("-").append(i).toString());
            sBuilder.delete(0, sBuilder.length());
        }
        for (final Thread thread : this.fetchWorkerPool) {
            thread.start();
        }
        logger.info("Fetch Worker Pool started !");
    }
Start a worker pool to fetch the message.

@throws TubeClientException   Exception thrown
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageFetchManager.java