Event Sourcing和CQRS落地（四）：深入使用-Axon-InfoQ

GTLC全球技术领导力峰会·上海站，首批讲师正式上线！了解详情 



 写点什么

在本系列的前一篇文章中，我们介绍了 CQRS 实现，以数据库为例使用 JPA 去做 View 的 ORM，本文将重点介绍 Axon 实现。

深入使用 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<T> upcast(Stream<T> 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;    }
    @Override    protected 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()    );
    @Override    protected boolean canUpcast(IntermediateEventRepresentation intermediateRepresentation) {        return upCasters.stream().anyMatch(o -> o.canUpcast(intermediateRepresentation));    }
    @Override    protected 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 {

    @Override    public String eventTypeName() {        return ContractCreatedEvent.class.getTypeName();    }
    @Override    public String outputRevision(String originRevision) {        final HashMap<String, String> revisionConvertMpp = new HashMap<>();        revisionConvertMpp.put(null, "1.0.0");
        return revisionConvertMpp.get(originRevision);    }
    @Override    public JsonNode doUpCastPayload(JsonNode document, IntermediateEventRepresentation intermediateEventRepresentation) {
        if (intermediateEventRepresentation.getType().getRevision() == null) {            ((ObjectNode) document).put("industryName", "互联网");        }
        return document;    }
    @Override    public MetaData doUpCastMetaData(MetaData document, IntermediateEventRepresentation intermediateEventRepresentation) {        return document;    }}
public class ContractUpdatedEventUpCaster extends SameEventUpCaster {

    @Override    public String eventTypeName() {        return ContractUpdatedEvent.class.getTypeName();    }
    @Override    public String outputRevision(String originRevision) {        final HashMap<String, String> revisionConvertMpp = new HashMap<>();        revisionConvertMpp.put(null, "1.0.0");
        return revisionConvertMpp.get(originRevision);    }
    @Override    public JsonNode doUpCastPayload(JsonNode document, IntermediateEventRepresentation intermediateEventRepresentation) {
        // 每个版本只有一种升级方案，然后链式一步一步升级        if (intermediateEventRepresentation.getType().getRevision() == null) {            ((ObjectNode) document).put("industryName", "互联网");        }
        return document;    }
    @Override    public 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 forget    void 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 long    void sendCommandAndWait(AbstractCommand command, long timeout, TimeUnit unit) throws TimeoutException, InterruptedException;}

复制代码

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

@Slf4jpublic 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@Builderpublic class MetaDataUser implements MetaDataUserInterface {
    private String name;
    private Long userId;
    private Long customerId;}
@AllArgsConstructor@Configurationpublic class CommandInterceptor implements MessageDispatchInterceptor {
    private final UIDGenerator uidGenerator;
    @Override    public BiFunction<Integer, GenericCommandMessage<AbstractCommand>, GenericCommandMessage<AbstractCommand>> handle(List messages) {        return (index, message) -> {
            // create command 自动生成 ID            if (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 可以有返回值，这个在官方文档中也有介绍，就不做过多的介绍。完整的例子 - branch session5

作者介绍：

周国勇，目前就职于杭州匠人网络创业，致力于楼宇资产管理的 SaaS 化，负责后端业务架构设计、项目管理，喜欢对业务模型的分析，热衷新技术的探索和实践，经常在踩坑的路上越走越远。

《Event Sourcing 和 CQRS 落地（二）：Event-Sourcing 实现》

《Event Sourcing 和 CQRS 落地（三）：CQRS 实现》