public static void main(String[] args) throws Throwable {
        // 1. get and initial parameters
        final String masterServers = args[0];
        final String subTopicAndFiterItems = args[1];
        final String groupName = args[2];
        int clientCount = Integer.parseInt(args[3]);
        if (clientCount <= 0) {
            clientCount = 1;
        }
        int paraFetchThreadCnt = -1;
        if (args.length > 5) {
            paraFetchThreadCnt = Integer.parseInt(args[4]);
        }
        final int fetchThreadCnt = paraFetchThreadCnt;
        final Map<String, TreeSet<String>> topicAndFiltersMap =
                MixedUtils.parseTopicParam(subTopicAndFiterItems);

        // 2. initial configure and build session factory object
        ConsumerConfig consumerConfig =
                new ConsumerConfig(masterServers, groupName);
        // 2.1. set consume from latest position if the consumer group is first consume
        consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_LATEST_OFFSET);
        // 2.2. set the fetch thread count of push consumer
        if (fetchThreadCnt > 0) {
            consumerConfig.setPushFetchThreadCnt(fetchThreadCnt);
        }
        // 2.3. build session factory object
        sessionFactory = new TubeSingleSessionFactory(consumerConfig);

        // 3. build and start consumer object
        for (int i = 0; i < clientCount; i++) {
            // 3.1. build and start consumer object
            PushMessageConsumer consumer =
                    sessionFactory.createPushConsumer(consumerConfig);
            // 3.2. set subscribed topic and Listener
            for (Map.Entry<String, TreeSet<String>> entry : topicAndFiltersMap.entrySet()) {
                MessageListener messageListener = new DefaultMessageListener(entry.getKey());
                consumer.subscribe(entry.getKey(), entry.getValue(), messageListener);
            }
            // 3.3 start consumer
            consumer.completeSubscribe();
            // 3.4 store consumer object
            consumerMap.put(consumer.getConsumerId(), consumer);
        }

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

        // 5. Resource cleanup when exiting the service
        //
        // 5.1 shutdown consumers
        // for (PushMessageConsumer consumer : consumerMap.values()) {
        // consumer.shutdown();
        // }
        // 5.2 shutdown session factory
        // sessionFactory.shutdown();
        // 5.3 shutdown statistic thread
        // msgRecvStats.stopStats();
    }
Consume messages in Push mode 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 subTopicAndFiterItems the topic(s) (and filter condition set) to consume on,
                      format is topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
              The 3rd parameter groupName is the name of consumer group;
              The 4th parameter clientCount is the amount of consumer;
              The 5th  parameter fetchThreadCnt is the count of fetch thread.
inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePushConsumerExample.java