Repository and Event Store

仓储和事件存储
仓储主要用于操作聚合。仓储通常在我们持久化数据的时候充当了一个网关角色。在CQRS中，仓储只做了根据聚合的ID来查询出对应的聚合。而其他一些类型的查询应该走查询数据库(query databases).

在Axon框架中，所有的仓储都必须实现Repository接口。这个接口定义了三个方法：load(identifier,version), load(identifier) and newInstance(factoryMethod). 这个load方法请请允许你从仓储中加载聚合。而version参数通常用于检查当前数据是否被更新过（请参阅高级冲突检测和解决方案）。newInstance则用于向仓储注册新的聚合实例。

根据你的基础持久存储需求，这里有很多基础实现，这些实现提供了大多数存储库所需的基本功能。Axon框架在保存聚合的当前状态的仓储（请参阅标准仓储）和存储聚合事件的仓储两者上做出了一些区分（请参阅事件溯源仓储）。

注意，Repository接口没有删除（标识符）方法。他是通过调 用聚合内的AggregateLifecycle.markDeleted（）方法来删除的。聚合删除和其他情况一样，他们是一种状态改变，唯一的区别是在很多情况下它是不可逆的。你应该在你的聚合上创建你自己认为有意义的方法，以此来将聚合的状态设置为’已删除’状态。你可以注册你想要发布的任何事件。

标准仓储
标准仓储实际存储的是聚合的当前状态。每次更新后，新的状态都会覆盖旧的状态。这使得应用程序的查询组件和命令组件可以使用相同的信息。根据你创建的应用程序的类型，这可能是最简单的方案。如果是这样的话，Axon提了一些模块来帮助你实现这样一个仓储。

Axon框架提供了一些开箱即用的标准仓储实现：GenericJpaRepository。这个聚合被期望是一个有效的JPA实体。他通过配置一个EntityManagerProvider提供EntityManager来管理实际的持久化，以及指定在仓储中的实际聚合类型的类。当聚合调用静态方法AggregateLifecycle.apply()时，你也可以通过发布到EventBus的事件来充当。

你也可以很容易实现你自己的仓储。那样的话，最好的方式就是继承于LockingRepository这个类。而聚合的包装类型，建议使用AnnotatedAggregate。 具体请查看GenericJpaRepository的源代码。

事件源仓储
聚合根可以根据配置的事件源仓储来加载事件从而重新构建它自己的状态。这些仓储不会存储聚合本身，他们存储的是聚合产生的一系列事件。基于这些事件，你可以在任何时间来重新生成聚合的状态。

EventSourcingRepository提供了Axon框架中任何事件源仓储所须的基本功能。但是他依赖于EventStore(请参阅事件存储实现)，它抽象出事件的实际存储机制。

当然，你也可以提供一个Aggregate工厂。AggregateFactory指定如何创建一个聚合实例。 一旦创建了聚合，EventSourcingRepository会用从事件仓储加载的事件来进行初始化它。 Axon框架附带了一些您可能会用到的AggregateFactory实现。 如果他们不够用，创建你自己的实现是非常容易的。

GenericAggregateFactory
GenericAggregateFactory是一个特殊的AggregateFactory实现，可用于任何类型的事件源聚合根。 GenericAggregateFactory会创建一个仓储管理的Aggregate类型的实例。 Aggregate类必须是非抽象类，并声明一个没有初始化的默认的无参数构造函数。

GenericAggregateFactory适用于大多数场景，其中聚合不需要特别注入不可序列化的资源。

SpringPrototypeAggregateFactory
取决于你的架构，使用spring将依赖关系注入到聚合中可能会有一些帮助。比如，您可以将查询仓储注入到您的聚合中，以确保某些值的存在（或不存在）。
为了将依赖关系注入到你的聚合中，你须要在定义了SpringPrototypeAggregateFactory的spring上下文中配置一个聚合根的原型bean. 它不是使用构造函数创建实例，而是使用Spring应用程序上下文来实例化聚合而且他会将其他依赖注入到你的聚合根中。
实现你自己的AggregateFactory
在有些情况下，GenericAggregateFactory并不能满足你的须求。比如，你有一个抽象聚合，他有多个实现（例如，PublicUserAccount和BackOfficeAccount都是继承于Account这个聚合）。你应该使用单个仓储而不是为每个聚合创建一个仓储，并且配置一个能知道不同实现的AggregateFactory。

