Spring Kafka 多并发消费者 - 提交偏移量失败答案

【问题标题】：Spring Kafka Multi Concurrency Consumer - Commit offset FailedSpring Kafka 多并发消费者 - 提交偏移量失败
【发布时间】：2020-08-08 05:21:43
【问题描述】：

我正在使用 Spring 和 Kafka 来使用来自 Kafka 主题的数据。我已将并发配置为 10。因此不同的线程轮询代理以获取消息并处理消息。即使经过一段时间（成功处理），我们还是会收到相同的消息返回给消费者的不同线程。我们能够在配置的 max.poll.interval.ms=1500000 内处理接收到的消息。

请找到以下配置的 Kafka 消费者属性。我已经通过 Kafka 配置了自动提交。

    group.id=ips-request-group //group id
    enable.auto.commit=true // auto commit
    auto.commit.interval.ms=60000 // auto commit inverval
    session.timeout.ms=100000 // session time out
    request.timeout.ms=610000 // request time out
    fetch.max.wait.ms=500  //polling time out
    heartbeat.interval.ms=20000 // heart beat interval
    auto.offset.reset=latest  //consuming latest messages.
    key.deserializer=org.apache.kafka.common.serialization.StringDeserializer // key
    value.deserializer=org.apache.kafka.common.serialization.StringDeserializer //value
    max.poll.records=10 // max polling records
    max.poll.interval.ms=1500000 // max polling interval ms /*

您能否帮我解决重复接收消息给 Kafka 消费者的问题。

【问题讨论】：

您需要提供更多信息；你怎么知道你有重复？是否由于其他原因（例如，另一个成员加入）发生了重新平衡。我建议永远不要使用 auto.commit - 侦听器容器以更可预测的方式进行提交（在每个 RECORD 或 BATCH 之后）更可靠。
嗨，Gary，当有任何消息出现时，我们正在验证。特定事务是否存在或不存在数据库。如果事务已经存在，我们将插入带有一些描述的跟踪器表。我们不能通过自动间隔提交消息偏移量吗？
也许问题出在出版方面。您的消费者也应该将分区和偏移量存储在数据库中；如果没有发生重新平衡，我看不出你怎么可能得到一个副本。
即使我们及时处理消息，我们也会在日志中收到以下消息。警告 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - 自动提交偏移量 ced 并将分区分配给另一个成员。这意味着后续调用 poll() 之间的时间比配置的 max.poll.interval.ms 长，这通常意味着轮询循环花费了太多时间处理消息。您可以通过增加会话超时或使用 max.poll.records 减少 poll() 中返回的批处理的最大大小来解决此问题。
这意味着重新平衡确实发生了；您需要在您的消费者中添加重复数据删除逻辑来处理这种情况。您需要减少max.poll.records 或增加max.poll.interval.ms。

标签： spring apache-kafka spring-kafka

【解决方案1】：

在处理每条记录后提交偏移量；设置auto.commit.enabled=false并将容器属性ackMode设置为AckMode.RECORD。

【讨论】：

目前我将确认模式设置为 AckMode.MANUAL 以手动提交偏移量，并发为 10 或更多。我正在使用 BatchMessageListener。我在侦听器中收到消息列表。你能帮我在处理后如何提交每条特定消息的偏移量吗？ public void onMessage(List> 记录) { records.forEach(action -> { this.messageListener.process(action.value()); }); }
使用批处理监听器并提交单个偏移量没有多大意义；改为使用记录侦听器。如果您真的想这样做，请添加 Consumer<?, ?> 作为参数并以这种方式手动提交每个偏移量。提交必须在侦听器线程上执行。