从Kafka 10消费者消费消息而不提交

Question

我要求从主题中读取消息，对它们进行批处理，然后将批处理推送到外部系统。如果批处理因任何原因失败，我需要再次使用同一组消息并重复该过程。因此，对于每个批处理，每个分区的 from 和 to 偏移量都存储在数据库中。为了实现这一点，我通过向读取器分配分区来为每个分区创建一个Kafka使用者，基于先前存储的偏移量，使用者寻求该位置并开始读取。我已关闭自动提交，并且不提交来自使用者的偏移量。对于每个批处理，我为每个分区创建一个新的使用者，从存储的最后一个偏移量读取消息并发布到外部系统。您是否认为在不提交偏移量和跨批次使用相同的使用者组的情况下使用消息有任何问题，但在任何时候每个分区都不会有多个使用者？

Answer 1

在kafka版本2中，我实现了这种行为，而无需数据库来存储偏移。以下是sping-boot-kafka的配置，但它也应该适用于任何kafka消费者api

spring:
  kafka:
    bootstrap-servers: ...
    consumer:
      value-deserializer: ...
      max-poll-records: 1000
      enable-auto-commit: false
      fetch-min-size: 262144 # 1/4 mb..
      group-id: ...
      fetch-max-wait: 10000 # we will consume every 10s or when 1/4 mb or 1000 records are accumulated.
      auto-offset-reset: earliest
    listener:
      type: batch
      concurrency: 7
      ack-mode: manual

这将提供最多1000条记录的批量消息（取决于负载）。然后，我将这些记录异步写入数据库，并计算得到的成功回调次数。如果成功写入等于接收到的批大小，则确认批，例如提交偏移量。即使在高负载生产环境中，这种设计也非常可靠。

Answer 2

我觉得你的设计很合理。

将偏移量提交到Kafka只是Kafka中一种方便的内置机制，用于跟踪偏移量。但是，没有任何要求使用它 - 您也可以使用任何其他机制来跟踪偏移（例如使用数据库，就像在您的案例中一样）。

此外，如果您手动分配分区，无论如何都不会有组管理。因此参数< code>group.id不起作用。更多详情请见http://docs.confluent.io/current/clients/consumer.html。