处理卡夫卡中具有依赖关系的数据时的最佳实践？

Posted 2023-01-08

[我们正在开发一个应用程序，它从不同的来源获取数据，一旦数据可用，我们将其处理，将其放在一起，然后继续将其移至另一个主题。

在我们的案例中，我们有3个主题，每个主题都将带来与来自不同主题的数据有关联的数据，在这种情况下，每个生成的实体可能会在同一时间被接收或不被接收（或时间段内），这是问题出现的时间，因为在继续进行该主题之前，需要将这3个实体合并为一个实体。

我们的想法是创建一个单独的主题，该主题将包含所有尚未处理的数据，然后有一个单独的线程，该线程将以固定的时间间隔检查该主题，并检查此主题的依赖项是否可用，如果可用，则我们从该单独的主题中删除该实体，否则，我们将其保留在那里直到解决为止。

在所有这些解释的结尾，我的问题是，这样做是否合理，或者卡夫卡提供了其他好的做法或策略来解决这种情况？

答案

[请注意，您无法从主题中删除邮件，并且该主题带有保留，因此除非您没有对主题设置足够的保留策略，否则该邮件可以清除。

所以您需要计划将消息存储在某处：

我可以看到下面的选项，但总是每个问题都有可能的解决方法：

1. 已处理所有消息并将未处理的消息转发给其他主题，例如A
2. 从主题A分离Kafka流消耗消息并存储到状态存储中
3. 安排一个带有时间间隔的punctuate（）方法
4. 迭代所有已存储消息的存储状态。
5. 检查依赖关系（如果有）从状态存储中删除消息并进行处理，或者发布回原始主题以再次处理。

您可以参考以下链接以供参考https://kafka.apache.org/10/documentation/streams/developer-guide/processor-api.html