如果我将 pollTimeout 设置为 1000 毫秒，Kafka Consumer 中会收到多少条消息答案

【问题标题】：How many messages in Kafka Consumer come if I set pollTimeout to 1000ms如果我将 pollTimeout 设置为 1000 毫秒，Kafka Consumer 中会收到多少条消息
【发布时间】：2021-06-13 18:05:34
【问题描述】：

使用 Spring Batch 实现 Kafka。开发了 Spring boot 应用程序，我的 Kafka 生产者不断产生消息。我想批量处理这些消息。但是当我触发作业时，作业会持续运行。所以我决定在KafkaItemReader中添加pollTimeout。这样我就可以停止我的工作。但是在触发 Job 时会有多少消息进入 Kafka。如果我将 pollTimeout 设置为 1000 毫秒，我无法在 google 中找到 KafkaItemReader 中会收到多少消息。

提示会很有帮助

@豆 KafkaItemReader item() { return new kafkaItemBuilder().partitions(0).consumerproperties(prop).name(“reader”).savedata(true).topic(name).pollTimeout( Duration.ofMillis(1000).build()}

【问题讨论】：

标签： spring-boot apache-kafka spring-batch spring-kafka

【解决方案1】：

批处理是关于固定数据集的。如果您的主题是连续的事件流，那么 Spring Batch 作业对您来说不是一个好的选择，流式解决方案更合适。 Spring Batch 期望您的ItemReader 在数据源耗尽时返回null，但在您的情况下，数据源永远不会耗尽，这就是您的工作永远不会完成的原因。

如果在此期间没有收到任何消息，超时属性实际上会让阅读器返回null。

【讨论】：

感谢您的回复！。 :) 我有一个问题，有没有办法让 KafkaItemReader 运行一段时间？这样它只能处理停止之前收到的消息吗？因为我的要求是触发 Spring 批处理时从 Kafka 读取消息。
当你运行你的工作时，读者应该从哪里开始阅读？从话题开始？从存储在kafka中的最后一个偏移量？从自定义偏移量？
第一次运行将从头开始。对于第二次运行，它不应从头开始，因为这些消息已被处理。所以我想从某个偏移量开始，对于第三次运行，我将从某个偏移量开始。开始处理这些消息。在这些运行期间，我的 Springboot 应用程序不会关闭。我将通过 RestController 触发作业，我将暴露它以触发 Spring 批处理。
好的，但这是一个不同的问题，您可以在此处查看如何执行此操作：stackoverflow.com/a/65882465/5019386。我相信我在这里回答了你最初的问题，所以请接受它：stackoverflow.com/help/someone-answers
感谢您的帮助！。我会调查的:)

【解决方案2】：

该属性是超时，而不是记录限制。

您可以对 max.poll.records 以及启动和停止消费者之间的时间段进行一些数学运算，但这只是一个估计值，而不是一个确切的数字，因为轮询超时只是等待的上限最大轮询记录数

如果您想以编程方式计算已处理消息的数量，我建议您获取偏移量差异或将消耗的记录数相加。

【讨论】：

嗨，我已经在我的 ConsumerProperties 中添加了这一点，但它是批处理作业正在持续运行我提到了我的 KafkaItemReader 的 Bean 定义 @Bean KafkaItemReader item() { return new kafkaItemBuilder().partitions(0).consumerproperties(prop).name(“reader”).savedata(true).topic(name).pollTimeout(Duration.ofMillis(1000).build()}
我对 Spring 还不够熟悉，无法告诉您应该使用什么，但 Kafka 通常并不意味着成为批处理框架的一部分，并且确实会持续运行