带有 Kafka 流 api 的 Java 代码帮助 WindowBy

Question

我有一个场景，其中有一个 Kafka 生产者流应用程序为某个主题生成 json 有效负载。 我想创建一个流应用程序，它应该对每一小时内到达该主题的每条消息进行计数。每隔一小时结束时，它应该调用一个方法，该方法应该创建一个表示计数的示例有效负载。计数可能会根据时间窗口而变化。我得到了 chatgpt 和副驾驶的帮助。 似乎什么都不起作用。需要一个示例 Java 应用程序。

非常感谢

Answer 1

如果您有多个消费者，您应该使用具有一致性的全局存储。就像 Redis 一样。