RocketMQ Pull接口说明文档

订阅模式

Pull接口一共有两种订阅模式，两种订阅模式互斥。

subscribe：订阅topic，参与rebalance。
assign：用户自己指定需要订阅的队列，不参与rebalance。

重要接口说明

接口所在的类：org.apache.rocketmq.client.consumer.DefaultLitePullConsumer

void start() ：启动consumer。
void shutdown()：关闭consumer。
void subscribe(String topic, String subExpression)：订阅某topic，可多次调用订阅多个topic。与assign方法互斥。
void unsubscribe(String topic)：取消订阅某个topic。
void assign(Collection<MessageQueue> messageQueues)：手动指定需要订阅的队列。与subscribe方法互斥。此接口不允许增量指定队列，如果被调用多次会替换掉原来订阅的队列。
List<MessageExt> poll() | List<MessageExt> poll(long timeout)：向指定的topic（subscribe模式）或者手动指定的队列（assign模式）拉取消息。如果超时仍未获取到消息则返回null。
void seek(MessageQueue messageQueue, long offset)：指定某个队列下次需要消费（拉取）的位置。如果同一个队列被调用多次，以最后指定的offset为准。需要注意该接口必须在consumer start以后被调用，且参数messagequeue必须在subscribe模式下被分配到的队列中（由于rebalance需要时间，因此subscribe后马上调用seek可能会抛出异常）或者assgin模式下用户指定的队列中，否则会抛出异常。如果指定的offset超出了队列最大或最小的offset，也会抛出异常。
void pause(Collection<MessageQueue> messageQueues) ：暂停指定队列消息的拉取（不会引起rebalance）。特别注意该方法只会停止队列的拉取，但由于poll接口实现采用预拉取的方式（消息会被拉取到缓存中），因此不会马上暂停这些队列相关消息的消费，当缓存中这些队列的消息被完全消费完后，才会暂停消费。
void resume(Collection<MessageQueue> messageQueues)：恢复指定队列消息的拉取。与pause接口对应。
Long offsetForTimestamp(MessageQueue messageQueue, Long timestamp)：按时间戳查找给定队列的偏移量。与RocketMQ原来的API一致。
Collection<MessageQueue> fetchMessageQueues(String topic)：获取某个topic下所有队列的信息。注意该接口必须在consumer start后才能被调用。
registerTopicMessageQueueChangeListener(String topic, TopicMessageQueueChangeListener topicMessageQueueChangeListener)：监听某个topic队列的变换，如果队列发生变化（比如发生扩/缩容），回调Listener函数。由于定时线程检查，会有一定的延迟。
void setAutoCommit(boolean autoCommit)：设置是否自动提交消费位点。默认为自动提交。如果采用自动提交，如果发生异常或程序崩溃，不会发生消息丢失，但可能会发生重复消费。
void commitSync()：手动提交当前被分配到的队列（subscribe模式）或者assign的队列的消费位点。

注意：一些需要网络通信的接口比如seek，fetchMessageQueue等需要在start接口后被调用。

Example

具体API的用法可以参考Kafka，但也有区别。这里列出了两种订阅模式的基本用法。

subscribe用法

public class LitePullConsumerSubscribe {

    public static volatile boolean running = true;

    public static void main(String[] args) throws Exception {
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("please_rename_unique_group_name");
        litePullConsumer.subscribe("TopicTest", "*");
        litePullConsumer.start();
        try {
            while (running) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                if (messageExts != null) {
                    System.out.printf("%s%n", messageExts);
                }
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }
}

assign用法

public class LitePullConsumerAssign {

    public static volatile boolean running = true;

    public static void main(String[] args) throws Exception {
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("please_rename_unique_group_name");
        litePullConsumer.setAutoCommit(false);
        litePullConsumer.start();
        Collection<MessageQueue> mqSet = litePullConsumer.fetchMessageQueues("TopicTest");
        List<MessageQueue> list = new ArrayList<>(mqSet);
        List<MessageQueue> assignList = new ArrayList<>();
        for (int i = 0; i < list.size() / 2; i++) {
            assignList.add(list.get(i));
        }
        litePullConsumer.assign(assignList);
        litePullConsumer.seek(assignList.get(0), 10);
        try {
            while (running) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                System.out.printf("%s %n", messageExts);
                litePullConsumer.commitSync();
            }
        } finally {
            litePullConsumer.shutdown();
        }

    }
}

2020-05-27

2020-05-27

RocketMQ Pull接口说明文档

订阅模式

重要接口说明

Example