Event Sourcing和CQRS实现

文章参考自：

https://github.com/soooban/AxonDemo

下面是具体内容：

Event Sourcing And CQRS
Event Sourcing 、CQRS 简述
Event Sourcing 简单来说就是记录对象的每个事件而不是记录对象的最新状态，比如新建、修改等，只记录事件内容，当需要最新的状态的时，通过堆叠事件将最新的状态计算出来。那么这种模式查询的时候性能会变的非常差，这个时候就涉及到了 CQRS ，简单的理解就是读写分离，通过事件触发，将最新状态保存到读库，查询全都走读库，理论上代码层，数据库层，都可以做到分离，当然也可以不分离，一般来说为了保证数据库性能，这里起码会将数据库分离。

为什么要使用

了解了 Event Sourcing 的基本内容之后，我们可以发现这个模式有很多的好处：

记录了对象的事件，相当于记录了整个历史，可以查看到任意一个时间点的对象状态；
都是以事件形式进行写入操作，理论上在高并发的情况下，没有死锁，性能会快很多；
可以基于历史事件做更多的数据分析。
Event Soucing 通常会和 DDD CQRS 一起讨论，在微服务盛行的前提下，DDD 让整个软件系统松耦合，而 Event Soucing 也是面向 Aggregate，这个模式很符合 DDD 思想，同时由于 Event Soucing 的特性，读取数据必然会成为瓶颈，这个时候 CQRS 就起到做用了，通过读写分离，让各自的职责专一，实际上在传统的方式中我们也可能会这么干，只是方式略微不同，比如有一个只读库，时时同步主库，让查询通过只读库进行，那么如果查询量特别大的时候，起码写库不会因为查询而下降性能。

背景

本着对新技术的尝试以及某些业务也确实有这方面的需求的出发点，对 Axon 做了一些尝试。

实现 Event Soucing
在了解了相关的基础之后，这里会以最简单的方式实现一个 EventSourcing 的例子，然后逐渐在之后的过程去丰富，本篇内容会实现一个将增删改操作使用 EventSoucing 取代的例子，读取部分暂时不做涉及。

Spring Boot 工程搭建
打开 http://start.spring.io/ 选择对应的版本(这里是 2.1.5 )以及相应的依赖，这里多选了一些之后会用到的服务：

添加配置文件:

spring:application:name: event-sourcing-servicedatasource:url: jdbc:mysql://localhost:3306/event?useUnicode=true&autoReconnect=true&rewriteBatchedStatements=TRUEusername: rootpassword: rootjpa:hibernate:ddl-auto: updateuse-new-id-generator-mappings: falseshow-sql: falseproperties:hibernate.dialect: org.hibernate.dialect.MySQL55Dialect

为了便于测试，这里我开启了 JPA 的自动更新，开发中你可能会使用 flyway 或者其他工具来管理数据库 schema 以及数据迁移。至此一个简单的服务就搭建完毕了。

Axon 依赖和配置
依赖添加
搭建好了工程之后，我们正式开始做一些 Axon 相关的事情，这个工程我们用一个简单的 Contract 业务来作为demo，首先添加依赖：
添加 Axon 依赖:

<!-- https://mvnrepository.com/artifact/org.axonframework/axon-spring-boot-starter -->
<dependency><groupId>org.axonframework</groupId><artifactId>axon-spring-boot-starter</artifactId><version>4.1.1</version><exclusions><exclusion><groupId>org.axonframework</groupId><artifactId>axon-server-connector</artifactId></exclusion></exclusions>
</dependency>
<!-- https://mvnrepository.com/artifact/com.google.guava/guava axon 依赖了 guava-->
<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>27.1-jre</version>
</dependency>

解释一下，axon 从 4.0 开始加入了 axon-server 这么一个东西，目的是将 event 的存储和分发剥离，以更好的发挥微服务的优点，但是在项目中引用这么一个不透明的东西感觉上不太好，所以这里就不采用 axon-server 了。axon-spring-boot-starter 就会采用 EmbeddedEventStore ，默认使用的是 JPA ，启动之后你也会发现 JPA 在数据库中创建了 5 张表，分别是 association_value_entry domain_event_entry saga_entry snapshot_event_entry token_entry，domain_event_entry 用来存储事件，snapshot_event_entry 用来存储快照，token_entry 用来记录 tracking event 的争夺，其他两张表用来存户 saga。

Axon 的配置

axon:serializer:general: jackson

这里指定使用了Jackson来进行序列化，模式 Axon 是使用 XML 进行序列化的，不方便查看，并且之后的事件升级都会很麻烦，所以进行了替换。

Axon Domain Model 定义
定义 aggregate

public interface ContractInterface {Long getId();@NotBlankString getName();@NotBlankString getPartyA();@NotBlankString getPartyB();
}
@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@Aggregate
public class ContractAggregate implements ContractInterface {@AggregateIdentifierprivate Long id;private String name;private String partyA;private String partyB;private boolean deleted = false;
}

定义 commands

@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
public class AbstractCommand {@TargetAggregateIdentifierprivate Long identifier;
}@Getter
@Setter
@NoArgsConstructor
public class UpdateContractCommand extends AbstractCommand implements ContractInterface {private String name;private String partyA;private String partyB;public UpdateContractCommand(Long identifier, String name, String partyA, String partyB) {super(identifier);this.name = name;this.partyA = partyA;this.partyB = partyB;}
}@Getter
@Setter
@NoArgsConstructor
public class CreateContractCommand extends UpdateContractCommand {public CreateContractCommand(Long identifier, String name, String partyA, String partyB) {super(identifier, name, partyA, partyB);}
}@NoArgsConstructor
@Getter
@Setter
public class DeleteContractCommand extends AbstractCommand {public DeleteContractCommand(Long identifier) {super(identifier);}
}
定义 events
@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
public class AbstractEvent {@TargetAggregateIdentifierprivate Long identifier;
}@Getter
@Setter
@NoArgsConstructor
public class ContractUpdatedEvent extends AbstractEvent implements ContractInterface {private String name;private String partyA;private String partyB;public ContractUpdatedEvent(Long identifier, String name, String partyA, String partyB) {super(identifier);this.name = name;this.partyA = partyA;this.partyB = partyB;}
}@Getter
@Setter
@NoArgsConstructor
public class ContractCreatedEvent extends ContractUpdatedEvent {public ContractCreatedEvent(Long identifier, String name, String partyA, String partyB) {super(identifier, name, partyA, partyB);}
}@Getter
@Setter
@NoArgsConstructor
public class ContractDeletedEvent extends AbstractEvent {public ContractDeletedEvent(Long identifier) {super(identifier);}
}

这里只抽象了统一的 command 和 event，在实际的业务开发过程中可以有更多的抽象。

实现各个 Handler
在这里我们实现 Event Soucing 的模式，把各个 Handler 都放在 aggregate 里面。

@CommandHandler
public ContractAggregate(CreateContractCommand command, MetaData metaData, UIDGenerator generator) {if (null == command.getIdentifier()) {command.setIdentifier(generator.getId());}AggregateLifecycle.apply(new ContractCreatedEvent(command.getIdentifier(), command.getName(), command.getPartyA(), command.getPartyB()), metaData);
}@CommandHandler
private void on(UpdateContractCommand command, MetaData metaData) {AggregateLifecycle.apply(new ContractUpdatedEvent(command.getIdentifier(), command.getName(), command.getPartyA(), command.getPartyB()), metaData);
}@CommandHandler
private void on(DeleteContractCommand command, MetaData metaData) {AggregateLifecycle.apply(new ContractDeletedEvent(command.getIdentifier()), metaData);
}@EventSourcingHandler
private void on(ContractCreatedEvent event) {this.setIdentifier(event.getIdentifier());this.onUpdate(event);
}@EventSourcingHandler
private void onUpdate(ContractUpdatedEvent event) {this.setName(event.getName());this.setPartyA(event.getPartyA());this.setPartyB(event.getPartyB());
}@EventSourcingHandler(payloadType = ContractDeletedEvent.class)
private void on() {this.setDeleted(true);
}

这里看到有一个 CommandHandler 的注解写在了构造方法上，那么在处理这个 Command 的时候，将会自动创建一个对象。另外这里的 MetaData 是在 command 发送的时候顺带的附加信息，可以是用户信息，机器信息等等，后续也会涉及这部分，这里就不深入了。