Aggregate Factory所做的大部分工作是创建未初始化的Aggregate实例。 它必须使用给定的聚合标识符和流中的第一个事件。通常来说，这个事件是一个创建事件，它包含关于预期的聚合类型的提示。你可以使用这些信息来选择他具体的哪一个实现类并调用其构造方法。确保该方法是没有使用任何事件，并且这个聚合必须是已经被初始化了。

同简单的存储库直接加载聚合的实现相比，基于事件初始化聚合可能是一项耗时的工作。而CachingEventSourcingRepository提供了一个缓存，可以从中加载聚合（如果可用）。

Event store 的实现
Event Sourcing仓储需要一个事件存储（event store）来存储和加载来自聚合的事件。 事件存储提供了事件总线的功能，除此之外，它还可以持久化已发布的事件，并且能够基于聚合标识符来检索事件。

Axon提供了一个开箱即用的事件存储，即EmbeddedEventStore。 它将事件的实际存储和检索委托给EventStorageEngine来管理。

这里有多个EventStorageEngine实现可用：

JpaEventStorageEngine
JpaEventStorageEngine将事件存储在兼容JPA的数据源中。 JPA Event Store存储的事件被称为实体。这些实体不仅包括了事件的序列化形式，还提供了为快速查找到这些实体而存储的一些元数据字段。要使用JpaEventStorageEngine，您必须在您的类路径上具有JPA（javax.persistence）注解。

默认情况下，事件存储须要配置持久化上下文（比如，在META-INF/persistence.xml文件中定义），这个上下文主要包括DomainEventEntry和SnapshotEventEntry类（他们都在org.axonframework.eventsourcing.eventstore.jpa包中）
下面是一个配置持久化上下文的例子：
<persistence xmlns="http://java.sun.com/xml/ns/persistence" version="1.0">
<persistence-unit name="eventStore" transaction-type="RESOURCE_LOCAL"> (1)
<class>org...eventstore.jpa.DomainEventEntry</class> (2)
<class>org...eventstore.jpa.SnapshotEventEntry</class>
</persistence-unit>
</persistence>
1.在这个例子中，事件存储有一个特定的持久化单元。但是，您可以选择将第三行添加到任何其他持久化单元配置中。

2. 该行将DomainEventEntry（由JpaEventStore使用的类）注册到持久性上下文中。
   注意：
   Axon使用锁定来防止两个线程同时访问相同的聚合。 但是，如果您在同一个数据库上有多个JVM，这不会对您有所帮助。 在这种情况下，您必须依靠数据库来检测冲突。 对事件存储的并发访问将导致主键唯一性冲突，因为该表记录的是单个事件，而每个事件有自己的序列号。 如果数据库中已有一个带了序列号的聚合事件，再将第二个相同的序列号（唯一标识）聚合事件写入时就会失败。
   JpaEventStorageEngine可以检测到此错误并将其转换为ConcurrencyException。 但是，每个数据库系统都会以不同的方式报告此冲突。 如果您使用JpaEventStore来注册的DataSource，它将尝试检测数据库的类型并找出哪些错误代码表示主键唯一性冲突。 或者，您可以提供一个PersistenceExceptionTranslator实例， 该实例可以知道给定的异常表示主键唯一性冲突。
   如果没有提供数据源或PersistenceExceptionTranslator，数据库驱动程序会把异常按原样抛出。

默认情况下，JPA事件存储引擎需要一个EntityManagerProvider实现，这个实现会返回供EventStorageEngine使用的EntityManager实例。 这也允许使用应用程序管理的持久性上下文。 EntityManagerProvider的责任是提供一个正确的EntityManager实例。当然也可以使用应用程序管理持久化上下文。EntityManagerProvider的责任是提供一个正确的EntityManager实例。

EntityManagerProvider有一些实现可直接使用，每个实现都可以满足不同的需求。SimpleEntityManagerProvider只会在实例化时才返回给它EntityManager实例。 这使得实现对于容器管理的上下文来说是一个简单的选项。 另外，还有一个ContainerManagedEntityManagerProvider，它返回默认的持久化上下文，并由Jpa事件存储默认使用。

