如何根据主题动态应用计划的kafka消费者？答案

【问题标题】：How to dinamically apply scheduled kafka consumer based on topic?如何根据主题动态应用计划的kafka消费者？
【发布时间】：2019-04-19 09:12:42
【问题描述】：

我目前正在与 kafka 上的消费者进行斗争，该消费者可以以某种方式安排未来的执行时间。

总结一下：我有一个大数据存储（.csv 文件），记录包含 2 列：时间戳和值。我正在尝试根据它们的时间戳处理这些值。第一条记录它必须由 kafka 立即使用，下一条记录应该在将来处理，延迟为“当前记录时间戳 - 前一个记录时间戳”（这不是一个很大的区别，只是几秒钟 = 结果将以毫秒为单位）等等。

所以基本上我找不到在 kafka 上实现消费者的解决方案，该消费者根据时间戳获取每条记录并使用确切的延迟。我必须模拟这些值，并且必须根据延迟将它们插入数据库中才能正常工作。

我尝试过使用执行程序来解决线程问题，但对于大数据来说，这不是一个正确的方法。

我尝试根据时间戳在生产者上创建动态主题，然后订阅它们，然后以某种方式使用队列进行处理。它没有用。

我希望 kafka 使用基于时间戳的延迟来消费每条记录。

【问题讨论】：

请注意，Kafka 接收（使用）来自Producer 的消息。 Kafka consumer 只是从 Kafka 读取消息并处理消息流，而不考虑它们之间的延迟。
如果你想延迟消费者，为什么不根据之前读取的值在消费者while循环中添加一个Thread.sleep()？
因为我有多个相同的数据集必须在系统时间戳上同时执行（或彼此非常接近）。如果我只是让线程休眠，则 2 个相同的值将具有不同的时间戳，甚至不会彼此接近。

标签： java apache-kafka

【解决方案1】：

我希望 kafka 使用基于延迟的每条记录时间戳

如果您在消息之间有特定的延迟，那么 Kafka 不是一个合适的解决方案。当您向 Kafka 发送消息时，在大多数情况下您使用网络。这可能会增加它自己的、不可预测的延迟。 Kafka 作为一个不同的进程运行，没有人能保证这个进程在什么时候准备好接收下一条消息。操作系统可以暂停进程，GC 可以启动等。这增加了另一个没人能预测的延迟。
此外，Kafka 并非设计为在收到消息时随时间运行。它更关注消息的顺序、低延迟和高吞吐量，而不是时间。

【讨论】：

感谢您的回复。是的，我也想过这个问题。好吧，你会推荐我什么解决方案？在我看来，我认为定期任务调度程序（具有固定的周期性延迟时间，例如 1 秒）检查存储，例如数据库中的表，其中包含之前保存的实体，这将是一个解决方案。
@AlexSween 不清楚你的目标是什么。可能还有其他解决方案不需要处理具有特定延迟的消息。既然是题外话，你能直接给我写信吗？
@AlexSween 可能你需要time series DB，比如 InfluxDB。