批量消息的消费者偏移量更新需要示例

Question

我希望我的消费者在收到总大小为 10MB 的消息后更新 ZK 的偏移量。

有没有办法在我收到 10MB 消息后自定义我的消费者以更新偏移量？

Answer 1

首先，将auto.commit.enable 属性设置为false 以禁用消费者的自动提交行为。然后将到目前为止您收到的消息的大小保持为某个变量，如果达到 10MB，则使用ConsumerConnector 接口的commitOffsets 方法提交偏移量。然后 ZK 中的偏移量将被更新。之后将 size 变量重置为 0。