如果您有一个名为“my PersistenceUnit”的持久性单元，您希望在JpaEventStore中使用它，那么这就是EntityManager Provider实现的例子：
public class MyEntityManagerProvider implements EntityManagerProvider {
​
private EntityManager entityManager;
​
@Override
public EntityManager getEntityManager() {
return entityManager;
}
​
@PersistenceContext(unitName = "myPersistenceUnit")
public void setEntityManager(EntityManager entityManager) {
this.entityManager = entityManager;
}
默认情况下。JPA事件存储实体是在DomainEventEntry和SnapshotEventEntry中。虽然这在大多数情况下就足够了，但您可能会遇到这些实体提供的元数据不足的情况。 或者您可能想要将不同聚合类型的事件存储在不同的表中。
在这种情况下，你可以继承JpaEventStorageEngine该类。它包含一些protected的方法，您可以重写以实现不用的行为。

警告：
注意，持久化提供者(如Hibernate)在EntityManager实现上使用了一级缓存。通常，这意味着在查询中使用或返回的所有实体都与EntityManager相连。它们只在事务被提交时被清除，或者在事务中执行明确的“清除”。尤其是当查询在事务上下文中执行时。
要解决这个问题，请确保只对非实体对象进行查询。您可以使用JPA的"SELECT new SomeClass(parameters) FROM..."风格查询来解决这个问题。或者，获取一批事件后调用entitymanager.flush()和entitymanager.clear()。当加载一个非常大的事件流时可能会导致OutOfMemoryException。

JDBC 事件存储引擎
JDBC事件存储引擎使用JDBC连接将事件存储在兼容JDBC的数据库中。 通常，这些是关系数据库。 理论上，任何具有JDBC驱动程序的东西都可以用来支持JDBC事件存储引擎。

与JPA相似，JDBC事件存储引擎将条目（entries）中的事件存储起来。 默认情况下，每个事件都存储在单个条目中，该条目与表中的一行相对应。 一个表用于存事件，另一个用于存快照。

JdbcEventStorageEngine使用ConnectionProvider获取连接。 通常，这些连接可以直接从DataSource获得。 但是，Axon会将这些连接绑定到工作单元，以便在工作单元中使用单个连接。 这确保了即使多个工作单元嵌套在同一个线程中，也可以使用单个事务来存储所有事件。
注意：
建议Spring用户使用SpringDataSourceConnectionProvider从DataSource连接到现有的事务。
MongoDB事件存储引擎
MongoDB是一个基于文档的NoSQL存储。 其可扩展性使其适合用作Event Store。 Axon提供了MongoEventStorageEngine，它使用MongoDB作为后台数据库。 它包含在Axon Mongo模块（Maven artifactId axon-mongo）中。

事件存储在两个独立的集合（collections）中：一个用于实际事件流，另一个用于快照。

默认情况下，MongoEventStorageEngine将每个事件存储在单独的文档中。 但是，可以通过过StorageStrategy来修改。 Axon提供的替代方案是DocumentPerCommitStorageStrategy，它为已经存储在单个提交中（即在同一个DomainEventStream中）的所有事件创建单个文档。

将整个提交在单个文档中的优点是：该提交以原子方式保存的。 此外，对于任何数量的事件，它只需要一次往返。 缺点是直接在数据库中查询事件变得更加困难。 例如，当重构领域模型时，如果将事件包含在“提交文档”中，则将事件从一个集合“转移”到另一个集合是比较困难的。
MongoDB不需要很多配置。 它所需要的只是对一个存储事件集合的引用，然后您就可以开始了。对于生产环境，您可能需要对集合上的索引进行双重检查。
Event Store Utilities
Axon提供了许多在某些情况下可能有用的事件存储引擎。

将多个事件存储到一起
SequenceEventStorageEngine是两个事件存储引擎的包装。在查询的时候，它会从两个事件存储引擎中返回事件。新增的事件会交给第二个事件存储引擎。例如，在出于性能原因使用两种不同的事件存储实现的情况下，这个效果是非常的明显。第一个是比较大但查询很慢的事件存储库，第二个为快速读取和写入而优化。

过滤存储事件
FilteringEventStorageEngine允许事件根据谓词（predicate）进行过滤。 只有符合此谓词的事件才会被存储。 请注意，使用事件存储作为事件源的事件处理程序可能无法接收这些事件，因为它们未被存储。

内存存储引擎
这里有一个EventStorageEngine的实现。它将存储的事件保存在内存中，他就是InMemoryEventStorageEngine类。虽然他可能优于其他任何的事件存储，但是他基本不用于生产环境中。但是，它对于需要将事件存储的短期(short-lived)工具或测试来说就非常有用。

Influencing the serialization process
事件存储需要用一种序列化方式把事件序列化后存储起来。 默认情况下，Axon使用XStreamSerializer，它使用XStream将事件序列化为XML。 XStream速度相当快，且比Java序列化更灵活。 此外，XStream序列化的结果是易读的。 对于日志记录和调试目的非常有用。

XstreamSerializer是可以配置的。 您可以定义它应该用于具体的包，类或甚至字段的别名。 除了缩短可能的长名称之外，还可以在事件的类定义发生更改时使用别名。 有关别名的更多信息，请访问XStream网站。

另外，Axon还提供了JacksonSerializer，它使用Jackson将事件序列化为JSON。 虽然它生成更紧凑的序列化表单，但它需要类遵循Jackson所要求的约定（或配置）。

您也可以实现自己的序列化类，只需创建一个实现Serializer类，并配置事件存储库以使用该实现类来替代默认值。

Serializing Events vs 'the rest'
从Axon 3.1版本开始，可以使用不同的序列化器来存储事件， 而Axon的有些对象变不须要序列化器（例如命令，快照，Sagas等）。 虽然XStreamSerializer几乎可以序列化任何对象，但是它的输出并不会百分百的和其他应用程序融合。 JacksonSerializer有更好的输出，但需要在序列化对象时指定结构。 这种结构通常存在于事件中，所以他非常合适的事件序列化器。

通过配置API，您可以简单地注册一个事件串行器如下：

Configurer configurer = ... // 初始化
configurer.configureEventSerializer(conf -> /* create serializer here*/);

如果未配置显式的eventSerializer，则使用已配置的主序列化程序对事件进行序列化（该序列化程序默认为XStream序列化程序）。

Event Upcasting(事件向上转型)
由于软件应用程序性质的不断变化，事件定义很可能随着时间而改变。 由于事件存储被视为只读和新增（不会有更新和删除操作，更新和删除操作都是新增事件和删除事件来做）数据源，因此应用程序必须能够读取所有事件，而不管它们何时添加。 这是事件向上转型的用处。

最初这个是来源于面向对象编程的概念，其中“子类在需要时自动转换为其超类”，向上转型（upcasting）的概念也可以应用于事件源。 Upcast一个事件意味着将其从原来的结构转变为新的结构。 与OOP的转换（upcasting）不同，事件转换（upcasting）不能完全自动完成，因为新事件的结构对于旧事件是未知的。 必须提供手工的Upcasters来指定如何将旧结构转换（upcasting）为新结构。
Upcasters是这样的一个类，它接受修订版为x的一个输入事件，并输出零个或多个修订版为x + 1的新事件。此外，upcasters是在一个处理链中处理，这意味着一个upcaster的输出被作为链中下一个upcaster的输入。 这允许您以增量方式更新事件，为每个新事件修订编写一个Upcaster，使它们变得小巧，独立并且易于理解。
注意：
也许Upcasting的最大好处是它允许你做非破坏性的重构，即完整的历史事件保持不变（不会被修改）。

在本节中，我们将教你如何编写Upcasters，随着Axon不同的Upcaster实现，并解释事件的序列化形式如何影响upcasters的编写。
为了允许upcaster查看他们正在接收的序列化对象的版本，事件存储将存储一个version以及事件的完全限定名称。这个version是由在序列化器中配置的RevisionResolver生成的。Axon提供了几个RevisionResolver的实现,比如AnnotationRevisionResolver,它检查在事件有效的payload的@Revision注解,SerialVersionUIDRevisionResolver 使用Java Serialization API和FixedValueRevisionResolver所定义的serialVersionUID，它总是返回一个预定义的值。后者在注入当前应用程序版本时是有用的。这将允许你看哪个版本的应用程序生成一个特定的事件。

Maven用户可以使用MavenArtifactRevisionResolver自动使用项目版本。它使用项目的groupId和artifactId进行初始化以获取版本。由于这只适用于由Maven创建的JAR文件，所以版本不能总是由IDE解决。如果某个版本无法解析，则返回null。

Maven用户可以使用MavenArtifactRevisionResolver,它会自动使用项目版本。它使用项目的groupId和artifactId进行初始化以获取版本。由于这只适用于由Maven创建的JAR文件，所以版本不能总是由IDE解决。如果某个版本无法被解析，则返回null。

Axon的upcasters不直接使用EventMessage，而是使用IntermediateEventRepresentation。 IntermediateEventRepresentation提供了检索所有必要字段的功能，以构建一个EventMessage（所以它也是一个upcasters的EventMessage）以及实际的upcast函数。这些upcasters功能在默认情况下只允许调整事件payload，payload类型以及添加事件的元数据。 upcast函数中事件的实际表示可能因使用的事件序列化器或期望的表单类型而异，所以IntermediateEventRepresentation的upcast函数允许选择期望的表示类型。其他字段，例如消息/聚合标识符，聚集类型，时间戳等不能通过IntermediateEventRepresentation进行调整。调整这些字段不是Upcaster的预期工作，因此这些选项不是由提供的IntermediateEventRepresentation实现提供的。

Axon框架中事件的基础Upcaster接口在IntermediateEventRepresentations流上工作，并返回一个IntermissionEventRepresentations的流。 因此，upcasting并不直接返回引入的upcast函数的最终结果，而是通过IntermediateEventRepresentations的处理链将每个upcast函数从一个修订链接传输到另一个。一旦这个过程发生，最终的结果被从它们中取出来，也就是说在序列化的事件上执行了实际的upcasting函数。

如何写一个upcaster
以下Java代码片段将作为一对一Upcaster（SingleEventUpcaster）的基本示例。

事件的旧版本：
@Revision("1.0")
public class ComplaintEvent {
private String id;
private String companyName;
​
// Constructor, getter, setter...
}
新版本的事件：
@Revision("2.0")
public class ComplaintEvent {
private String id;
private String companyName;
private String description; // New field
​
// Constructor, getter, setter...
}

Upcaster:
// Upcaster from 1.0 revision to 2.0 revision
public class ComplaintEventUpcaster extends SingleEventUpcaster {
private static SimpleSerializedType targetType = new SimpleSerializedType(ComplainEvent.class.getTypeName(), "1.0");
​
@Override
protected boolean canUpcast(IntermediateEventRepresentation intermediateRepresentation) {
return intermediateRepresentation.getType().equals(targetType);
}
​
@Override
protected IntermediateEventRepresentation doUpcast(IntermediateEventRepresentation intermediateRepresentation) {
return intermediateRepresentation.upcastPayload(
new SimpleSerializedType(targetType.getName(), "2.0"),
org.dom4j.Document.class,
document -> {
document.getRootElement()
.addElement("description")
.setText("no complaint description"); // Default value
return document;
}
);
}
}

Spring boot配置:
@Configuration
public class AxonConfiguration {
​
@Bean
public SingleEventUpcaster myUpcaster() {
return new ComplaintEventUpcaster();
}
​
@Bean
public JpaEventStorageEngine eventStorageEngine(Serializer serializer,
DataSource dataSource,
SingleEventUpcaster myUpcaster,
EntityManagerProvider entityManagerProvider,
TransactionManager transactionManager) throws SQLException {
return new JpaEventStorageEngine(serializer,
myUpcaster::upcast,
dataSource,
entityManagerProvider,
transactionManager);
}
}

Content type conversion
升级器(upcaster)在给定的内容类型（例如dom4j Document）上工作。 为了提高upcasters之间的额外灵活性，处理链中的各upcasters之间的内容类型可能会有所不同。 Axon将尝试使用ContentTypeConverters自动在内容类型之间进行转换。 它将搜索从类型x到类型y的最短路径，执行转换并将转换后的值传递到请求的升级器。 出于性能原因，只有在接收升级器上的canUpcast方法为真时才会执行转换。

ContentTypeConverters可能取决于使用的序列化程序的类型。 试图将bytes[]转换为dom4j Document除非使用将事件写为XML的序列化程序，否则将无济于事。 为确保UpcasterChain可以访问特定于序列化程序的ContentTypeConverters，可以将对序列化程序的引用传递给UpcasterChain的构造函数。
建议：
要获得最佳性能，请确保同一链中的所有升级者（其输出是另一个输入者）使用相同的内容类型。

如果您所需要的内容类型转换不是由Axon提供的，您可以使用ContentTypeConverter接口编写自己的内容。
XStreamSerializer支持Dom4J以及XOM作为XML文档。JacksonSerializer支持Jackson的JsonNode。

快照
当聚合会存活很长时间时，并且它们的状态不断变化时，它们会产生大量事件。我们不得不加载所有这些事件来重建聚合的状态可能会对性能产生巨大的影响。快照事件是具有特殊用途的域事件：它将任意数量的事件汇总为一个事件。通过定期创建和存储快照事件，事件存储不必返回长期事件列表而只是返回最后一个快照事件和快照创建后发生的所有事件。

例如，库存物品往往会经常变化。每次出售物品时，事件会将库存减少一个。每次装了运新货物时，库存都会增加一些数量。如果您每天销售一百件商品，那么您每天至少会产生100个事件。几天后，您的系统将花费太多时间去获取所有这些事件，以确定是否应该提出“ItemOutOfStockEvent”。而当一个快照事件替代很多这些事件，只需通过存储当前的库存数量即可。

创建一个快照
快照的创建可以由多种因素来触发，例如自上次快照创建的事件数量，初始化聚合的时间超过某个阈值，基于时间等。目前，Axon提供了一种机制，允许您根据事件计数阈值来触发创建快照。

SnapshotTriggerDefinition接口定义了何时应该创建快照。

EventCountSnapshotTriggerDefinition提供了在加载聚合需要的事件数量超过特定阈值时触发快照创建的机制。如果加载聚合所需的事件数量超过了配置的阈值，触发器会通知Snapshotter为聚合创建快照。

快照触发器配置在事件溯源仓储中，并具有许多可选配置属性：

Snapshotter：设置实际的快照程序实例，负责创建和存储实际的快照事件;

Trigger： 设置触发快照创建的阈值;

Snapshotter负责实际创建快照。通常，快照是一个应尽可能的和业务处理程序分开。因此，建议在不同的线程中运行快照。 Snapshotter接口声明了一个方法：scheduleSnapshot（），它将聚合的类型和唯一标识符作为参数。

Axon提供了一个AggregateSnapshotter，它可以创建和存储AggregateSnapshot实例。 这是一种特殊类型的快照，因为它包含了具体的实际聚合实例。 Axon提供的仓储知道这种类型的快照，并将从中提取聚合，而不是实例化一个新聚合。 在快照事件之后加载的所有事件都将流式传输到提取的聚合实例。

注意：
您必须保证使用的Serializer实例（默认为XStreamSerializer）能够序列化您的聚合。 XStreamSerializer要求您使用Hotspot JVM，否则您的聚合必须具有可访问的默认构造函数或实现了Serializable接口。

AbstractSnapshotter提供了一组基本的属性，使您可以修改创建快照的方式：

EventStore主要用于加载过去事件并存储快照的事件存储。此事件存储必须实现SnapshotEventStore接口。

Executor设置执行程序，例如ThreadPoolExecutor，它将提供线程来处理实际的快照创建。默认情况下，快照是在调用scheduleSnapshot（）方法的线程中创建的，通常不推荐用于生产。

AggregateSnapshotter提供了另外的属性：

AggregateFactories是这样一个属性，它用于配置创建聚合实例的工厂。配置多个聚合工厂允许您使用单个Snapshotter为各种聚合类型创建快照。 EventSourcingRepository实现提供对他们使用的AggregateFactory的访问。这可用于在Snapshotter中配置与仓库中使用的相同的聚合工厂。

注意：
如果您在另一个线程中执行创建快照，请确保根据需要为基础事件存储配置正确的事务管理。

 Spring用户可以使用SpringAggregateSnapshotter，当需要创建快照时，SpringAggregateSnapshotter将自动从应用程序上下文中查找正确的AggregateFactory。

​