java - 使用spring kafka在kafka消费者中手动提交偏移量的方法

Question

我正在尝试找出使用 Spring-Kafka (1.1.0.RELEASE) 在 Kafka 消费者中手动提交偏移量的方法。我知道，最好为健壮的客户端实现提交这些偏移量，这样其他消费者就不会处理重复的事件，这些事件最初可能由现已死亡的消费者处理，或者因为触发了再平衡。

我看到有两种方法可以处理这个 -

• 将 ACK_MODE 设置为 MANUAL_IMMEDIATE 并在监听器实现中调用 ack.acknowledge() API

  ConcurrentKafkaListenerContainerFactory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
  
  @KafkaListener(topics = "anotherBatchTopic", containerFactory = "batchContainerFactory")
  public void listen(List<ConsumerRecord<byte[], String>> batch, Acknowledgment acknowledgment) throws Exception {
      logger.info("===*** batch size : " + batch.size() + "***===");
      batch.forEach(System.out::println);
      acknowledgment.acknowledge();
  }
  

• 实现 ConsumerRebalanceListener。

  ConcurrentKafkaListenerContainerFactory.getContainerProperties().setConsumerRebalanceListener(new ConsumerRebalanceListener() {
  
  @Override
  public void onPartitionsRevoked(final Collection<TopicPartition> collection) {
  
  }
  
  @Override
  public void onPartitionsAssigned(final Collection<TopicPartition> collection) {
  
  }
  });
  

但是，使用这种方法，我不知道如何获得对消费者的引用，以调用 consumer.commitSync() 或 consumer.commitASync() API。由于一些技术限制，我无法移动到支持 ConsumerAwareRebalanceListener 并引用了 Consumer 的最新版本的 spring-kafka。

那么如何利用 ConsumerRebalanceListener 将偏移量提交给 Kafka 呢？

此外，我正在使用 ConcurrentKafkaListenerContainerFactory.setConcurrency() API 启动多个消费者监听器线程，因此如果特定消费者线程终止，它是否有自己的 ConsumerRebalanceListener 实例？

Answer 1

消费者再平衡监听器不能用于提交偏移量；与 1.x;唯一的方法是通过 Acknowledgment 参数。

1.1.0 已经很老了；建议所有 1.x 用户至少升级到 1.3.5；由于 KIP-62，它具有更简单的线程模型 - 请参阅 the project page .

当前版本 2.1.6(2.1.7 将于今天发布)有更多选项，包括 ConsumerAwareMessageListener，您可以在其中获得对消费者的完全访问权限。

您不能直接与版本 < 1.3 的监听器的使用者交互，因为使用者不是线程安全的，必须在不同的线程上调用监听器。