public static void main(String[] args) throws Throwable {
        // 1. get and initial 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);
        }
        final Map<String, TreeSet<String>> topicAndFiltersMap =
                MixedUtils.parseTopicParam(pubTopicAndFilterItems);

        // 2. initial configure, session factory object, and producer object
        TubeClientConfig clientConfig =
                new TubeClientConfig(masterServers);
        sessionFactory = new TubeSingleSessionFactory(clientConfig);
        messageProducer = sessionFactory.createProducer();
        messageProducer.publish(topicAndFiltersMap.keySet());

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

        // 4. 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);

        // 5. send message to server
        long sentCount = 0;
        int roundIndex = 0;
        int targetCnt = buildTopicFilterTuples.size();
        while (msgCount < 0 || sentCount < msgCount) {
            // 5.1 Rotate to get the attribute information to be sent
            roundIndex = (int) (sentCount++ % targetCnt);
            Tuple2<String, String> target = buildTopicFilterTuples.get(roundIndex);

            // 5.2 send message
            try {
                // 5.2.1 Send data asynchronously, recommended
                messageProducer.sendMessage(MixedUtils.buildMessage(target.getF0(),
                        target.getF1(), bodyData, sentCount), new DefaultSendCallback());
                // Or
                // 5.2.2 Send message synchronous, not recommended
                // MessageSentResult result = messageProducer.sendMessage(message);
                // if (!result.isSuccess()) {
                // logger.error("Sync-send message failed!" + result.getErrMsg());
                // }
            } catch (TubeClientException | InterruptedException e) {
                logger.error("Send message failed!", e);
            }

            // 5.3 Cool sending
            // Attention: only used in the test link, to solve the problem of
            // frequent sending failures caused by insufficient test resources.
            MixedUtils.coolSending(sentCount);
        }

        // 6. Clean up resources and exit the service after the task is completed
        // Attention: TubeMQ client is suitable for serving as a resident service,
        // not suitable for creating Producer objects message by message
        messageProducer.shutdown();
        sessionFactory.shutdown();
        msgSendStats.stopStats();
        logger.info("The message sending task has been completed!");
    }
Produce messages through a single-session factory instance.

@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.
inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessageProducerExample.java