问题描述
我很难理解如何连接到默认的 Axon 事件存储并查看那里存储了哪些事件?有没有办法以某种方式预览它?
我如何重放存储的事件以重新创建存储实体的特定状态?我试图在互联网上找到示例、教程或视频课程,但找不到任何东西 ?... 如何制作快照然后检索它?找不到任何关于如何操作的在线示例... 如果之前有人问过这个问题,有人可以建议或分享 StackOverflow 上讨论的链接吗?
谢谢?
解决方法
我觉得这里有三个问题。 首先是标题,其次是否有任何教程,其次是如何重新创建实体。为了清楚起见,最好在此之外创建单独的问题,但无论如何我都会回答它们。
1.读取存储在 Axon 事件存储中的事件
如果您使用 [Axon Server][1],阅读事件就像浏览 Axon Server 的仪表板一样简单。从那里你可以点击“搜索”按钮,打开搜索盘,让你检查所有已存储的事件。那里有一个充实的查询语言,详细解释[这里][2]。如果您不使用 Axon Server,那么您的事件要么存储在 RDBMS 中,要么存储在 MongoDb 中。如果您想知道如何读取,我建议您阅读所使用的 RDBMS 实例的文档或 MongoDb 关于如何查询表的文档。
2.轴突教程材料
我假设简单地搜索“轴突编码示例”*应该*为您提供大量信息,但让我在这个答案中为您提供一些信息:- Axon Reference Guide
- Axon Intro Video Series
- Axon Hotel Demo
- Baeldung's Axon Article
- Axon's Intro Training
- Domain Driven Design
- Command Query Responsibility Segregation
- Event Sourcing
3.在 Axon 中重新创建模型
现在,当谈到读取事件时,几乎在所有情况下,您都*永远*不必使用 Axon 框架手动执行此操作。 您只需使用 `@EventHandler` 批注将一个方法标记为能够处理事件,框架将自动找到处理程序并将其注册到 `EventBus`/`EventStore`。对于实体的重新创建,同样适用,但增加了一个层。随着 Axon 促进 CQRS 的使用,您将拥有不同的模型。命令模型是 Axon 通过要找到的聚合细节here 所支持的。每次从聚合的 Repository
加载聚合时,框架都会完全为您从其自己的事件重新创建聚合。这是因为 Axon 实现了一个 EventSourcingRepository
,它只是遵循每次重新创建实体的范例。
如果您要重新创建的实体是查询模型的一部分,您将必须告诉负责向您的 @EventHandler
提供事件的组件从头开始(因此再次不需要让您自己阅读事件)。轴突中跟踪 EventStore
的技术方面称为 EventProcessor
。如果您想要一个可以重放这些事件以重新创建实体的 EventProcessor
,则必须使用 TrackingEventProcessor
(TEP)。正是 TEP 公开了一个 resetTokens()
操作,该操作实质上将事件处理器在事件流中的点更改为过去的点;从而允许您重新创建您的实体。可以在 here 中找到关于此的文档。
4.在 TrackingEventProcessor 上调用重置
如果您想重新创建查询模型,无论是因为其格式已更改,还是仅仅因为您有一个新模型,您都需要重置 TrackingToken(s)
(TEP) 的 TrackingEventProcessor
。请注意,这要求您使用 TEP 作为事件处理组件(包含 @EventHandler
注释方法的类)的处理器来更新您想要重新创建的模型。只有 TEP 提供了重置选项,因为它是唯一使用 TrackingTokens
来跟踪事件流进度的事件处理器。通过调整它在该流上的位置,您可以有效地调用重置。
如果要重置 TEP,首先需要知道处理器的名称。默认情况下,这将是事件处理组件 (EHC) 的包名称。但是,建议使用 EHC 上的 @ProcessingGroup
注释为您的处理组以及事件处理器提供不同的名称。它有助于对组进行推理,但也可以更轻松地调用重置。
有了名字,我们就可以在 Axon 的配置中找到 TEP。为此,您最好使用 EventProcessingConfiguration
,因为它公开了一个 eventProcessor(String,Class<T>)
方法,该方法为您返回一个 Optional<EventProcessor>
。有了这个选项,您就可以开始重置操作。
在重置 TEP 之前,您必须确保没有单个 TEP 实例在处理事件。这意味着您必须首先使用 TrackingEventProcessor#shutDown
方法关闭 TEP。在此之后,可以调用重置,这有效地将 TEP 的 TrackingToken(s)
调整到新位置。在此之后,您的 TEP 可以再次启动。
在代码中,这将如下所示:
public void resetTrackingProcessor(EventProcessingConfiguration config,String processorName) {
config.eventProcessor(processorName,TrackingEventProcessor.class)
.ifPresent(tep -> {
tep.shutDown();
tep.resetTokens();
tep.start();
});
}
请注意,上述内容在单节点系统中完全可以正常工作,因为 shutdown
操作同时关闭了唯一存在的 TEP。但是,如果您已将应用程序分发到多个实例中,那么您 将需要找出一种方法,在对其中一个实例调用重置之前关闭所有这些实例。简而言之,您需要一个在给定应用程序范围之外的中心组件,它了解所有现有的 TEP。自己创建它是完全可行的(我过去曾这样做过),但现在我建议您使用 Axon Server 来避免麻烦。
Axon Server 充当中间人,不仅用于事件存储和消息分发,还可以作为监控系统的地方。因此,您的事件处理器也是如此,添加了关闭、启动、释放令牌、拆分和合并操作。