apply()

本篇第2点详解；

1.10 使用快照提升性能

长生命周期的聚合可能会有大量事件；随时间推移，加载和重放这些事件会变得越来越低效；常见解决方法是定期持久保存聚合状态的快照；

1.11 幂等方式的消息处理

• 基于关系型数据库事件储存库的幂等消息处理； 可以将消息ID插入PROCESSED_MESSAGES表，作为插入EVENTS表的事件的事务的一部分 [相同消息被接受时，若数据库中已经存在ID，则忽略该消息请求]；

往往功能有限，需要使用不同的机制来实现幂等消息处理；
• 消息接收方必须以某种原子化的方式同时完成事件持久化和记录消息ID;
• 解决方案：消息消费者把消息的ID存储在处理它时生成的事件中，它通过验证聚合的所有事件中是否包含该消息ID来做重复验证；
• 该解决方案的问题：一些消息的处理可能不会生成任何事件； 一个解决方案：始终发布事件；如果聚合不发出事件，则应用程序仅保存记录消息ID的伪事件；事件接收方必须忽略这些伪事件；

1.12 领域事件的演化

事件溯源应用程序的结构分三个层次：

• 由一个或多个聚合组成；
• 定义每个聚合发出的事件；
• 定义事件的结构；

每个级别可能发生的不同类型的更改：

• 服务的领域模型随着时间的推移而发展，这些变化会自然发生；
• 不向后兼容段更改都需要更改该事件类型的消费者；

通过向上转换（Upcasting）来管理结构的变化：

• 事件溯源框架不是将事件迁移到新的版本，而是在从事件存储库加载事件时执行转换；
• 通常用称为“向上转换”的组件将各个事件从旧版本更新为更新的版本；

1.13 事件溯源的好处与弊端

好处：

• 可靠地发布领域事件；
• 保留聚合的历史；
• 最大限度地避免对象与关系的“阻抗失调”问题（持久化事件而不是聚合本身）；
• 为开发者提供一个“时光机”；

弊端：

• 这类编程模式有一定的学习曲线；
• 基于消息传递的应用程序的复杂性； 指处理非等幂事件时；
• 解决方法：为每个事件分配单调递增的ID；

指事件和快照的结构随时间推移变得臃肿；
• 解决方法：从事件存储库加载事件时，将事件升级到最新版本；即将事件版本处理与聚合的代码分开，简化聚合；

欧洲的GDPR给予用户对其数据的擦除权，而用户信息可能嵌入在事件结构中，如邮箱作为聚合的主键，应用程序必须在不删除事件的情况下清除特定用户信息；
• 解决方法：使用加密密钥；使用假名技术（如：UUID令牌代替电子邮箱作为聚合ID）；

指可能会使用嵌套的更为复杂的且可能低效的查询；
• 解决方法：参考《第7章 CQRS方法实现查询》；

2. 实现事件存储库

• 实现事件存储库有多种方法，一种是实现自己的事件存储库和事件溯源代码框架；另一种是使用专用事件存储库；
• 专用事件存储库通常提供丰富的功能集、更好的性能和可扩展性；
• 如：Event Store、Lagom、Axon、Eventuate SaaS；

2.1 Eventuate Local事件存储库的工作原理

Eventuate Local的事件数据库结构：

• events：存储事件（最核心）； 与本篇1.2点的图类似；

储存每个实体的当前版本；用于实现乐观锁；

储存每个实体的快照；
• 其支持find()、create()、update()三个操作；

通过订阅Eventuate Local的事件代理接受事件：

• 服务通过订阅事件代理来使用事件，事件代理具有每个聚合类型的主题；
• 主题是分区的消息通道，使接收方能够在保持消息排序的同时进行水平扩展；

Eventuate Local的事件中继把事件从数据库传播到消息代理：

• 事件中继将插入事件数据库的事件传播到事件代理；
• 它尽可能使用事务日志拖尾，或轮询其他数据库；
• 事件部署为独立进程；

2.2 针对Java语言的Eventuate Client框架提供的主要类和接口

图解：

• 通过ReflectiveMutableCommandProcessingAggregate类定义聚合； 该类是聚合的基类，是一个泛型类；
• 有两个类型参数：具体的聚合类、聚合命令类的超类；
• 使用反射将命令和事件分别分派给process()和apply()方法；

聚合的命令类必须扩展特定于聚合的基接口，该接口本身必须扩展Command接口；
• 如：Order聚合的命令扩展了Ordercommand；

聚合的事件类必须扩展Event接口，这是一个没有方法的标识接口；

该类是一个泛型类，它接收的参数是聚合类和聚合的基命令类；
• 提供三种重载方法：save()创建聚合、find()查找聚合、update()更新聚合；
• 主要由服务使用，在服务响应外部请求时创建和更新聚合；

Eventuate Client框架还提供了用于编写事件处理程序的API，如：

