public void initTask() throws Exception {
        // initial client configure
        TubeClientConfig clientConfig = new TubeClientConfig(masterServers);
        clientConfig.setRpcTimeoutMs(rpcTimeoutMs);
        // initial sent data
        sentData = MixedUtils.buildTestData(msgDataSize);
        // initial topic send round
        topicSendRounds = MixedUtils.buildTopicFilterTupleList(topicAndFiltersMap);
        startTime = System.currentTimeMillis();
        // initial send thread service
        sendExecutorService =
                Executors.newFixedThreadPool(sendThreadCnt, new ThreadFactory() {

                    @Override
                    public Thread newThread(Runnable runnable) {
                        return new Thread(runnable, "sender_" + producerMap.size());
                    }
                });
        // initial producer object
        if (reuseConn) {
            // if resue connection, use TubeSingleSessionFactory class
            MessageSessionFactory msgSessionFactory =
                    new TubeSingleSessionFactory(clientConfig);
            this.sessionFactoryList.add(msgSessionFactory);
            for (int i = 0; i < clientCount; i++) {
                MessageProducer producer = msgSessionFactory.createProducer();
                producer.publish(topicAndFiltersMap.keySet());
                producerMap.put(producer, new MsgSender(producer));
                // send send task
                sendExecutorService.submit(producerMap.get(producer));
            }
        } else {
            for (int i = 0; i < clientCount; i++) {
                // if not resue connection, use TubeMultiSessionFactory class
                MessageSessionFactory msgSessionFactory =
                        new TubeMultiSessionFactory(clientConfig);
                this.sessionFactoryList.add(msgSessionFactory);
                MessageProducer producer = msgSessionFactory.createProducer();
                producer.publish(topicAndFiltersMap.keySet());
                producerMap.put(producer, new MsgSender(producer));
                // send send task
                sendExecutorService.submit(producerMap.get(producer));
            }
        }
        isStarted = true;
    }
Initializes the TubeMQ producer client(s) with the specified requirements.
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliProducer.java