使用Apache Kafka实现Event sourcing与CQRS

Event sourcing事件溯源作为应用程序架构模式日益普及，事件溯源将应用程序所做的状态更改建模为不可变的序列，也就是“事件日志”，也就是将触发状态更改的事件存储在不可变日志中，并将状态更改重新建模为对日志中事件的响应。这里讨论Kafka Streams将如何帮助将事件溯源和CQRS付诸实践。

我们来举个例子。考虑一个Facebook的社交网络应用程序（尽管是一个完全假设的），当用户更新个人页面，需要通知几个应用程序:
(1) 搜索应用程序：重新建立索引;
(2)?新闻源应用程序。
(3)?数据仓库ETL应用程序，将最新数据加载到支持各种分析查询等的中央数据仓库中。

事件来源包括Web应用程序的更改，将用户个人页面的更新动作建模为一种事件 - 发生过的重要事情 - 将其写入中央日志，如Kafka主题。在这种状况下，需要响应个人页面更新事件的所有应用程序，只需订阅Kafka主题并创建相应的物化视图即可 - 无论是缓存写入，索引Elasticsearch中的事件还是简单地计算内存集合，个人页面的Web应用程序本身也订阅相同的Kafka主题，并将更新写入个人页面的数据库中。

事件溯源：一些权衡
使用事件溯源构建应用程序有几个优点：它提供了对对象的每个状态进行更改的完整日志; 故障排除更容易。通过将用户意图表达为不可变事件的有序日志，事件溯源为业务提供了一个审计和合规日志，还具有提供数据来源的附加优势。它可以弹性应用; 回滚应用程序，相当于重播事件日志和重新处理数据。具有更好的性能特点; 写入和读取可以独立地缩放。它支持松散耦合的应用程序架构; 一个使得更容易走向基于微服务的架构。但最重要的是:

事件采购能够构建前向兼容的应用程序体系结构 - 能够在将来添加更多需要处理相同事件但创建不同实体化视图的应用程序。

当然缺点是：事件溯源具有较高的学习曲线; 这是一个新的和不熟悉的编程模型。事件日志可能需要更多的编程工作来查询它，因为它需要将事件转换为适合查询的所需状态。

卡夫卡作为事件溯源的骨架
事件溯源涉及维护多个应用订阅的不变得事件序列。Kafka是一种高性能，低延迟，可扩展和耐用的日志，由全球数以千计的公司使用，经过大规模的测试。因此，当转向基于事件溯源的应用程序架构时，Kafka是一个天然的存储事件的骨架。

事件溯源与CQRS
事件溯源和CQRS应用架构模式也是相关的。命令查询责任分离（CQRS）是事件溯源最常用的应用程序体系架构模式。CQRS涉及将应用程序分为两部分：命令端的命令系统实现更新状态，而查询端则是获取信息而不改变状态。CQRS提供关注的分离 - 命令或写操作方面都是关于业务; 它不关心查询，基于各种数据的不同物化视图、或物化视图的优化存储等等。另一方面，查询或读取方面都是关于读操作访问的; 其主要目的是使查询快速有效。

事件溯源与CQRS一起工作的方式是将部分应用程序的模型更新看作为对事件日志或Kafka主题的写入。同时有一个订阅该Kafka主题的事件处理程序，（根据需要）转换事件，并写入物化视图到用于读取的存储中。最后，应用程序的读取操作会针对读存储库发出查询。

CQRS有一些优点 - 它将写入和读取的负载分离，允许每个部分都独立地缩放; 此外，读取存储可以针对应用的查询模式进行优化; 比如图表应用程序可以使用Neo4j作为其读取存储，搜索应用程序可以使用Lucene索引，提供的简单内容webapp可以使用嵌入式缓存。除了技术优势，CQRS还具有组织效益 - 通过解耦写入和读取路径，您可以分离负责写入和读取路径业务逻辑的团队。

CQRS和Kafka Streams
那么，Kafka Streams是如何实现CQRS？订阅事件日志（卡夫卡的主题）的事件处理程序，将消费事件，处理这些事件，并将所发生的更新用于读取存储。在事件流上执行低延迟转换的这个过程称为流处理。在发布Apache Kafka的0.10版中，社区同时发布了Kafka Streams; 这是一个强大的流处理引擎，用于对Kafka主题进行转换建模。

Kafka Streams非常适合在使用CQRS进行事件溯源的应用程序中构建事件处理EventHandler程序组件。它是一个库，因此它可以嵌入到任何标准的Java应用程序中，实现事件流的转换。例如，这里是使用Kafka Streams进行字数计数的代码段; 您可以在Confluent示例github资源库中访问整个程序的代码
。

KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> textLines = builder.stream(stringSerde, stringSerde,"TextLinesTopic");
Pattern pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS);
KStream<String, Long> wordCounts = textLines
                                      .flatMapValues(value-> Arrays.asList(pattern.split(value.toLowerCase())))
                                      .map((key, word) -> new KeyValue<>(word, word))
                                      .countByKey("Counts")
                                      .toStream();
wordCounts.to(stringSerde, longSerde, "WordsWithCountsTopic");
KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
streams.start();

1.应用状态作为外部数据存储
Kafka Streams拓扑的输出可以是Kafka主题（如上面的示例所示）或写入外部数据存储如关系数据库。在这个范式中，事件处理程序被建模为Kafka Streams拓扑，应用程序状态被建模为用户信任和操作的外部数据存储。也就是只使用Kafka Streams作为事件处理程序，应用程序状态是外部存储，作为Kafka Streams拓扑结构的最终输出到这个外部存储。

2.应用状态存储在卡夫卡流的内部

作为另外一个候选方案，除了对事件处理程序使用Kafka Streams之外，Kafka Streams还提供了对应用程序状态进行存储的有效方法 - 它支持本地，分区和持久状态的开箱即用。这个本地状态可以是RocksDB存储，也可以是内存中的hashmap。

这样做的方式是嵌入Kafka Streams库进行有状态流处理,存储应用程序状态，作为状态存储区的分片或分区。状态存储区的分区方式与应用程序的密钥空间相同。因此，服务于特定应用程序实例的查询所需的所有数据都可以在状态存储分片中本地获取。这个本地状态库的容错由Kafka Streams提供，这是通过将所有对状态更新透明地记录到高可用性和持久的Kafka主题上实现的。可以防止应用程序实例死机，或者本地状态存储分片丢失。

实际上，Kafka Streams使用Kafka作为其本地嵌入式数据库的提交日志，传统数据库也是这样原理:事务或redo日志是事实的根源，数据表仅仅是存储在事务日志中的数据的物化视图而已。

将Kafka Streams用于构建CQRS应用程序有更多优点 - 负载平衡和故障切换也内置在Kafka Streams中; 如果一个应用程序实例失败，Kafka Streams将自动重新分配Kafka主题的分区以及其余应用程序实例中的内部状态存储分片。类似地，Kafka Streams允许弹性缩放 ; 如果启动使用Kafka Streams进行CQRS的应用程序的新实例，则会自动分配状态存储区中的现有分片以及Kafka主题的分区也均匀移动到新启动的应用程序实例中。所有这些功能都以透明的方式提供给Kafka Streams的用户。

使用Kafka Streams移植到基于CQRS的模式的应用程序不需要担心应用程序的容错能力，可用性和可伸缩性及其状态。

在即将发布的Apache Kafka版本中，Kafka Streams将允许其嵌入式状态库进行查询。在即将发布的Apache Kafka版本中，Kafka Streams将允许其嵌入式状态商店可查询。

零售库存应用程序
现在我们来举个例子来说明本文中引入的概念如何实现 - 如何使用Kafka和Kafka Streams来为应用程序启用事件溯源和CQRS。

有一个管理所有商店库存的实体零售商; 当新的货物到达或新的销售发生时，它更新库存表并知道商店库存的当前状态，可以查询库存表。

如果我们将事件溯源架构模式应用于此库存应用程序，则新的货件将在Shipment这个Kafka主题中表示为事件。同样，新的销售将被表示为Sales这个Kafka主题中的一个事件，也许是由Sales应用程序编写。为了简单起见，我们假设在销售和发货主题中的Kafka消息的key是{store id，item id}，值value是商店中商品数量总数。

库存应用程序中的事件处理程序被建模为包含销售和发货Kafka主题的Kafka Streams。创建且更新状态存储 - 库存Inventory数据表 - 表示以连续方式更新的库存的当前状态。

总结
事件溯源为应用程序提供记录状态变化提供了有效的手段。这意味着事务的恢复是简单和有效的，因为它完全基于像Kafka这样的有序日志。CQRS更进一步，它是将原始事件转化为可查询的视图; Kafka Streams提供了以流式方式创建这些查询视图所需的功能，因此用户可以直接与此视图进行交互。结果是基于Apache Kafka的事件溯源和基于CQRS的应用程序架构，将利用卡夫卡的核心竞争力 - 性能、扩展性、安全、可靠性。

最重要的是，通过这种方式构建有状态的应用程序，使公司组织能够最终得到一个松散耦合的应用程序体系架构 - 一个具有弹性和可扩展性的应用程序架构，更容易进行故障排除和升级，最重要的是向前兼容的。

Event sourcing, CQRS, stream processing and Apache