消费者示例

1 Consumer 示例

TubeMQ 提供了两种方式来消费消息： 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);
       }
   }

Back to top