启动项目后，JPA 应该会在数据库中生成几张表，其中 worker_id 是之前我们编写的 Spring Cloud 的 ID 生成器所产生的，剩余的表都是 Axon 自己产生的，这里我使用的数据库并没有使用mb4编码，因为在mb4编码下 Axon 的索引会过长，这个也不是问题，因为实际开发过程中，我们可以将生成的语句自己修改了下，将主键的长度改小一点即可，后面在完善的过程中也会涉及到。

编写接口
aggreate 以及各个 handler 都已经写完了，那么我们开始编写接口，让工程可以顺利的跑起来。

@RestController
@RequestMapping("/contracts")
@AllArgsConstructor
public class ContractController {private final CommandGateway commandGateway;@PostMappingpublic void createContract(@RequestBody @Valid CreateContractCommand command) {commandGateway.send(command);}@PutMapping("/{id}")public void updateContract(@PathVariable("id") Long id, @RequestBody @Valid UpdateContractCommand command) {command.setIdentifier(id);commandGateway.send(command);}@DeleteMapping("/{id}")public void deleteContract(@PathVariable("id") Long id) {commandGateway.send(new DeleteContractCommand(id));}
}

启动工程并顺序执行 POST UPDATE DELETE 操作，你会发现domain_event_entry中多了三条记录，这张表就是用来记录事件的，可以看到这里详细的记录了每个事件的发生时间、内容、附加信息、类型等信息。至此本次 Event Soucing 的例子就结束了，主要将传统的增删改操作改造成了事件记录的形式。下次将会从 CQRS 的角度实现数据的读取。

CQRS实现

实现 CQRS
在实现了 EventSoucing 之后，亟待解决的问题是查询了，理论上同一 Service 可以做到多数据源，甚至多数据库，这篇文章就暂时以同一个数据库为例子，同样使用 JPA 去做 view 的 ORM。

建立 entity
第一步当然是建立对应的 Entity 和对应的 Repository 了：

@Entity
@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
public class ContractView implements ContractInterface {@Id@Column(length = 64)private Long id;private String name;private String partyA;private String partyB;private boolean deleted = false;}@Repository
public interface ContractViewRepository extends JpaRepository<ContractView, Long> {}

实现 view 的存储
接下来我们就实现事件发生之后，view 的存储，这个过程是由一个独立的 EventHandler 来实现的，新建ContractViewHandler:

@Component
@Slf4j
@AllArgsConstructor
@Transactional
public class ContractViewHandler {private final EventSourcingRepository<ContractAggregate> customEventSourcingRepository;private final ContractViewRepository contractViewRepository;/*** 任何 contract 事件发生之后，重新计算 aggregate 的最新状态，转换成 view 之后存储到本地** @param event   any event from contract* @param message domain event wrapper*/@EventHandlerpublic void on(AbstractContractEvent event, DomainEventMessage<AbstractContractEvent> message) {log.info(MessageFormat.format("{0}: {1} , seq: {2}, payload: {3}", message.getType(), message.getAggregateIdentifier(), message.getSequenceNumber(), message.getPayload()));updateContractView(message.getAggregateIdentifier());}@Transactionalpublic void updateContractView(String id) {LockAwareAggregate<ContractAggregate, EventSourcedAggregate<ContractAggregate>> lockAwareAggregate = customEventSourcingRepository.load(id);ContractAggregate aggregate = lockAwareAggregate.getWrappedAggregate().getAggregateRoot();ContractView contractView = contractViewRepository.findById(Long.valueOf(id)).orElse(new ContractView());contractView.setId(aggregate.getIdentifier());contractView.setDeleted(aggregate.isDeleted());contractView.setName(aggregate.getName());contractView.setPartyA(aggregate.getPartyA());contractView.setPartyB(aggregate.getPartyB());contractViewRepository.save(contractView);}
}

接收到事件之后，把 aggregate 使用mapstruct转换到 view ，然后直接保存就完成了 view 的存储。看起来好像事情干完了，然而一启动就发现EventSourcingRepository这个 bean 找不到，实际上这个 bean 是需要我们自己定义的，在 config 中加入：

@Bean
public EventSourcingRepository<ContractAggregate> contractAggregateRepository(EventStore eventStore) {return new EventSourcingRepository<>(ContractAggregate.class, eventStore);
}

启动并发送一个 POST 请求，在查看contract_view表，已经有一条记录被插入。

让 view 存储过程异步
似乎看起来，上面的流程已经没什么问题了，然而实际上是有问题的，我们先来看一张图：
image

这张图描述了一个 Event Soucing 和 CQRS 的理想模型，可以看到 event 存储和 view 的存储应该是分离的，view 的存储是等 event 存储之后异步进行存储，同理事件的发送也是这样，那么我们上面的例子有没有实现 view 的存储是分离的呢，很遗憾，Axon 默认的实现并不是这样的，当 event handler 丢出 runtime exception 之后，事件并没有被存储，也就是说他们应该是都在一个 tranaction 里面。而 Axon 确实提供了 event 的异步处理，官方文档里有提到过，但是实际上仍然没有解决问题，因为 Axon 在 Subscribe 模式的 handler 调用过程中，并不会等事件事务存储之后再去调用，而是存储的同时依次调用，也就是说虽然可以做到 view 的 handler 异步化，但仍然做不到保证事件的存储之后再去更新 view（理论上 axon 的tracking event 也是可以保证消息发送和 view handler 异步的，但是这个模式下，Axon 会隔一段时间去扫表，并且只有一个节点可以处理，所以我觉得这种方式不太好就没有用）。所以我们改进一下 view 的流程：
image

这样 view 的存储是分离了，但是事件是否发出却没有得到保证，也就是可靠消息，关于可靠消息这一块，后面将专门搞个话题处理可靠消息以及可靠消息下 view 层的实现，这里为了只涉及 CQRS 就暂时不做深入。

基于 Aggregate 的查询实现
有了 view 的存储之后，查询方式就和以前传统的 jpa 方式一样了，那么我们有些时候需要从 aggregate 查询最新的状态，比如在 view 处理错误的时候，其实 Axon 也提供了相关的实现，从上面的 view handler 也可以看到，我们可以通过一个 repository 去 load 一个 aggregate，那么通过 aggregate 查询的实现也就比较简单了，这里就不贴代码了。

让 Aggregate 可以查询历史状态
首先我们先看以下 Axon 存储的整体结构：

Repository ->EventStore->EventStorageEngine 其中EventStorageEngine 提供了最底层的查询存储功能，EventStore进行封装、过滤和优化，整个 aggregate 的过程就是从 DB 中 fetch 所有的事件（这里会做一个 snapshot 的优化），然后将事件发送出去，那么我们为了尽量不去改写底层的查询，可以在EventStore做一个内存过滤然后向 Repository 输出接口，其实这个效率也还行，因为本身一个对象的事件有限，并不会吃掉很多资源去做过滤这件事情。

自定义 EventStore
这里比较简单，只要参照原来EmbeddedEventStore里面的 readEvents 方法，然后将里面的内容按照时间过滤一下。

