使用 kafka 进行 CQRS答案

【问题标题】：Using kafka for CQRS使用 kafka 进行 CQRS
【发布时间】：2020-03-11 23:48:43
【问题描述】：

阅读了很多关于 kafka 用作事件存储和 CQRS 潜在良好候选者的信息。我想知道，由于kafka中的消息保留时间有限，那么在kafka保留消息的磁盘中删除消息后，事件将如何重播？

从逻辑上讲，当这些消息从 kafka 外部存储（在从 kafka 主题读取消息之后）存储在 db (sql/nosql) 中时，从事件存储的角度来看，这比 kafka 更有意义。

代替上述，鉴于我的理解是正确的，即使 kafka 的实际意图只是一个高吞吐量的消息传递系统，在 CQRS 中使用 kafka 的真正用例是什么？

【问题讨论】：

这能回答你的问题吗？ Using Kafka as a (CQRS) Eventstore. Good idea?
“有限的保留时间”不正确。事实上，您可以无限保留并将其用作数据存储：It's Okay To Store Data In Kafka - Confluent
这是kafka无限期保存事件并将其用作真正的事件存储的典型用例吗？

【解决方案1】：

您可以使用事件存储和 CQRS 的 Kafka。您可以使用 Kafka Stream 处理命令生成的所有事件，并将实体的快照存储在更改日志主题中，并将更改日志主题存储在满足您要求的一个或多个 NOSQL 数据库中。此外，所有事件都可以存储在数据库（PostgresSql）中。重要的是要知道 Kafka 可以用作存储（其以高可用方式存储文件）或用作消息查询。

【讨论】：

是 kafka 流将只是一个普通的 java 应用程序，那么我也可以容器化吗？
是的。 Kafka Streams 是一个可以在 Java 应用程序中导入的库。 Kafka Streams 是 Kafka Consumer 和 Producer 类的包装器，它们使用 Kafka 集群中的数据提供流处理，如映射、过滤、连接和聚合。

【解决方案2】：

保留时间：您可以根据需要设置保留时间，甚至可以将消息永久保留在主题中。

使用 Kafka 作为数据存储：当然可以。有一个名为Log Compaction 的功能。假设以下场景：

插入 ID=10、名称=Apple、价格=10 的产品
插入 ID=20、名称=Orange、价格=20 的产品
更新 ID=10 的产品，价格变为 30

当一个主题打开日志压缩时，后台作业会定期清理该主题的消息。这项工作将检查是否有任何消息具有相同的密钥，然后只保留最终的。在上述情况下，写入 Kafka 的消息将采用以下格式：

消息 1：密钥=1，名称=Apple，价格=10
消息 2：Key=2，Name=Orange，Price=20
消息 3：Key=1，Name=Apple，Price=30（现在每个更新都包含所有字段，因此可以自包含）

日志压缩后，topic会变成：

消息 1：键=2，名称=橙色，价格=20
消息 2：Key=1，Name=Apple，Price=30（保留 ID=1 的最新记录）

实际上，Kafka 使用日志压缩功能使 Kafka 成为持久数据存储。

【讨论】：