Pull Consumer

There are two kinds of Pull methods in RocketMQ. Pull Consumer is the more primitive one, which does not provide related subscription methods, The queue should be specified to pull while calling the pull method, and it needs to update the offset itself. The other one is the Lite Pull Consumer, which provides Subscribe and Assign mode, making it more convenient to use.

Pull Consumer

The Pull Consumer example is as follows:

  1. public class PullConsumerTest {
  2.   public static void main(String[] args) throws MQClientException {
  3.     DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
  4.     consumer.setNamesrvAddr("127.0.0.1:9876");
  5.     consumer.start();
  6.     try {
  7.       MessageQueue mq = new MessageQueue();
  8.       mq.setQueueId(0);
  9.       mq.setTopic("TopicTest");
  10.       mq.setBrokerName("jinrongtong-MacBook-Pro.local");
  11.       long offset = 26;
  12.       PullResult pullResult = consumer.pull(mq, "*", offset, 32);
  13.       if (pullResult.getPullStatus().equals(PullStatus.FOUND)) {
  14.         System.out.printf("%s%n", pullResult.getMsgFoundList());
  15.         consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
  16.       }
  17.     } catch (Exception e) {
  18.       e.printStackTrace();
  19.     }
  20.     consumer.shutdown();
  21.   }
  22. }

First, the DefaultMQPullConsumer should be initialized and started, then constructs the queue MessageQueue to be pulled. Besides constructing it, fetchSubscribeMessageQueues method can also be called as shown below to get all the queues of a certain Topic and pull from the selected queue.

  1. Set<MessageQueue> queueSet =  consumer.fetchSubscribeMessageQueues("TopicTest");

After finding or constructing the queue, call the pull method to start pulling. The parameters such as the queue to be pulled, the filter expression, the offset to be pulled, and the maximum number of messages to be pulled should be passed in it. The PullResult will be returned after the operation is completed, and the PullStatus in the PullResult indicates the result status, as shown below:

  1. public enum PullStatus {
  2.     /**
  3.      * Founded
  4.      */
  5.     FOUND,
  6.     /**
  7.      * No new message can be pull
  8.      */
  9.     NO_NEW_MSG,
  10.     /**
  11.      * Filtering results can not match
  12.      */
  13.     NO_MATCHED_MSG,
  14.     /**
  15.      * Illegal offset,may be too big or too small
  16.      */
  17.     OFFSET_ILLEGAL
  18. }

FOUND means the message was pulled, NO_NEW_MSG means no new message was found, NO_MATCHED_MSG means no matching message, OFFSET_ILLEGAL means the incoming pull offset are illegal and may be large or small. If the pull status is FOUND, we can get the list of pulled messages via the getMsgFoundList method of PullResult. Finally, if the consumption is complete, the consumption offset are updated via the updateConsumeOffset method.

Lite Pull Consumer

Lite Pull Consumer is a Pull Consumer introduced in RocketMQ 4.6.0, which is simpler to use than the original Pull Consumer and provides two modes, Subscribe and Assign. The Subscribe mode example is as follows:

  1. public class LitePullConsumerSubscribe {
  2.     public static volatile boolean running = true;
  3.     public static void main(String[] args) throws Exception {
  4.         DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("lite_pull_consumer_test");
  5.         litePullConsumer.subscribe("TopicTest", "*");
  6.         litePullConsumer.setPullBatchSize(20);
  7.         litePullConsumer.start();
  8.         try {
  9.             while (running) {
  10.                 List<MessageExt> messageExts = litePullConsumer.poll();
  11.                 System.out.printf("%s%n", messageExts);
  12.             }
  13.         } finally {
  14.             litePullConsumer.shutdown();
  15.         }
  16.     }
  17. }

First of all, initialize DefaultLitePullConsumer and set ConsumerGroupName. Call the subscribe method afterward to subscribe to a topic and start it. Unlike the Push Consumer, LitePullConsumer pulls messages by the poll interface and returns the corresponding message list if it can pull the message, otherwise, it returns null. The maximum number of messages per pull can be set with setPullBatchSize, and the LitePullConsumer will automatically commits the offset by default. In the Subscribe mode, multiple LitePullConsumer under the same consumer group are load-balanced for consumption, consistent with the PushConsumer.

The following is an example of the Assign mode:

  1. public class LitePullConsumerAssign {
  2.     public static volatile boolean running = true;
  3.     public static void main(String[] args) throws Exception {
  4.         DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("please_rename_unique_group_name");
  5.         litePullConsumer.setAutoCommit(false);
  6.         litePullConsumer.start();
  7.         Collection<MessageQueue> mqSet = litePullConsumer.fetchMessageQueues("TopicTest");
  8.         List<MessageQueue> list = new ArrayList<>(mqSet);
  9.         List<MessageQueue> assignList = new ArrayList<>();
  10.         for (int i = 0; i < list.size() / 2; i++) {
  11.             assignList.add(list.get(i));
  12.         }
  13.         litePullConsumer.assign(assignList);
  14.         litePullConsumer.seek(assignList.get(0), 10);
  15.         try {
  16.             while (running) {
  17.                 List<MessageExt> messageExts = litePullConsumer.poll();
  18.                 System.out.printf("%s %n", messageExts);
  19.                 litePullConsumer.commitSync();
  20.             }
  21.         } finally {
  22.             litePullConsumer.shutdown();
  23.         }
  24.     }
  25. }

Assign mode still initializes DefaultLitePullConsumer at the beginning, here we use manual submission of offset, so set AutoCommit to false and then start the consumer. Unlike Subscribe mode, Assign mode does not have an automatic load balancing mechanism and requires the user to specify the queue to be pulled. Therefore, in the example, the queue under Topic is first fetched with fetchMessageQueues, and then half of the previous queue is fetched. The example also calls the seek method, which sets the offset in the first queue to be fetched from 10. Immediately after entering the loop keep calling the poll method to pull messages, and after pulling the messages call the commitSync method to manually submit the offset.