• @EventSubscriber

  注解指定持久化订阅方的ID；

• @EventHandlerMethod

  注解将creditReserved()方法标识为事件处理程序；

3. 同时使用Saga和事件溯源

3.1 使用事件溯源实现协同式Saga

• 事件溯源的事件驱动属性使得实现基于协同式的Saga非常简单；
• 当聚合被更新时，它会发出一个事件；不同聚合的事件处理程序可以接受该事件，并更新该聚合；事件溯源框架自动使每个事件处理程序具有幂等性；
• 事件溯源代码提供了Saga所需的机制，包括基于消息传递的进程间通信、消息去重，以及原子化状态更新和消息发送；
• 弊端：事件体现双重目的性，即事件溯源使用事件来表示状态更改，但使用事件实现Saga协同，需要聚合即使没有状态更改也必须发出事件； 解决方法：使用编排式来实现复杂的Saga；

3.2 创建编排式Saga

当关系型数据库作为事件存储库时，应该如何创建Saga编排器：

• 比较简单，使用

  @Transactional

  注解，使Eventuate Local框架与Eventuate Tram Saga框架在同一个ACID事务中更新时间存储库并创建Saga编排器即可；

当非关系型数据库作为事件存储库时，应该如何创建Saga编排器：

• 由于NoSQL数据库的事务模型功能有限，应用程序将无法以原子方式创建或更新两个不同的对象；
• 服务必须具有一个事件处理程序，该事件处理程序将创建Saga编排器来响应聚合发出的领域事件；

3.3 使用事件处理程序创建Saga编排器的案例

好处：保证松耦合，因为OrderService之类的服务不再明确地实例化Saga；

问题：如何处理重复事件保证幂等性，解决方法如下：

• 从事件的唯一属性中导出Saga的ID；有多种选择，其中一种是使用发出事件的聚合的ID作为Saga的ID，适用于为响应聚合创建事件而创建的Saga；
• （有效）使用事件ID作为Saga ID；因为事件ID的唯一性能保证Saga ID也是唯一的；

3.4 实现基于事件溯源的Saga参与方

• 命令式消息的幂等处理； 很容易解决：Saga参与方在处理消息时生成的事件中记录消息ID；在更新聚合之前，Saga参与方通过在事件中查找消息ID来验证它之前是否处理过该消息；

解决方法：让Saga参与方继续向Saga编排器的回复通道发送回复消息； 当Saga命令处理程序创建或更新聚合时，它会安排将SagaReplyRequested伪事件与聚合发出的实际事件一起保存在事件存储库中；
• SagaReplyRequested伪事件的事件处理程序使用事件中包含的数据构造回复消息，然后将其写入Saga编排器的回复通道；
• 如3.5点图所示；

3.5 基于事件溯源的Saga参与方的例子

3.6 实现基于事件溯源的Saga编排器

• 使用事件溯源持久化Saga编排器； 可以使用以下事件持久化Saga： SagaOrchestratorCreated：Saga编排器已创建；
• SagaOrchestratorUpdated：Saga编排器已更新；

关键在于如何以原子方式更新Saga的状态并发送命令；
• 对于NoSQL事件存储库，关键在于持久化SagaCommandEvent，它表示要发送命令；然后事件处理程序订阅SagaCommandEvents并将每个命令式消息发送到适当的通道；
• 详情请见下图：

• 确保只处理一次回复消息； 类似前面描述的机制；编排器将回复消息的ID存储在处理回复时发出的事件中；

4. 本章小结

• 事件溯源将聚合作为一系列事件持久化保存。每个事件代表聚合的创建或状态更改。应用程序通过重放事件来重建聚合的当前状态。事件溯源保留领域对象的历史记录，提供准确的审计日志，并可靠地发布领域事件；
• 快照通过减少必须重放的事件数来提高性能；
• 事件存储在事件存储库中，该存储库是数据库和消息代理的混合。当服务在事件存储库中保存事件时，它会将事件传递给订阅者；
• Eventuate Local是一个基于MySQL和Apache Kafka的开源事件存储库。开发人员使用Eventuate Client框架来编写聚合和事件处理程序；
• 使用事件溯源的一个挑战是处理事件的演变。应用程序在重放事件时可能必须处理多个事件版本。一个好的解决方案是使用向上转换，当事件从事件存储库加载时，它会将事件升级到最新版本；
• 在事件溯源应用程序中删除数据非常棘手。应用程序必须使用加密和假名等技术，以遵守欧盟GDPR等法规，确保在应用程序中彻底清除个人数据；
• 事件溯源可以很简单实现基于协调的Saga。服务具有事件处理程序，用于监听基于事件溯源的聚合发布的事件；
• 我们也可以使用事件溯源技术实现Saga编排器。你可以编写专门使用事件存储库的应用程序；

最后