public static void main(String[] args) throws Throwable {
        // 1. get call parameters
        final String masterServers = args[0];
        final String pubTopicAndFilterItems = args[1];
        final long msgCount = Long.parseLong(args[2]);
        int pkgSize = 1024;
        if (args.length > 3) {
            pkgSize = MixedUtils.mid(Integer.parseInt(args[3]), 1, 1024 * 1024);
        }
        int clientCnt = 2;
        if (args.length > 4) {
            clientCnt = MixedUtils.mid(Integer.parseInt(args[4]), 1, 100);
        }
        int sessionFactoryCnt = 10;
        if (args.length > 5) {
            sessionFactoryCnt = MixedUtils.mid(Integer.parseInt(args[5]), 1, 20);
        }
        final Map<String, TreeSet<String>> topicAndFiltersMap =
                MixedUtils.parseTopicParam(pubTopicAndFilterItems);

        // 2. build multi-session factory
        TubeClientConfig clientConfig = new TubeClientConfig(masterServers);
        for (int i = 0; i < sessionFactoryCnt; i++) {
            sessionFactoryProducerMap.put(i,
                    new Tuple2<>(new TubeMultiSessionFactory(clientConfig), new HashSet<>()));
        }

        // 3. build multi-thread message sender
        sendExecutorService =
                Executors.newFixedThreadPool(clientCnt, new ThreadFactory() {

                    @Override
                    public Thread newThread(Runnable runnable) {
                        return new Thread(runnable);
                    }
                });

        // 4. initial and statistic thread
        Thread statisticThread =
                new Thread(msgSendStats, "Sent Statistic Thread");
        statisticThread.start();

        // 5. build the content of the message to be sent
        // include message body, attributes, and time information template
        final byte[] bodyData =
                MixedUtils.buildTestData(pkgSize);
        List<Tuple2<String, String>> buildTopicFilterTuples =
                MixedUtils.buildTopicFilterTupleList(topicAndFiltersMap);

        // 6. Rotating build and start producers in the session factory list
        // In the same process, different TubeMultiSessionFactory objects can create
        // independent connections for the same Broker.
        // We increase the concurrent throughput of the system by increasing the
        // number of links. Here we distribute the clients evenly on
        // different TubeMultiSessionFactory objects to
        // improve data production performance in the single process
        for (int indexId = 0; indexId < clientCnt; indexId++) {
            Tuple2<MessageSessionFactory, Set<MessageProducer>> sessionProducerMap =
                    sessionFactoryProducerMap.get(indexId % sessionFactoryCnt);
            MessageProducer producer = sessionProducerMap.getF0().createProducer();
            producer.publish(topicAndFiltersMap.keySet());
            sessionProducerMap.getF1().add(producer);
            sendExecutorService.submit(new Sender(indexId,
                    producer, bodyData, buildTopicFilterTuples, msgCount));
        }

        // 7. wait all tasks finished
        try {
            // wait util sent message's count reachs required count
            long needSentCnt = msgCount * clientCnt;
            while (totalSentCnt.get() < needSentCnt) {
                logger.info("Sending task is running, total = {}, finished = {}",
                        needSentCnt, totalSentCnt.get());
                Thread.sleep(30000);
            }
        } catch (Throwable e) {
            logger.error("Throwable: ", e);
        }
        // clear resources
        sendExecutorService.shutdownNow();
        for (int i = 0; i < sessionFactoryCnt; i++) {
            sessionFactoryProducerMap.get(i).getF0().shutdown();
        }
        msgSendStats.stopStats();
        logger.info("Sending task is finished, total sent {} messages", totalSentCnt.get());
    }
Produce messages through multi-session factory instances.

@param args   Startup parameter array, including the following parts:
              The 1st parameter masterServers is the master address(es) to connect to,
                      format is master1_ip:port[,master2_ip:port];
              The 2nd parameter pubTopicAndFilterItems is the topic(s) (and filter condition set) to publish to,
                      format is topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
              The 3rd parameter msgCount is the message amount that needs to be sent;
              The 4th parameter pkgSize is the message's body size that needs to be sent;
              The 5th parameter clientCount is the amount of producer;
              The 6th parameter sessionFactoryCnt is the amount of session factory.
inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MAMessageProducerExample.java