@Slf4j
public class CustomEmbeddedEventStore extends EmbeddedEventStore {public CustomEmbeddedEventStore(Builder builder) {super(builder);}public static CustomEmbeddedEventStore.Builder builder() {return new CustomEmbeddedEventStore.Builder();}public DomainEventStream readEvents(String aggregateIdentifier, Instant timestamp) {Optional<DomainEventMessage<?>> optionalSnapshot;try {optionalSnapshot = storageEngine().readSnapshot(aggregateIdentifier);} catch (Exception | LinkageError e) {log.warn("Error reading snapshot. Reconstructing aggregate from entire event stream.", e);optionalSnapshot = Optional.empty();}DomainEventStream eventStream;// 加上时间判断，如果 snapshot 在指定的时间之间，那么可以使用，否则直接读取所有的 eventsif (optionalSnapshot.isPresent() && optionalSnapshot.get().getTimestamp().compareTo(timestamp) <= 0) {DomainEventMessage<?> snapshot = optionalSnapshot.get();eventStream = DomainEventStream.concat(DomainEventStream.of(snapshot),storageEngine().readEvents(aggregateIdentifier,snapshot.getSequenceNumber() + 1));} else {eventStream = storageEngine().readEvents(aggregateIdentifier);}eventStream = new IteratorBackedDomainEventStream(eventStream.asStream().filter(m -> m.getTimestamp().compareTo(timestamp) <= 0).iterator());Stream<? extends DomainEventMessage<?>> domainEventMessages = stagedDomainEventMessages(aggregateIdentifier);return DomainEventStream.concat(eventStream, DomainEventStream.of(domainEventMessages));}// 这里为了构建自己的 eventStore 就把 builder 搬过来了，略去了内部实现，具体可以看源码public static class Builder extends EmbeddedEventStore.Builder {...}
}

自定义 Repository
首先分析一下EventSourcingRepository的继承关系：

EventSourcingRepository->LockingRepository->AbstractRepository->Repository 其中Repository是个 interface

其实最好的方式是在 Repository interface 中加一个 load(Long, Instant) 方法，但是这样我们需要改造的地方就有点多了，因为要一层层的添加实现，还是考虑接地气一点，直接在 EventSourcingRepository 中实现这一功能（相当于把几个父类的事情一起干了）。

@Slf4j
public class CustomEventSourcingRepository<T> extends EventSourcingRepository<T> {private final CustomEmbeddedEventStore eventStore;private final SnapshotTriggerDefinition snapshotTriggerDefinition;private final AggregateFactory<T> aggregateFactory;private final LockFactory lockFactory;public static <T> Builder<T> builder(Class<T> aggregateType) {return new Builder<>(aggregateType);}protected CustomEventSourcingRepository(Builder<T> builder) {super(builder);this.eventStore = builder.eventStore;this.aggregateFactory = builder.buildAggregateFactory();this.snapshotTriggerDefinition = builder.snapshotTriggerDefinition;this.lockFactory = builder.lockFactory;}protected EventSourcedAggregate<T> doLoadWithLock(String aggregateIdentifier, Instant timestamp) {DomainEventStream eventStream = eventStore.readEvents(aggregateIdentifier, timestamp);SnapshotTrigger trigger = snapshotTriggerDefinition.prepareTrigger(aggregateFactory.getAggregateType());if (!eventStream.hasNext()) {throw new AggregateNotFoundException(aggregateIdentifier, "The aggregate was not found in the event store");}EventSourcedAggregate<T> aggregate = EventSourcedAggregate.initialize(aggregateFactory.createAggregateRoot(aggregateIdentifier, eventStream.peek()),aggregateModel(), eventStore, trigger);aggregate.initializeState(eventStream);if (aggregate.isDeleted()) {throw new AggregateDeletedException(aggregateIdentifier);}return aggregate;}protected LockAwareAggregate<T, EventSourcedAggregate<T>> doLoad(String aggregateIdentifier, Instant timestamp) {Lock lock = lockFactory.obtainLock(aggregateIdentifier);try {final EventSourcedAggregate<T> aggregate = doLoadWithLock(aggregateIdentifier, timestamp);CurrentUnitOfWork.get().onCleanup(u -> lock.release());return new LockAwareAggregate<>(aggregate, lock);} catch (Exception ex) {log.debug("Exception occurred while trying to load an aggregate. Releasing lock.", ex);lock.release();throw ex;}}public LockAwareAggregate<T, EventSourcedAggregate<T>> load(String aggregateIdentifier, Instant timestamp) {if (timestamp == null) {return this.load(aggregateIdentifier);}UnitOfWork<?> uow = CurrentUnitOfWork.get();Map<String, LockAwareAggregate<T, EventSourcedAggregate<T>>> aggregates = managedAggregates(uow);LockAwareAggregate<T, EventSourcedAggregate<T>> aggregate = aggregates.computeIfAbsent(aggregateIdentifier,s -> doLoad(aggregateIdentifier, timestamp));uow.onRollback(u -> aggregates.remove(aggregateIdentifier));prepareForCommit(aggregate);return aggregate;}public static class Builder<T> extends EventSourcingRepository.Builder<T> {...// 这里我略去了 builder class 内部的方法，具体的可以看源码。}
}

可以看到这里我们在 timestamp 为空的情况下，直接走了原来的 load 方法，剩下的就是参照原来的写法，直接在 CustomEventSourcingRepository 实现几个父类的 loadXXX 方法即可。

另外配置中更新下我们需要的 Repository Bean：

@Configuration
@RegisterDefaultEntities(packages = {"org.axonframework.eventsourcing.eventstore.jpa"
})
public class AxonContinueConfiguration {@Beanpublic CustomEmbeddedEventStore eventStore(EventStorageEngine storageEngine, AxonConfiguration configuration) {return CustomEmbeddedEventStore.builder().storageEngine(storageEngine).messageMonitor(configuration.messageMonitor(EventStore.class, "eventStore")).build();}@Beanpublic CustomEventSourcingRepository<ContractAggregate> contractAggregateRepository(CustomEmbeddedEventStore eventStore,ParameterResolverFactory parameterResolverFactory) {return CustomEventSourcingRepository.builder(ContractAggregate.class).eventStore(eventStore).parameterResolverFactory(parameterResolverFactory).build();}@Beanpublic EventStorageEngine eventStorageEngine(Serializer defaultSerializer,PersistenceExceptionResolver persistenceExceptionResolver,@Qualifier("eventSerializer") Serializer eventSerializer,AxonConfiguration configuration,EntityManagerProvider entityManagerProvider,TransactionManager transactionManager) {return JpaEventStorageEngine.builder().snapshotSerializer(defaultSerializer).upcasterChain(configuration.upcasterChain()).persistenceExceptionResolver(persistenceExceptionResolver).eventSerializer(eventSerializer).entityManagerProvider(entityManagerProvider).transactionManager(transactionManager).build();}@Beanpublic EventProcessingConfigurer eventProcessingConfigurer(EventProcessingConfigurer eventProcessingConfigurer) {eventProcessingConfigurer.usingSubscribingEventProcessors();return eventProcessingConfigurer;}
}

这里由于 EventStorageEngine 的 Auto Config 在自定义了 eventStore 之后就不起作用了，所以这里把 JpaEventStoreAutoConfiguration 中的内容搬过来了。另外在自定义上面三个 bean 之后默认的 event mode 也莫名其妙的变成了 tracking event，所以这里第四个 bean 对默认的 processor 做了修改。

Query Command 的应用
Axon 基于 Command 这种模式，将查询也做了一部分，这里我们尝试下用它的 Query Command 来实现各种查询。

添加 Query ：

@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
public class QueryContractCommand {@NotBlank@NotNullprivate String id;private Instant endDate;
}
添加 Handler ：
@Component
@AllArgsConstructor
@Slf4j
public class ContractQueryHandler {private final CustomEventSourcingRepository<ContractAggregate> contractAggregateRepository;@QueryHandlerpublic ContractAggregate on(QueryContractCommand command) {LockAwareAggregate<ContractAggregate, EventSourcedAggregate<ContractAggregate>> lockAwareAggregate = contractAggregateRepository.load(command.getId().toString(), command.getEndDate());return lockAwareAggregate.getWrappedAggregate().getAggregateRoot();}
}
Controller 中发送命令：
private final QueryGateway queryGateway;@GetMapping("/{id}")
public ContractAggregate getContract(@PathVariable("id") Long id) {QueryContractCommand command = new QueryContractCommand(id, Instant.now());return queryGateway.query(command, ContractAggregate.class).join();
}

快照

深入使用 Axon
实现 snapshot
在前面文章中，我们略微涉及了一些 snapshot 的概念，其实这个概念还是比较好理解的，当事件堆积到一定的程度，每次 load 都会花费一定的时间，这个时候自然会想到 snapshot，先将一部分事件进行计算，然后生成一个 snapshot，后续 load 的时候先读取 snapshot，这样就省去了很多计算过程。如果你读过之前改写 event store 部分的代码，就会发现每个 aggregate 实际上只会存储一个 snapshot，每当新的生成的时候会替换老的。对一个 aggregate 来说，snapshot 同样和事件差不多，它都可以当做事件来处理，但是对于 aggregate 来说，我并不想去处理 snapshot，我只是需要一个计算好的结果而已。基于这个，Axon 给我们提供了 AggregateSnapshotter，这类 snapshot 可以直接还原 aggregate 状态。

了解大概的原理之后，我们就要做的事情其实比较明确了：

告诉 Axon 我们要在什么时候触发 snapshot 的生成。
告诉 Axon 我们要生成什么样的 snapshot。

@Bean
public CustomEventSourcingRepository<ContractAggregate> contractAggregateRepository(CustomEmbeddedEventStore eventStore,SnapshotTriggerDefinition snapshotTriggerDefinition,ParameterResolverFactory parameterResolverFactory) {return CustomEventSourcingRepository.builder(ContractAggregate.class).eventStore(eventStore).snapshotTriggerDefinition(snapshotTriggerDefinition).parameterResolverFactory(parameterResolverFactory).build();
}@Bean
public SnapshotTriggerDefinition snapshotTriggerDefinition(Snapshotter snapshotter) {return new EventCountSnapshotTriggerDefinition(snapshotter, 5);
}@Bean
public AggregateSnapshotter snapShotter(CustomEmbeddedEventStore eventStore, ParameterResolverFactory parameterResolverFactory) {return AggregateSnapshotter.builder().eventStore(eventStore).parameterResolverFactory(parameterResolverFactory).aggregateFactories(Collections.singletonList(new GenericAggregateFactory<>(ContractAggregate.class))).build();
}

SnapshotTriggerDefinition是用来定义 snapshot 策略的，这里使用了 Axon 提供的策略，是基于 event count 的。
AggregateSnapshotter告诉 Axon 我们需要生成这类 snapshot，其实还有一类 snapshot 叫SpringAggregateSnapshotter，如果我们使用的是系统自带的EventSourcingRepository，那么可以直接使用这类 snapshot，实质上看代码，它就是取找了EventSourcingRepository这个 bean 的 AggregateFactory。
Repository 初始化的时候指定snapshotTriggerDefinition，那么在执行 load 方法的时候，就会去触发了。
启动项目，PUT 某个 aggregate 5次之后，你就会发现snapshot_event_entry这张表里多了记录，snapshot 就生成了。

实现 upcaster
upcaster 的概念也是比较好理解的，比如 create 事件，后期加入了一个新的字段，而老的事件没这个字段，这个时候就需要一个升级过程，把老的事件升级成新的事件，大部分的升级都是单个事件内的增减变动字段，也有比较复杂的情况，一个事件变多个，或者多个变一个等等。在 Axon 中把这个过程称为 upcaster，对应的 interface 是 Upcaster，里面只有一个方法Stream upcast(Stream intermediateRepresentations);看这个方法，理论上来说可以实现 N 对 N 的转换，Axon 提供了 one to one 和 one to many 的实现，这两种转换应该可以满足 90% 以上的需求了，我们在开发过程中本身也应该对删除事件谨慎一点，一般不会去删除一个事件，可以在消费的时候忽略就行了，而不是直接删除这个事件。

那么要进行升级，事件本身必须要有个版本的概念，Axon 提供了@Revision注解用以标明事件的版本，如果没有标注则为 null，由于之前的例子都是没有 version 的，那么我们往 contract 里面添加个字段 industryName。更新event、command、aggregate、model、view，然后给 AbstractEvent加上@Reivision(“1.0.0”)注解，启动工程并发送一个 GET contracts/{id} 请求原来已经建立的数据，这时候会发现 industryName 为 null，下面我们实现默认行业为 “互联网”。

由于大部分的升级都是针对同一个事件的增减字段，这里建立了SameEventUpCaster：

public abstract class SameEventUpCaster extends SingleEventUpcaster {protected boolean canUpcast(IntermediateEventRepresentation intermediateRepresentation) {return outputType(intermediateRepresentation.getType()) != null;}@Overrideprotected IntermediateEventRepresentation doUpcast(IntermediateEventRepresentation intermediateRepresentation) {return intermediateRepresentation.upcast(outputType(intermediateRepresentation.getType()),JsonNode.class,d -> this.doUpCastPayload(d, intermediateRepresentation),metaData -> this.doUpCastMetaData(metaData, intermediateRepresentation));}public SimpleSerializedType outputType(SerializedType originType) {return new SimpleSerializedType(eventTypeName(), outputRevision(originType.getRevision()));}public abstract String eventTypeName();public abstract String outputRevision(String originRevision);public abstract JsonNode doUpCastPayload(JsonNode document, IntermediateEventRepresentation intermediateEventRepresentation);public abstract MetaData doUpCastMetaData(MetaData document, IntermediateEventRepresentation intermediateEventRepresentation);}

建立一个处理事件转换的中心，目的是将转换操作分发下去:

public class ContractEventUpCaster extends SingleEventUpcaster {private static List<SameEventUpCaster> upCasters = Arrays.asList(new ContractCreatedEventUpCaster(),new ContractUpdatedEventUpCaster());@Overrideprotected boolean canUpcast(IntermediateEventRepresentation intermediateRepresentation) {return upCasters.stream().anyMatch(o -> o.canUpcast(intermediateRepresentation));}@Overrideprotected IntermediateEventRepresentation doUpcast(IntermediateEventRepresentation intermediateRepresentation) {SameEventUpCaster upCaster = upCasters.stream().filter(o -> o.canUpcast(intermediateRepresentation)).findAny().orElseThrow(RuntimeException::new);return upCaster.doUpcast(intermediateRepresentation);}
}

实现各个事件的具体升级方式：

public class ContractCreatedEventUpCaster extends SameEventUpCaster {@Overridepublic String eventTypeName() {return ContractCreatedEvent.class.getTypeName();}@Overridepublic String outputRevision(String originRevision) {final HashMap<String, String> revisionConvertMpp = new HashMap<>();revisionConvertMpp.put(null, "1.0.0");return revisionConvertMpp.get(originRevision);}@Overridepublic JsonNode doUpCastPayload(JsonNode document, IntermediateEventRepresentation intermediateEventRepresentation) {if (intermediateEventRepresentation.getType().getRevision() == null) {((ObjectNode) document).put("industryName", "互联网");}return document;}@Overridepublic MetaData doUpCastMetaData(MetaData document, IntermediateEventRepresentation intermediateEventRepresentation) {return document;}
}public class ContractUpdatedEventUpCaster extends SameEventUpCaster {@Overridepublic String eventTypeName() {return ContractUpdatedEvent.class.getTypeName();}@Overridepublic String outputRevision(String originRevision) {final HashMap<String, String> revisionConvertMpp = new HashMap<>();revisionConvertMpp.put(null, "1.0.0");return revisionConvertMpp.get(originRevision);}@Overridepublic JsonNode doUpCastPayload(JsonNode document, IntermediateEventRepresentation intermediateEventRepresentation) {// 每个版本只有一种升级方案，然后链式一步一步升级if (intermediateEventRepresentation.getType().getRevision() == null) {((ObjectNode) document).put("industryName", "互联网");}return document;}@Overridepublic MetaData doUpCastMetaData(MetaData document, IntermediateEventRepresentation intermediateEventRepresentation) {return document;}
}
添加配置：
@Bean
public EventStorageEngine eventStorageEngine(Serializer defaultSerializer,PersistenceExceptionResolver persistenceExceptionResolver,@Qualifier("eventSerializer") Serializer eventSerializer,EntityManagerProvider entityManagerProvider,EventUpcaster contractUpCaster,TransactionManager transactionManager) {return JpaEventStorageEngine.builder().snapshotSerializer(defaultSerializer).upcasterChain(contractUpCaster).persistenceExceptionResolver(persistenceExceptionResolver).eventSerializer(eventSerializer).entityManagerProvider(entityManagerProvider).transactionManager(transactionManager).build();
}
@Bean
public EventUpcaster contractUpCaster() {return new ContractEventUpCaster();
}

启动项目，再次请求 /contracts/{id} ，数据已经更新了。但是有个问题，那就是 view 视图的数据还没有更新，这部分的数据还是需要编写脚本去做升级的，暂时没有什么更好的办法。

Command 优化
自定义CommandGateway
Axon 提供的CommandGateway接口，我们看到任意的 object 都能发送，并且不能附带 MetaData，所以这里我们进行一个自定义：

public interface ContractCommandGateway {// fire and forgetvoid sendCommand(AbstractCommand command);// method that will wait for a result for 10 seconds@Timeout(value = 6, unit = TimeUnit.SECONDS)Long sendCommandAndWaitForAResult(AbstractCommand command);// method that will wait for a result for 10 seconds@Timeout(value = 6, unit = TimeUnit.SECONDS)void sendCommandAndWait(AbstractCommand command);// method that attaches meta data and will wait for a result for 10 seconds@Timeout(value = 6, unit = TimeUnit.SECONDS)ContractAggregate sendCommandAndWaitForAResult(AbstractCommand command,@MetaDataValue("userId") String userId);// this method will also wait, caller decides how longvoid sendCommandAndWait(AbstractCommand command, long timeout, TimeUnit unit) throws TimeoutException, InterruptedException;
}

增加重试机制，由于分布式的问题，event 在存储的时候还是会发生资源争夺，在 InnoDB 下的表现就是 A 节点存储了 seq 为 1 的 event，B 节点在 A 存储前内存中先读取到了 0 ，然后进行存储，这个时候就会有重复的风险，比较好的做法是将同一个 aggregate 的操作尽量分配到一个节点下面去处理，但是事实上完全避免是不太可能的，所以了这里我们需要一个重试机制去做补偿，Axon 也提供了这个机制，代码如下：
@Slf4j
public class CommandRetryScheduler implements RetryScheduler {

@Override
public boolean scheduleRetry(CommandMessage commandMessage, RuntimeException lastFailure, List<Class<? extends Throwable>[]> failures, Runnable commandDispatch) {log.info(MessageFormat.format("aggregate [{0}] execute [{1}] retry [{2}] time", commandMessage.getIdentifier(), commandMessage.getCommandName(), failures.size()));if (failures.size() > 2) {return false;}commandDispatch.run();return true;
}

}
截获 command 做一些事情，比如从 Security Context 中获取 user 信息并作为 MetaData 发送，或者为 CreateCommand 类型的 command 自动生成一个 ID，而不用我们手动去构建：

public interface MetaDataUserInterface {String getName();Long getUserId();Long getCustomerId();
}@Getter
@Setter
@Builder
public class MetaDataUser implements MetaDataUserInterface {private String name;private Long userId;private Long customerId;
}@AllArgsConstructor
@Configuration
public class CommandInterceptor implements MessageDispatchInterceptor {private final UIDGenerator uidGenerator;@Overridepublic BiFunction<Integer, GenericCommandMessage<AbstractCommand>, GenericCommandMessage<AbstractCommand>> handle(List messages) {return (index, message) -> {// create command 自动生成 IDif (message.getPayload() instanceof CreateContractCommand) {CreateContractCommand payload = (CreateContractCommand) message.getPayload();payload.setIdentifier(uidGenerator.getId());}// 添加 user info 作为 MetaData，由于项目不设计 security 这里就简单的附加一个假的用户Map<String, MetaDataUserInterface> map = new HashMap<>();map.put("user", MetaDataUser.builder().customerId(1L).name("Test").userId(2L).build());return map.isEmpty() ? message : message.andMetaData(map);};}
}

这里为了不涉及 Spring Security，我直接 new 了一个测试对象进去，大家可以取自己的 user。

最后添加配置代码：

@Bean
public ContractCommandGateway getCommandGateway(SimpleCommandBus simpleCommandBus, CommandInterceptor commandInterceptor) {return CommandGatewayFactory.builder().commandBus(simpleCommandBus).retryScheduler(new CommandRetryScheduler()).dispatchInterceptors(commandInterceptor).build().createGateway(ContractCommandGateway.class);
}

最后替换掉 command gateway 的调用：

private final ContractCommandGateway contractCommandGateway;@PostMapping
public Long createContract(@RequestBody @Valid CreateContractCommand command) {return contractCommandGateway.sendCommandAndWaitForAResult(command);
}@PutMapping("/{id}")
public void updateContract(@PathVariable("id") Long id, @RequestBody @Valid UpdateContractCommand command) {command.setIdentifier(id);contractCommandGateway.sendCommandAndWait(command);
}@DeleteMapping("/{id}")
public void deleteContract(@PathVariable("id") Long id) {contractCommandGateway.sendCommandAndWait(new DeleteContractCommand(id));
}

这里大家可能注意到，我将原来 async 的发送都改成了 sync 的，因为 async 的调用，如果过程异常，其实这个请求并不会丢出异常，sendCommandAndWaitForAResult command 可以有返回值，这个在官方文档中也有介绍，就不做过多的介绍。

实现可靠消息

什么是可靠消息
微服务盛行的时代下，消息成为了不可缺少的组件，首先我们看一个例子，contract 系统创建了一个合同，然后发送创建合同的消息。看似简单，实际上分析一下它的出错可能性，会有以下几种：

创建合同成功，发送消息失败；
创建合同失败，发送消息成功；
创建合同成功，发送消息成功；
创建合同失败，发送消息失败；
同时成功或者同时失败，这个情况是一致的，是正确的，我们需要关心的就是不一致的情况。那么最简单的办法，就是让创建合同和发送消息成为一个事务，要么一起成功，要么一起失败，但是这么做的话耦合性太强，本身合同创建成功了，却因为消息发送的失败强制回滚。这个时候，可能就想到了存储消息数据，将合同创建和消息数据的存储作为一个事务，消息发送成功之后再去删除消息数据，定期去扫描未发送的消息数据，来保证消息的发送。但是这么做还是有一定的代价的，需要实现消息的存储，消息存储和合同创建还是耦合在一起的，不过这样的模式到 Event Sourcing 下面那就比较理想了，因为本身消息数据和 event 是一样的，存储了 event 相当于完成了存储消息数据，只需要在 event 下做一个标记即可。

做完了上面这些，就能保证消息一定从 producer 到 broker 这一过程，当然要做到消息不丢，必然产生的结果就是消息可能会重复，情况就是 broker 收到了消息，但是没有通知到 producer，这种情况下 producer 是认为消息没有投递成功的，会出现重复投递的情况。保证了消息一定送达 broker 之后，就是 consumer 和 broker 的关系了，consumer 在消费之后需要告诉 broker 消费成功，否则 broker 需要一直保存这些消息。当然消费端可能需要做更多的事情，比如保证同一 aggregate 事件的顺序消费。下面文章会以在 Axon 框架上做一些拓展，以分别实现 consumer 和 producer。

拓展 DomainEventEntry
上面也说到了，在 Event Sourcing 模式下，我们只需要给事件加上一个是否投递的标志，这里我们就看看如何实现（这里只针对 JPA 做了实现）。

建立对应的 entity 以取代DomainEventEntry：

@Entity(name = "DomainEventEntry")
@Getter
@Setter
@Table(indexes = @Index(columnList = "aggregateIdentifier,sequenceNumber", unique = true))
public class CustomDomainEventEntry extends AbstractSequencedDomainEventEntry<byte[]> {@NotNull@Column(columnDefinition = "tinyint(1) default 0")private boolean sent = false;public CustomDomainEventEntry(DomainEventMessage<?> eventMessage, Serializer serializer) {super(eventMessage, serializer, byte[].class);this.setSent(false);}/*** Default constructor required by JPA*/protected CustomDomainEventEntry() {}
}

建立对应的 storage 以取代 JpaEventStorageEngine：

public class CustomJpaEventStorageEngine extends JpaEventStorageEngine {public CustomJpaEventStorageEngine(Builder builder) {super(builder);}@Overrideprotected Object createEventEntity(EventMessage<?> eventMessage, Serializer serializer) {return new CustomDomainEventEntry(asDomainEventMessage(eventMessage), serializer);}public static CustomJpaEventStorageEngine.Builder builder() {return new CustomJpaEventStorageEngine.Builder();}// 此处略去了 builder 部分代码public static class Builder extends JpaEventStorageEngine.Builder {...}
}

更新对应的 config：

@Bean
public EventStorageEngine eventStorageEngine(Serializer defaultSerializer,PersistenceExceptionResolver persistenceExceptionResolver,@Qualifier("eventSerializer") Serializer eventSerializer,EntityManagerProvider entityManagerProvider,EventUpcaster contractUpCaster,TransactionManager transactionManager) {return CustomJpaEventStorageEngine.builder().snapshotSerializer(defaultSerializer).upcasterChain(contractUpCaster).persistenceExceptionResolver(persistenceExceptionResolver).eventSerializer(eventSerializer).entityManagerProvider(entityManagerProvider).transactionManager(transactionManager).build();
}

handler 实现可靠消息的生产端
做好了准备工作再发送消息就比较清晰了，我们需要做的就是在事件存储后去尝试发送消息，然后标记为已发送即可，在之前的实现 CQRS 中我们留了一个坑，就是 view 端的更新不是在事件存储之后，这里我们就去实现发消息在事件存储之后，然后 view 层去监听消息更新。具体的实现就是利用 entity postPersist 去监听存储，在 transaction 成功后去尝试发送消息，代码如下：

@EntityListeners(CustomDomainEventEntryListener.class)
public class CustomDomainEventEntry extends AbstractSequencedDomainEventEntry<byte[]> {...
}@Component
@Slf4j
public class CustomDomainEventEntryListener {private static CustomDomainEventEntryRepository customDomainEventEntryRepository;private static ContractPublisher contractPublisher;@Autowiredpublic void init(CustomDomainEventEntryRepository customDomainEventEntryRepository, ContractPublisher contractPublisher) {this.customDomainEventEntryRepository = customDomainEventEntryRepository;this.contractPublisher = contractPublisher;}@PostPersistvoid onPersist(CustomDomainEventEntry entry) {TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {@Overridepublic void afterCompletion(int status) {if (status == TransactionSynchronization.STATUS_COMMITTED) {CompletableFuture.runAsync(() -> sendEvent(entry.getEventIdentifier()));}}});}@Transactionalpublic void sendEvent(String identifier) {CustomDomainEventEntry eventEntry = customDomainEventEntryRepository.findByEventIdentifier(identifier);if (!eventEntry.isSent()) {contractPublisher.sendEvent(eventEntry);eventEntry.setSent(true);customDomainEventEntryRepository.save(eventEntry);}}
}@Repository
public interface CustomDomainEventEntryRepository extends JpaRepository<CustomDomainEventEntry, String> {/*** 查找事件** @param identifier** @return*/CustomDomainEventEntry findByEventIdentifier(String identifier);
}@Component
@AllArgsConstructor
@Slf4j
public class ContractEventPublisher {public void sendEvent(DomainEvent event) {// use stream to send message herelog.info(MessageFormat.format("prepare to sending message : {0}]", new Gson().toJson(event)));}public void sendEvent(CustomDomainEventEntry event) {// use com.craftsman.eventsourcing.stream to send message hereObjectMapper mapper = new ObjectMapper();HashMap payload = null;HashMap metaData = null;try {payload = mapper.readValue(event.getPayload().getData(), HashMap.class);metaData = mapper.readValue(event.getMetaData().getData(), HashMap.class);} catch (Exception exception) {log.error(MessageFormat.format("byte[] to string failed; exception: {0}", exception));}if (payload == null || metaData == null) {log.warn(MessageFormat.format("nothing to send; exception: {0}", event.getEventIdentifier()));return;}DomainEvent domainEvent = new DomainEvent(event.getType(),event.getAggregateIdentifier(),event.getPayload().getType().getName(),event.getPayload().getType().getRevision(),event.getSequenceNumber(),event.getEventIdentifier(),event.getTimestamp(),payload,metaData);this.sendEvent(domainEvent);}
}

DomainEvent 是为了统一消息的格式包装的类，具体可以看代码这里就不贴了
ContractEventPublisher 作为消息统一发送出口，为了不涉及 rabbitmq 暂时以 log 的形式代替消息发送，后续在 Spring Cloud Stream 优化中实现完整的流程
实现消费端
从DomainEvent的属性中，我们可以看到有一个sequenceNumber字段，这个字段可以用来保证同一 aggregate 的事件顺序，那么在消费端可以以 type aggregate sequenceNumber 形成一张表，用来记录每个 aggregate 的最新状态，如果 aggregate 数据量比较大，也可以分表存储，一般 aggregate_id 索引之后，单表性能在百万级别，应该都没什么问题。这样在消费的时候先比较 sequenceNumber 差异，只消费差异值为 1 的事件，就可以保证同一 aggregate 的事件被顺序消费。之后会写篇关于 Spring Cloud Stream 的文章，用来作为服务之间的桥梁，并解决框架用 header 作为路由之后引起的问题，这里暂时不做深入。

Spring Cloud Stream 优化
有哪些问题
Spring Cloud Stream（以下简称 SCS ）是 Spring Cloud 提供的消息中间件的抽象，但是目前也就支持 kafka 和 rabbitmq，这篇文章主要会讨论一下如何让 SCS 更好的服务我们之前搭建的 Event Sourcing、CQRS 模型。以下是我在使用 SCS 的过程中存在的一些问题：

StreamListener用来做事件路由分发并不是很理想，SPEL 可能会写的很长（我尝试过用自定义注解代替原生的注解，从而达到简化的目的，但是会出现一些莫名其妙的事件混乱）。
如果配合之前的模型使用，我们需要保证消息的顺序消费，每个方法都需要去 check 事件的当前 seq，很不方便。
在没有 handler 处理某个 type 的事件时，框架会给出一个 warn，然而这个事件可能在 consumer 这里根本不关心。
解决方案
为了解决上面的问题，我们可以这么处理，先统一一个入口将 SCS 的消息接收，然后我们自己构建一个路由系统，将请求分发到我们自己定义的注解方法上，并且在这个过程中将 seq 的检查也给做了，大体的流程是这个样子的：

image

这样以上几点问题都会得到解决，下面我们来看看具体如何实现：

首先定义一个注解用于接受自己分发的事件：

@Target( {ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface StreamEventHandler {String[] payloadTypes() default {""};String[] types();
}

types 对应 Stream 本身 Inuput 的类型， payloadTypes 对应事件类型，比如 ContractCreated，我们要做的效果是这个 payloadTypes 可以不写，直接从方法的第一个参数读取 class 的 simapleName。

定义用于记录 aggregate sequenceNumber 的 entity 和 repository ：

@Entity
@Table(indexes = @Index(columnList = "aggregateIdentifier,type", unique = true))
@Getter
@Setter
@NoArgsConstructor
public class DomainAggregateSequence {@Id@GeneratedValueprivate Long id;private Long sequenceNumber;private Long aggregateIdentifier;private String type;
}@Repository
public interface DomainAggregateSequenceRepository extends JpaRepository<DomainAggregateSequence, Long> {/*** 根据 aggregate id 和 type 找到对应的记录** @param identifier* @param type** @return*/DomainAggregateSequence findByAggregateIdentifierAndType(Long identifier, String type);}

由于暂时没有找到监听所有已绑定 channel 的事件的方法，这里实现一个类提供一个 dispatch 的方法用于分发：

@Slf4j
@Component
@AllArgsConstructor
public class StreamDomainEventDispatcher implements BeanPostProcessor {private final ObjectMapper mapper;private final DomainAggregateSequenceRepository domainAggregateSequenceRepository;private HashMap<Object, List<Method>> beanHandlerMap = new HashMap<>();@Autowiredpublic StreamDomainEventDispatcher(ObjectMapper mapper, DomainAggregateSequenceRepository domainAggregateSequenceRepository) {this.mapper = mapper;this.domainAggregateSequenceRepository = domainAggregateSequenceRepository;}@Transactionalpublic void dispatchEvent(DomainEvent event, String type) {log.info(MessageFormat.format("message [{0}] received", event.getEventIdentifier()));// 1. 检查是否是乱序事件或者重复事件Long aggregateIdentifier = Long.parseLong(event.getAggregateIdentifier());String eventType = event.getType();Long eventSequence = event.getSequenceNumber();DomainAggregateSequence sequenceObject = domainAggregateSequenceRepository.findByAggregateIdentifierAndType(aggregateIdentifier, eventType);if (sequenceObject == null) {sequenceObject = new DomainAggregateSequence();sequenceObject.setSequenceNumber(eventSequence);sequenceObject.setAggregateIdentifier(aggregateIdentifier);sequenceObject.setType(eventType);} else if (sequenceObject.getSequenceNumber() + 1 != eventSequence) {// 重复事件，直接忽略if (sequenceObject.getSequenceNumber().equals(eventSequence)) {log.warn(MessageFormat.format("repeat event ignored, type[{0}] aggregate[{1}] seq[{2}] , ignored", event.getType(), event.getAggregateIdentifier(), event.getSequenceNumber()));return;}throw new StreamEventSequenceException(MessageFormat.format("sequence error, db [{0}], current [{1}]", sequenceObject.getSequenceNumber(), eventSequence));} else {sequenceObject.setSequenceNumber(eventSequence);}domainAggregateSequenceRepository.save(sequenceObject);// 2. 分发事件到各个 handlerbeanHandlerMap.forEach((key, value) -> {Optional<Method> matchedMethod = getMatchedMethods(value, type, event.getPayloadType());matchedMethod.ifPresent(method -> {try {invoke(key, method, event);} catch (IllegalAccessException | InvocationTargetException e) {throw new StreamHandlerException(MessageFormat.format("[{0}] invoke error", method.getName()), e);}});if (!matchedMethod.isPresent()) {log.info(MessageFormat.format("message [{0}] has no listener", event.getEventIdentifier()));}});log.info(MessageFormat.format("message [{0}] handled", event.getEventIdentifier()));}@Transactionalpublic Optional<Method> getMatchedMethods(List<Method> methods, String type, String payloadType) {// 这里应该只有一个方法，因为将 stream 的单个事件分成多个之后，无法保证一致性List<Method> results = methods.stream().filter(m -> {StreamEventHandler handler = m.getAnnotation(StreamEventHandler.class);List<String> types = new ArrayList<>(Arrays.asList(handler.types()));List<String> payloadTypes = new ArrayList<>(Arrays.asList(handler.payloadTypes()));types.removeIf(StringUtils::isBlank);payloadTypes.removeIf(StringUtils::isBlank);if (CollectionUtils.isEmpty(payloadTypes) && m.getParameterTypes().length != 0) {payloadTypes = Collections.singletonList(m.getParameterTypes()[0].getSimpleName());}boolean isTypeMatch = types.contains(type);String checkedPayloadType = payloadType;if (StringUtils.contains(checkedPayloadType, ".")) {checkedPayloadType = StringUtils.substringAfterLast(checkedPayloadType, ".");}boolean isPayloadTypeMatch = payloadTypes.contains(checkedPayloadType);return isTypeMatch && isPayloadTypeMatch;}).collect(Collectors.toList());if (results.size() > 1) {throw new StreamHandlerException(MessageFormat.format("type[{0}] event[{1}] has more than one handler", type, payloadType));}return results.size() == 1 ? Optional.of(results.get(0)) : Optional.empty();}@Transactionalpublic void invoke(Object bean, Method method, DomainEvent event) throws IllegalAccessException, InvocationTargetException {int count = method.getParameterCount();if (count == 0) {method.invoke(bean);} else if (count == 1) {Class<?> payloadType = method.getParameterTypes()[0];if (payloadType.equals(DomainEvent.class)) {method.invoke(bean, mapper.convertValue(event.getPayload(), DomainEvent.class));} else {method.invoke(bean, mapper.convertValue(event.getPayload(), payloadType));}} else if (count == 2) {Class<?> payloadType0 = method.getParameterTypes()[0];Class<?> payloadType1 = method.getParameterTypes()[1];Object firstParameterValue = mapper.convertValue(event.getPayload(), payloadType0);Object secondParameterValue = event.getMetaData();// 如果是 DomainEvent 类型则优先传递该类型，另外一个参数按照 payloadType > metaData 优先级传入if (payloadType0.equals(DomainEvent.class)) {firstParameterValue = mapper.convertValue(event, payloadType0);secondParameterValue = mapper.convertValue(event.getPayload(), payloadType1);}if (payloadType1.equals(DomainEvent.class)) {secondParameterValue = mapper.convertValue(event, payloadType1);}method.invoke(bean, firstParameterValue, secondParameterValue);}}@Overridepublic Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {return bean;}@Overridepublic Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {Class<?> targetClass = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean) : bean.getClass();Method[] uniqueDeclaredMethods = ReflectionUtils.getUniqueDeclaredMethods(targetClass);List<Method> methods = new ArrayList<>();for (Method method : uniqueDeclaredMethods) {StreamEventHandler streamListener = AnnotatedElementUtils.findMergedAnnotation(method,StreamEventHandler.class);if (streamListener != null) {methods.add(method);}}if (!CollectionUtils.isEmpty(methods)) {beanHandlerMap.put(bean, methods);}return bean;}}

这里参照了 SCS 本身手机 handler的方式，会将有 @StreamEventHandler 注解的方法都找出来做一个记录。在 dispatchEvent 的时候会更新事件的 seq 并且按照 type 去调用各个标有注解的方法。

实现一个比较简单的例子：

@Slf4j
@Component
@Transactional
@AllArgsConstructor
public class DomainEventDispatcher {private final StreamDomainEventDispatcher streamDomainEventDispatcher;@StreamListener(target = ChannelDefinition.CONTRACTS_INPUT, condition = "headers['messageType']=='eventSourcing'")public void handleBuilding(@Payload DomainEvent event) {streamDomainEventDispatcher.dispatchEvent(event, ChannelDefinition.CONTRACTS_INPUT);}
}@Component
@AllArgsConstructor
@Transactional
public class ContractEventHandler {@StreamEventHandler(types = ChannelDefinition.CONTRACTS_INPUT)public void handle(ContractCreatedEvent event) {// 实现你的 view 层更新业务}
}

注意：

AbstractDomainEventDispatcher中监听所有 bean 加载完成不能用 InitializingBean 接口，否则@Transactional会失效，这个有兴趣的同学可以研究一下@Transactional的机制。
至此以上几点就优化完了。

其他优化
错误处理
基于 SCS 的默认配置，存在一个致命的问题，那就是当消息处理失败（重试三次）之后，消息直接没了，这个相当于就是消息丢失了。那么解决方案其实也是比较简单的，一般有两种解决方案：

拒绝这个消息，丢在 broker 原先的队列里。
将这个消息记录到一个错误的 queue 中等待修复，后续可能将消息转发回去，也可能直接就删除了消息（比如重复的消息）。
方案 1 这么做可能会出的问题就是，这个消息反复消费，反复失败，引起循环问题从而导致服务出现问题，这个就需要在 broker 做一些策略配置了，为了让 broker 尽可能的简单，我们这里采用方案 2，要实现的流程是这样的：

首先让 SCS 为我们自动生成一个 DLQ
spring:
application:
name: event-sourcing-service
datasource:
url: jdbc:mysql://localhost:3306/event?useUnicode=true&autoReconnect=true&rewriteBatchedStatements=TRUE
username: root
password: root
jpa:
hibernate:
ddl-auto: update
use-new-id-generator-mappings: false
show-sql: false
properties:
hibernate.dialect: org.hibernate.dialect.MySQL55Dialect
rabbitmq:
host: localhost
port: 5672
username: creams_user
password: Souban701
cloud:
stream.bindings:
contract-events: # 这个名字对应代码中@input(“value”) 的 value
destination: contract-events # 这个对应 rabbit 中的 channel
contentType: application/json # 这个指定传输类型，其实可以默认指定，但是目前每个地方都写了，所以统一下

      contract-events-input:destination: contract-eventscontentType: application/jsongroup: event-sourcing-servicedurableSubscription: truestream.rabbit.bindings.contract-events-input.consumer:autoBindDlq: truerepublishToDlq: truedeadLetterQueueName: contract-error.dlq
logging:level.org:springframework:web: INFOcloud.sleuth: INFOapache.ibatis: DEBUGjava.sql: DEBUGhibernate:SQL: DEBUGtype.descriptor.sql: TRACEaxon:serializer:general: jackson

加上这个配置之后，rabbit 会给这个队列创建一个 .dlq 后缀的队列，异常消息都会被塞到这个队列里面（消息中包含了异常信息以及来源），等待我们处理，deadLetterQueueName指定了 DLQ 的名称，这样所有的失败消息都会存放到同一个 queue中。大部分的情况下，消息的异常都是由于 consumer 逻辑错误引起的，所以我们需要一个处理这些失败的消息的地方，比如在启动的时候自动拉取 DLQ 中的消息然后转发到原来的 queue 中去远程原有的业务逻辑，如果处理不了那么还是会继续进入到 DLQ 中。

在启动的时候拉取 DLQ 中的消息转发到原来的 queue 中。

@Component
public class DLXHandler implements ApplicationListener<ContextRefreshedEvent>, ApplicationContextAware {private final RabbitTemplate rabbitTemplate;private ApplicationContext applicationContext;private static final String DLQ = "contract-error.dlq";@Autowiredpublic DLXHandler(RabbitTemplate rabbitTemplate) {this.rabbitTemplate = rabbitTemplate;}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) {this.applicationContext = applicationContext;}@Overridepublic void onApplicationEvent(ContextRefreshedEvent event) {// SCS 会创建一个 child context ，这里需要判断下正确的 context 初始化完成if (event.getApplicationContext().equals(this.applicationContext)) {// 启动后获取 dlq 中所有的消息，进行消费Message message = rabbitTemplate.receive(DLQ);while (message != null) {rabbitTemplate.send(message.getMessageProperties().getReceivedRoutingKey(), message);message = rabbitTemplate.receive(DLQ);}}}
}

由于 SCS 没有提供给我们类似的接口，这里使用了 rabbitmq 的接口来获取消息。

完善之前的 CQRS 例子
经常上述这些基础操作之后，汇过来实现 CQRS 就比较清晰了，只需要监听相关的事件，然后更新视图层即可。

添加时间的监听

@StreamEventHandler(types = ChannelDefinition.CONTRACTS_INPUT)
public void handle(ContractCreatedEvent event, DomainEvent<ContractCreatedEvent, HashMap> domainEvent) {QueryContractCommand command = new QueryContractCommand(event.getIdentifier(), domainEvent.getTimestamp());ContractAggregate aggregate = queryGateway.query(command, ContractAggregate.class).join();ContractView view = new ContractView();view.setIndustryName(aggregate.getIndustryName());view.setId(aggregate.getIdentifier());view.setPartyB(aggregate.getPartyB());view.setPartyA(aggregate.getPartyA());view.setName(aggregate.getName());view.setDeleted(aggregate.isDeleted());contractViewRepository.save(view);
}

StreamDomainEventDispatcher 对传参做了一些处理，当有两个参数的时候会将 DomainEvent 传递，因为有些时候可能会用到一些字段，比如时间、附加信息等等。这里在消费事件的时候，可以根据时间去查询 aggregate 的状态，然后直接做一个映射，也可以根据事件直接对 view 层做 CUD ，个人觉得在性能和速度不存在大问题的时候直接去查询一下 aggregate 当时的状态做一个映射即可，毕竟比较简单。

删除原来的 ContractViewHandler 即可。
服务优化
失败消息的补偿机制
由于消息存在发送失败的情况，比如 broker 临时下线或者不可用了，尽管这种情况很少，我们最好做一个机制可以定期或者手动检查，并且尝试自己发送，这里我们就来实现这个机制。

提供未发送的 event 查询
在CustomDomainEventEntryRepository中加入：

/*** 查找未发送的事件** @param pageable** @return*/
Page<CustomDomainEventEntry> findBySentFalse(Pageable pageable);/*** 查询未发送事件的数量* @return*/
Long countBySentFalse();
将未发送事件的数量集成到 actuator，让我们可以事实看到失败消息的数量：
@Component
@AllArgsConstructor
public class EventHealthContributor implements InfoContributor {private final CustomDomainEventEntryRepository customDomainEventEntryRepository;@Overridepublic void contribute(Info.Builder builder) {Long count = customDomainEventEntryRepository.countBySentFalse();builder.withDetail("failedMessage", count);}
}

打开 http://localhost:8080/actuator/info 应该就可以看到我们的失败消息数量。

定期检查并且自动发送
建立对应的 service 和 controller：

@Service
@Slf4j
@AllArgsConstructor
public class ScheduleService {private final CustomDomainEventEntryRepository customDomainEventEntryRepository;private final ContractEventPublisher contractEventPublisher;@Scheduled(cron = "0 0 12 * * ?")@SchedulerLock(name = "failedMessageDiscoveryTask")public void failedMessageDiscovery() {Integer page = 0;PageRequest request = PageRequest.of(page, 1000);Page<CustomDomainEventEntry> results = customDomainEventEntryRepository.findBySentFalse(request);log.warn(MessageFormat.format("发现 [{0}] 条失败消息，尝试重新发送", results.getTotalElements()));sendFailedMessage(results.getContent());while (results.hasNext()) {request = PageRequest.of(page + 1, 1000);results = customDomainEventEntryRepository.findBySentFalse(request);sendFailedMessage(results.getContent());}log.info("所有失败消息尝试发送完毕");}private void sendFailedMessage(Collection<CustomDomainEventEntry> failedEvents) {failedEvents.forEach(e -> {contractEventPublisher.sendEvent(e);e.setSent(true);customDomainEventEntryRepository.save(e);});}
}

有些时候我们可能需要自己触发一个修复操作，可以把这个写成 API：

/*** 用于修复 view 视图和 aggregate 的不一致性，以及未发送消息的重试*/
@Slf4j
@AllArgsConstructor
@RestController
@RequestMapping("/repair")
public class DataRepairController {private final ScheduleService scheduleService;private static final String SECRET = "e248b98418db4cdcb069e8a1c08f6bb7";@GetMapping("/message")@Asyncpublic void repairMessage(@RequestParam("secret") String secret) {if (!StringUtils.equals(secret, SECRET)) {return;}scheduleService.failedMessageDiscovery();}
}

pom中加入：


<dependency><groupId>net.javacrumbs.shedlock</groupId><artifactId>shedlock-spring</artifactId><version>1.0.0</version>
</dependency>
<dependency><groupId>net.javacrumbs.shedlock</groupId><artifactId>shedlock-provider-jdbc-template</artifactId><version>1.0.0</version>
</dependency>

数据库中执行脚本：

CREATE TABLE shedlock(name VARCHAR(64), lock_until TIMESTAMP(3) NULL, locked_at TIMESTAMP(3) NULL, locked_by  VARCHAR(255), PRIMARY KEY (name)
)

使用SchedulerLock定义了每晚 12 点开始检查并自动发送，由于 SchedulerLock 在集群下的问题，这里使用了 shedlock 加锁，使得只有一个实例会执行该代码。这样就完成了自动修复与手动修复的接口暴露。

view 和 aggregate 之间的不一致补偿
在并发比较高的时候，可能会出现同时 update 一条记录的情况，这个时候需要上锁，JPA 可以自动创建并管理乐观锁，乐观锁会在 update 同一条记录的时候直接返回一个错误，我们只需要在 entity 上加上 version字段即可：

@Entity
@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
public class ContractView implements ContractInterface {@Id@Column(length = 64)private Long id;private String name;private String partyA;private String partyB;private boolean deleted = false;private String industryName;private long sequenceNumber;@Versionprivate Long version;
}

某些情况下我们可能还需要手动干预 aggregate 和 view 之间的一致性问题，比如在线上运行中莫名其妙有一个 id 为 xxx 的数据出现了不一致的情况，可能是在某些极端情况下消费端出现了逻辑错误导致了数据错误，但是又很难找到这批数据的特征。这个时候其实可以写一个 DataRepair 的 API，专门重新生成 view，或者说范围性的重新生成 view（因为真正在生产环境全部重新生成 view，必定是很耗时的一件事情），所以如果之前的 view 层代码是通过读取 aggregate 状态之后做映射的话，这里就会方便很多。比如：

@Service
@AllArgsConstructor
public class ContractViewService {private final QueryGateway queryGateway;private final ContractViewRepository contractViewRepository;public void updateViewFromAggregateById(Long aggregateIdentifier, Instant time) {QueryContractCommand command = new QueryContractCommand(aggregateIdentifier, time);ContractAggregate aggregate = queryGateway.query(command, ContractAggregate.class).join();ContractView view = contractViewRepository.findById(aggregateIdentifier).orElse(new ContractView());ContractAggregateViewMapper.mapAggregateToView(aggregate, view);contractViewRepository.save(view);}
}@Component
@AllArgsConstructor
@Transactional
public class ContractEventHandler {private final ContractViewService contractViewService;@StreamEventHandler(types = ChannelDefinition.CONTRACTS_INPUT)public void handle(ContractCreatedEvent event, DomainEvent<ContractCreatedEvent, HashMap> domainEvent) {contractViewService.updateViewFromAggregateById(event.getIdentifier(), domainEvent.getTimestamp());}
}@Slf4j
@AllArgsConstructor
@RestController
@RequestMapping("/repair")
public class DataRepairController {private final ScheduleService scheduleService;private final ContractViewService contractViewService;private static final String SECRET = "e248b98418db4cdcb069e8a1c08f6bb7";@GetMapping("/message")@Asyncpublic void repairMessage(@RequestParam("secret") String secret) {if (!StringUtils.equals(secret, SECRET)) {return;}scheduleService.failedMessageDiscovery();}@PostMapping("/aggregate")@Asyncpublic void repairAggregate(@RequestParam("secret") String secret, Long aggregateIdentifier) {if (!StringUtils.equals(secret, SECRET)) {return;}contractViewService.updateViewFromAggregateById(aggregateIdentifier, Instant.now());}
}

事件失败之后的补偿
在updateContractView的时候可能会出现各种异常，由于 view 是消费消息处理的，所以重试机制就在 SCS 和 rabbitmq 这里了，默认是重试三次。

实现分布式 CommandBus
为什么需要分布式
前面我们也提到过，为了防止资源争夺等问题的出现，最好尽可能的保证同一 aggregate 的内容让同一个 service 去处理。这个时候就需要分布式 command bus 了，我们知道 Spring Cloud 都是以 http 通讯的，这种一般请求的分布不受我们控制，好在 Axon 框架为我们提供了和 Spring Cloud 融合的功能，下面就看看具体怎么实现。

Spring Cloud Connector 实现
Spring Cloud Connector 实际上就是在每个节点用 ServiceInstance.Metadata 记录了自己的 routing 规则来让别的节点知道如何去做 routing ，但是在某些服务发现的实现下 ServiceInstance.Metadata 是不可编辑的，这个时候就会在生成一个 API 来返回策略（理论上效率应该低很多，毕竟要走下请求）。Axon 对配置支持的也比较好：

<dependency><groupId>org.axonframework.extensions.springcloud</groupId><artifactId>axon-springcloud</artifactId><version>4.1</version>
</dependency>

然后配置文件中将分布式 command 打开就好了：

axon:serializer:general: jacksondistributed:enabled: truespring-cloud:fallback-to-http-get: truefallback-url: /axon-routing

完整例子：

https://github.com/soooban/AxonDemo