private void pushConsumer(MessageSessionFactory messageSessionFactory, ConsumerConfig consumerConfig)
                throws TubeClientException, InterruptedException {
            messagePushConsumer = messageSessionFactory.createPushConsumer(consumerConfig);
            messagePushConsumer.subscribe(topicName, null, new MessageListener() {

                @Override
                public void receiveMessages(PeerInfo peerInfo, List<Message> messages) throws InterruptedException {
                    for (Message message : messages) {
                        System.out.println(new String(message.getData()));
                        msgCount.getAndIncrement();
                    }
                }

                @Override
                public Executor getExecutor() {
                    return null;
                }

                @Override
                public void stop() {
                }
            });
            messagePushConsumer.completeSubscribe();
            CountDownLatch latch = new CountDownLatch(1);
            latch.await(10, TimeUnit.MINUTES);
        }
Create a pushConsumer and consume messages

@param messageSessionFactory
@param consumerConfig
@throws TubeClientException
@throws InterruptedException
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/MessageCommand.java