public void initTask() throws Exception {
        // initial consumer configure
        ConsumerConfig consumerConfig =
                new ConsumerConfig(masterServers, groupName);
        consumerConfig.setRpcTimeoutMs(rpcTimeoutMs);
        consumerConfig.setPushFetchThreadCnt(fetchThreadCnt);
        consumerConfig.setConsumePosition(consumePos);
        startTime = System.currentTimeMillis();
        // initial consumer object
        if (isPushConsume) {
            DefaultMessageListener msgListener =
                    new DefaultMessageListener();
            if (reuseConn) {
                // if reuse connection, need use TubeSingleSessionFactory class
                MessageSessionFactory msgSessionFactory =
                        new TubeSingleSessionFactory(consumerConfig);
                this.sessionFactoryList.add(msgSessionFactory);
                for (int i = 0; i < clientCount; i++) {
                    PushMessageConsumer consumer1 =
                            msgSessionFactory.createPushConsumer(consumerConfig);
                    for (Map.Entry<String, TreeSet<String>> entry : topicAndFiltersMap.entrySet()) {
                        consumer1.subscribe(entry.getKey(), entry.getValue(), msgListener);
                        TOPIC_COUNT_MAP.put(entry.getKey(), new AtomicLong(0));
                    }
                    consumer1.completeSubscribe();
                    consumerMap.put(consumer1, null);
                }
            } else {
                for (int i = 0; i < clientCount; i++) {
                    MessageSessionFactory msgSessionFactory =
                            new TubeMultiSessionFactory(consumerConfig);
                    this.sessionFactoryList.add(msgSessionFactory);
                    PushMessageConsumer consumer1 =
                            msgSessionFactory.createPushConsumer(consumerConfig);
                    for (Map.Entry<String, TreeSet<String>> entry : topicAndFiltersMap.entrySet()) {
                        consumer1.subscribe(entry.getKey(), entry.getValue(), msgListener);
                        TOPIC_COUNT_MAP.put(entry.getKey(), new AtomicLong(0));
                    }
                    consumer1.completeSubscribe();
                    consumerMap.put(consumer1, null);
                }
            }
        } else {
            if (reuseConn) {
                MessageSessionFactory msgSessionFactory =
                        new TubeSingleSessionFactory(consumerConfig);
                this.sessionFactoryList.add(msgSessionFactory);
                for (int i = 0; i < clientCount; i++) {
                    PullMessageConsumer consumer2 =
                            msgSessionFactory.createPullConsumer(consumerConfig);
                    for (Map.Entry<String, TreeSet<String>> entry : topicAndFiltersMap.entrySet()) {
                        consumer2.subscribe(entry.getKey(), entry.getValue());
                        TOPIC_COUNT_MAP.put(entry.getKey(), new AtomicLong(0));
                    }
                    consumer2.completeSubscribe();
                    consumerMap.put(consumer2,
                            new TupleValue(consumer2, msgCount, fetchThreadCnt));
                }
            } else {
                for (int i = 0; i < clientCount; i++) {
                    MessageSessionFactory msgSessionFactory =
                            new TubeMultiSessionFactory(consumerConfig);
                    this.sessionFactoryList.add(msgSessionFactory);
                    PullMessageConsumer consumer2 =
                            msgSessionFactory.createPullConsumer(consumerConfig);
                    for (Map.Entry<String, TreeSet<String>> entry : topicAndFiltersMap.entrySet()) {
                        consumer2.subscribe(entry.getKey(), entry.getValue());
                        TOPIC_COUNT_MAP.put(entry.getKey(), new AtomicLong(0));
                    }
                    consumer2.completeSubscribe();
                    consumerMap.put(consumer2,
                            new TupleValue(consumer2, msgCount, fetchThreadCnt));
                }
            }
        }
        isStarted = true;
    }
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