编写基于事件的CQRS读取模型

关于事件源和CQRS的讨论似乎通常集中在CQRS上下文中的整体系统架构或领域驱动设计的各种形式。但是，尽管也有一些有趣的考虑，但读取模型经常被忽略。在本文中，我们将展示一个通过使用事件流填充视图模型的示例实现。 es_cqrs_projection_funnel_300_2

总览

读取模型的想法非常简单。您获取事件日志，使用适当的功能在最初为空的数据模型上应用（重放）所有事件，然后获得填充的模型。代码看起来像：

List<Event> events = getEvents();
Model model = Model.empty();
for (Event event : events) {apply(model, event);
}

通过函数式编程，我们可以使此过程更短：

Model m = reduce(getEvents(),Model.empty(),(m, e) -> apply(m, e));

这就是本质。请注意，这仅仅是抽象的轮廓，实际的实现可能会有所不同，包括缓冲，批处理（或流式传输），持久性等。

申请活动

应用事件的实际Java代码可能类似于以下内容：

EventProcessingResult processEvents() {if (getState().isRunning()) {int batchSize = getEventsPerIteration();List<Event> events = eventStore.getEventsForAllStreams(getLastEventId(),batchSize);if (events.isEmpty()) {return NO_EVENTS_TO_PROCESS;} else {return processEvents(events);}} else {return NOT_RUNNING;}
}EventProcessingResult processEvents(List<Event> events) {try {for (Event event : events) {dispatchEvent(event);}return SUCCESS;} catch (RuntimeException e) {return FAILURE;}
}

总而言之，它非常简单明了。在处理单个事件和整个批处理之前和之后，可以使用钩子来增强它。这样的钩子可以用来：

实施交易，
插入监控，
实施错误处理，
根据速度计算批次大小，
执行任意操作，例如设置某些内容或每批重新计算一次。

最后一个有趣的部分是dispatchEvent方法。除了遍历类型层次结构，错误处理并将其全部设置为可选项之外，它还可以归结为：

void dispatchEvent(Event e) {Method handler = projector.getClass().findMethod("on", e.getClass());handler.invoke(projector, e);
}

换句话说，对于每种事件类型（如OrderCreated ），我们在projector对象上寻找一个名为on的公共方法on该方法采用匹配类型的单个参数。

以上所有都是引擎的一部分，是支持许多视图模型的基础架构。实际上，实现投影所需的全部工作就是为投影仪提供有趣的事件类型的处理程序。所有其他事件都将被忽略。

它可能看起来像这样：

public class OrderProjector {@Injectprivate OrderDao orders;public void on(OrderCreated e) {orders.save(new Order(e.getOrderNumber()));}public void on(OrderApproved e) {Order o = orders.find(e.getOrderNumber());o.setApproved(true);}
}

投影线

让我们讨论一下多线程。共享的可变状态会立即带来许多问题，应尽可能避免。处理它的方法之一是首先没有并发性，例如，通过限制对单个线程的写操作。在大多数情况下，单线程写入器与ACID事务相结合足以应付写入负载。（读取/查询负载可能很重，并且使用许多线程–此处所有详细信息仅与写入有关。）

从查询事件存储到更新视图模型数据库，该线程负责将事件应用于读取的模型。通常，它只是从商店加载一批事件并应用它们。只要有更多事件要处理，它就会继续，并在捕获到事件后进入睡眠状态。在一定时间后或事件存储通知新事件时，它将唤醒。

我们还可以控制此线程的生命周期。例如，我们提供了一种以编程方式暂停和恢复每个投影线程的方法，即使在管理GUI中也是如此。

projection_admin

推还是拉？

使用数据库支持的事件存储，可以很容易地重复查询新事件。这是拉模型。不幸的是，这也意味着您可能最终会轮询过多并产生不必要的负载，或者轮询频率太低，因此可能需要更长的时间才能将更改传播到视图模型。

这就是为什么除了轮询事件存储之外，最好引入通知，以在保存新事件后立即唤醒读取的模型。这有效地成为了具有最小延迟和负载的推送模型。我们发现JGroups是完成这项工作的非常好的工具–它支持多种协议，并且易于设置，与成熟的消息队列相比，所涉及的麻烦要少得多。

通知可能包含也可能不包含实际事件。

在后一种（和更简单的）设计中，它们仅传播已保存新事件的信息及其顺序ID（以便所有预测都可以估算出它们背后的数量）。唤醒后，执行程序可以从查询事件存储开始沿其正常路径继续。

为什么？因为处理来自单一来源的事件比较容易，但是更重要的是，由于数据库支持的事件存储可轻松保证排序，并且消息丢失或重复不存在任何问题。鉴于我们正在按主键顺序读取单个表，而且大多数情况下数据仍在RAM高速缓存中，因此查询数据库的速度非常快。瓶颈在于投影线程正在更新其读取模型数据库。

但是，将事件数据放入通知中没有任何障碍（可能是出于大小或网络流量方面的考虑）。这可能会减少事件存储上的负载并节省一些数据库往返时间。投影仪将需要维护一个缓冲区，并在需要时退回查询事件存储。或者系统可以使用更可靠的消息队列。

重新开始投影

除了暂停/恢复之外，上面的屏幕截图还显示了另一项操作：重新启动。从外观上看是无害的，这是一个非常不错且功能强大的功能。

由于视图模型是完全从事件日志派生的，因此可以随时将其丢弃（从某个初始状态/足够旧的快照开始）并重新创建。在事件日志中，数据是安全的，这是事实的最终来源。

当有关视图的任何内容发生更改时，此功能很有用：添加了字段或表，修复了错误，计算出的内容有所不同。当发生这种情况时，通常从一开始就更容易（或需要），而不是例如实施大量的SQL迁移脚本。

甚至可以进行完全自动化，以便在系统启动并检测到DB模式与相应的Java模型不匹配时，它可以自动重新创建模式并重新处理事件日志。就像使用Hibernate create-drop策略运行一样，不同之处在于它不会丢失数据。

性能

该解决方案在性能方面可能看起来非常有限。

单线程作家可能引起人们的注意。实际上，单个线程通常足够快，可以轻松地跟上负载。并发不仅更难以实现和维护，而且还引入了争用。读取（查询）可以是多线程的，并且易于扩展。

通过拥有多个读取模型，例如从管理和“交易”数据中分离分析，我们也获得了很多收益。每个模型都是单线程的（用于编写），但是多个模型并行使用事件。最后，可以将解决方案修改为使用分片或某种fork-join处理。

另一个有趣的观点是从头开始重新启动投影 。

一个好的解决方案是类似于kappa体系结构：

保持过时的预测正常运行并回答所有查询。
开始新的投影，例如到另一个数据库。只要让它处理事件，就不要指向任何流量。
当新的预测赶上时，请重定向流量并关闭旧的预测。

在很小的情况下，尤其是对于开发而言，甚至可以在同一情况下在线重新启动。它取决于以下问题的答案：重新处理所有事件需要多长时间？这个投影陈旧30分钟是否可以接受？如果没人使用系统，我们可以在晚上或周末进行部署吗？我们需要重播所有历史吗？

这里要考虑的另一个因素是持久性。如果瓶颈太大，无法进一步优化，请考虑使用内存视图模型。

加起来

本质上，这就是实现使用事件存储区的读取模型所需要的全部。有了线性事件存储并在单个线程中处理所有内容，它变得非常简单。如此之多，最终实际上只是一个循环，实现了开头所示的减少。

在以后的文章中，我将更深入地探讨实施预测的实际问题。

翻译自: https://www.javacodegeeks.com/2015/09/writing-an-event-sourced-cqrs-read-model.html