如何阅读Kafka主题中的所有记录

Question

我正在使用kafka:Kafka2.12-2.1.0，在客户端使用spring kafka，但遇到了一个问题。

我需要通过阅读Kafka主题中的所有现有消息来加载内存中的映射。为此，我启动了一个新的使用者（具有唯一的使用者组id，并将偏移量设置为最早)。然后我迭代使用者（poll方法）以获取所有消息，并在使用者记录变为空时停止。

我尝试了很少的其他方法（比如使用偏移量数），但还没有找到任何解决方法，除了在某个地方保留另一个记录，告诉我有多少消息在主题中需要阅读之前，我停下来。

有什么好主意吗？

Answer 1

根据我的理解，您试图实现的是在您的应用程序中构建一个基于特定主题中已经存在的值的映射。

对于此任务，您可以使用Kafka Streams DSL中的Ktable，而不是手动轮询主题，它将自动构造一个可读的键值存储，该存储具有容错、启用复制并自动填充新值。

您只需在流上调用groupByKey，然后使用聚合即可完成此操作。

KStreamBuilder builder = new KStreamBuilder();
KStream<String, Long> myKStream = builder.stream(Serdes.String(), Serdes.Long(), "topic_name");
KTable<String, Long> totalCount = myKStream.groupByKey().aggregate(this::initializer, this::aggregator);

点击此处阅读更多关于Kafaka流概念的内容

然后我迭代使用者（poll方法）以获取所有消息，并在使用者记录变为空时停止

Kafaka是一个消息流平台。您流式传输的任何html" target="_blank">数据都在持续更新，您可能不应该以期望在一定数量的消息之后停止消费的方式使用它。如果在您阻止消费者后有新消息进来，您将如何处理？

您在这里的具体用例是什么？可能有一个很好的方法来使用Kafaka语义本身。