Consumer Example

1 Consumer Example

TubeMQ provides two ways to consumer message, PullConsumer and PushConsumer:

1.1 PullConsumer

public class PullConsumerExample {
    public static void main(String[] args) throws Throwable {
        final String masterHostAndPort = "localhost:8000";
        final String topic = "test";
        final String group = "test-group";
        final ConsumerConfig consumerConfig = new ConsumerConfig(masterHostAndPort, group);
        consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_LATEST_OFFSET);
        final MessageSessionFactory messageSessionFactory = new TubeSingleSessionFactory(consumerConfig);
        final PullMessageConsumer messagePullConsumer = messageSessionFactory.createPullConsumer(consumerConfig);
        messagePullConsumer.subscribe(topic, null);
        messagePullConsumer.completeSubscribe();
        // wait for client to join the exact consumer queue that consumer group allocated
        while (!messagePullConsumer.isPartitionsReady(1000)) {
            ThreadUtils.sleep(1000);
        }
        while (true) {
            ConsumerResult result = messagePullConsumer.getMessage();
            if (result.isSuccess()) {
                List<Message> messageList = result.getMessageList();
                for (Message message : messageList) {
                    System.out.println("received message : " + message);
                }
                messagePullConsumer.confirmConsume(result.getConfirmContext(), true);
            }
        }
    }
}

1.2 PushConsumer

public class PushConsumerExample {

    public static void test(String[] args) throws Throwable {
        final String masterHostAndPort = "localhost:8000";
        final String topic = "test";
        final String group = "test-group";
        final ConsumerConfig consumerConfig = new ConsumerConfig(masterHostAndPort, group);
        consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_LATEST_OFFSET);
        final MessageSessionFactory messageSessionFactory = new TubeSingleSessionFactory(consumerConfig);
        final PushMessageConsumer pushConsumer = messageSessionFactory.createPushConsumer(consumerConfig);
        pushConsumer.subscribe(topic, null, new MessageListener() {
            @Override
            public void receiveMessages(PeerInfo peerInfo, List<Message> messages) throws InterruptedException {
                for (Message message : messages) {
                    System.out.println("received message : " + new String(message.getData()));
                }
            }
            @Override
            public Executor getExecutor() {
                return null;
            }
            @Override
            public void stop() {
                //
            }
        });
        pushConsumer.completeSubscribe();
        CountDownLatch latch = new CountDownLatch(1);
        latch.await(10, TimeUnit.MINUTES);
    }
}