apache-pulsar - 通过 Pulsar 消费者批量读取消息

Question

我正在使用节点 pulsar 客户端来使用来自 Pulsar 主题的消息。消费者使用shared订阅模式订阅主题。目前，每次调用receive都会从主题中获取一条消息。有没有办法批量接收消息？

Answer 1

一条一条地获取消息并不意味着 Pulsar 客户端没有在后台使用批处理和其他优化技术。Pulsar Java 消费者的官方文档receiverQueueSize定义了定义消息累积的参数。默认情况下，Pulsar 消费者为其参数使用合理的值，并且对于大多数应用程序来说它应该表现得相当好。您是否遇到任何问题或性能缓慢？

更新

从 Apache Pulsar 2.4.1 版本开始，可以使用 consumer 批量接收消息。首先，应该使用BatchReceivePolicy配置创建消费者（将值更改为更适合您的用例）：

Consumer<GenericRecord> consumer = pulsarClient
        .newConsumer(Schema.AUTO_CONSUME())
        .batchReceivePolicy(BatchReceivePolicy.builder()
            .maxNumMessages(5000)
            .maxNumBytes(10 * 1024 * 1024) 
            .timeout(1, TimeUnit.SECONDS).build())
// .. other configuration such as topic and subscription

二、使用batchReceive方法获取一批消息：

Messages<GenericRecord> messages = consumer.batchReceive();

处理完所有消息后，只需确认所有消息：

consumer.acknowledge(messages);