CQRS Event Sourcing之简单场景落地分析

前言

CQRS全称为Command Query Responsibility Segregation，是领域驱动编程思想中的一个概念，当然也可以脱离DDD，当作读写分离去使用。

传统Rest模式中，DTO -> PO基本上是一样的，是一种面向数据库模型编程，且读和写操作的模型耦合，也不太方便将领域数据映射到页面显示。

CQRS将读和写分为Query与Command。
其中Command属于写操作，应该声明为void 或者返回id。
其中Query属于读操作，不应该存在修改状态行为，返回具体数据类型。

简单应用

首先抽象出Command和CommandHandler的概念，前者代表命令，后者代表命令处理者，Query同理。

public interface Command<R> {
}

public interface CommandHandler<R, C extends Command<R>> {
   /**
    * command handle
    *
    * @param command
    * @return
    */
   R handle(C command);
}

public interface Query<R> {
}

public interface QueryHandler<R, C extends Query<R>> {
   /**
    * query handle
    *
    * @param query
    * @return
    */
   R handle(C query);
}

基于Spring实现的话，可以使用IOC容器现成的applicationContext工厂实现Command Handler打表。

public class CommandProvider<H extends CommandHandler<?, ?>> {
    private final ApplicationContext applicationContext;
    private final Class<H> type;

    CommandProvider(ApplicationContext applicationContext, Class<H> type) {
        this.applicationContext = applicationContext;
        this.type = type;
    }

    public H get() {
        return applicationContext.getBean(type);
    }
}

public class QueryProvider<H extends QueryHandler<?, ?>> {
    private final ApplicationContext applicationContext;
    private final Class<H> type;

    QueryProvider(ApplicationContext applicationContext, Class<H> type) {
        this.applicationContext = applicationContext;
        this.type = type;
    }

    public H get() {
        return applicationContext.getBean(type);
    }
}

public class CommandHandlerRegistrar {
    private Map<Class<? extends Command>, CommandProvider> commandProviderMap = new HashMap<>();
    private Map<Class<? extends Query>, QueryProvider> queryProviderMap = new HashMap<>();

    public CommandHandlerRegistrar(ApplicationContext applicationContext) {
        String[] names = applicationContext.getBeanNamesForType(CommandHandler.class);
        for (String name : names) {
            registerCommand(applicationContext, name);
        }
        names = applicationContext.getBeanNamesForType(QueryHandler.class);
        for (String name : names) {
            registerQuery(applicationContext, name);
        }
    }

    private void registerCommand(ApplicationContext applicationContext, String name) {
        Class<CommandHandler<?, ?>> handlerClass = (Class<CommandHandler<?, ?>>)applicationContext.getType(name);
        Class<?>[] generics = GenericTypeResolver.resolveTypeArguments(handlerClass, CommandHandler.class);
        Class<? extends Command> commandType = (Class<? extends Command>)generics[1];
        commandProviderMap.put(commandType, new CommandProvider(applicationContext, handlerClass));
    }

    private void registerQuery(ApplicationContext applicationContext, String name) {
        Class<QueryHandler<?, ?>> handlerClass = (Class<QueryHandler<?, ?>>)applicationContext.getType(name);
        Class<?>[] generics = GenericTypeResolver.resolveTypeArguments(handlerClass, QueryHandler.class);
        Class<? extends Query> queryType = (Class<? extends Query>)generics[1];
        queryProviderMap.put(queryType, new QueryProvider(applicationContext, handlerClass));
    }

    @SuppressWarnings("unchecked")
    <R, C extends Command<R>> CommandHandler<R, C> getCmd(Class<C> commandClass) {
        return commandProviderMap.get(commandClass).get();
    }

    @SuppressWarnings("unchecked")
    <R, C extends Query<R>> QueryHandler<R, C> getQuery(Class<C> commandClass) {
        return queryProviderMap.get(commandClass).get();
    }
}

再抽象出EventBus

public interface EventBus {
    /**
     * command
     *
     * @param command
     * @param <R>
     * @param <C>
     * @return
     */
    <R, C extends Command<R>> R executeCommand(C command);

    /**
     * query
     *
     * @param query
     * @param <R>
     * @param <Q>
     * @return
     */
    <R, Q extends Query<R>> R executeQuery(Q query);
}

public class SpringEventBus implements EventBus {
    private final CommandHandlerRegistrar registry;

    public SpringEventBus(CommandHandlerRegistrar registry) {
        this.registry = registry;
    }

    @Override
    public <R, C extends Command<R>> R executeCommand(C command) {
        CommandHandler<R, C> commandHandler = (CommandHandler<R, C>)registry.getCmd(command.getClass());
        return commandHandler.handle(command);
    }

    @Override
    public <R, Q extends Query<R>> R executeQuery(Q query) {
        QueryHandler<R, Q> queryHandler = (QueryHandler<R, Q>)registry.getQuery(query.getClass());
        return queryHandler.handle(query);
    }
}

@Configuration即完成了Command Handler注册发现。

    @Bean
    public CommandHandlerRegistrar registry(ApplicationContext applicationContext) {
        return new CommandHandlerRegistrar(applicationContext);
    }

    @Bean
    public EventBus commandBus(CommandHandlerRegistrar registry) {
        return new SpringEventBus(registry);
    }

然后在Controller层就可以直接依赖EventBus做读写处理，替换以前的service操作。

@RestController
@RequiredArgsConstructor
public class PoliciesController {
    private final EventBus bus;

    @PostMapping
    public ResponseEntity<CreatePolicyResult> createPolicy(@RequestBody CreatePolicyCommand command) {
        return ok(bus.executeCommand(command));
    }

    @PostMapping("/confirmTermination")
    public ResponseEntity<ConfirmTerminationResult> terminatePolicy(@RequestBody ConfirmTerminationCommand command) {
        return ok(bus.executeCommand(command));
    }

    @PostMapping("/confirmBuyAdditionalCover")
    public ResponseEntity<ConfirmBuyAdditionalCoverResult> buyAdditionalCover(@RequestBody ConfirmBuyAdditionalCoverCommand command) {
        return ok(bus.executeCommand(command));
    }

    @PostMapping("/find")
    public Collection<PolicyInfoDto> find(@RequestBody FindPoliciesQuery query) {
        return bus.executeQuery(query);
    }

    @GetMapping("/details/{policyNumber}/versions")
    public ResponseEntity<PolicyVersionsListDto> getPolicyVersions(@PathVariable String policyNumber) {
        return ok(bus.executeQuery(new GetPolicyVersionsListQuery(policyNumber)));
    }

    @GetMapping("/details/{policyNumber}/versions/{versionNumber}")
    public ResponseEntity<PolicyVersionDto> getPolicyVersionDetails(@PathVariable String policyNumber, @PathVariable int versionNumber) {
        return ok(bus.executeQuery(new GetPolicyVersionDetailsQuery(policyNumber, versionNumber)));
    }

}

这里是一个Command和Query操作分发的实现雏形，有几点细节。

（1） EventBus实现有多种方式，Controller依赖抽象即可替换，本质是Scan到所有CommandHandler子类以后打一张map表，key是Command Class，value是CommandProvider工厂。这里自研注解在ImportBeanDefinitionRegistrar流程操作BeanDefinition也可以，自己用scanner跳过spring打表也可以。

（2）Bus就不区分Command和Query了，他属于dispatcher。

（3）读和写的模型分开了，写入参Command实现类，读入参Query实现类。

往下看一下Handler逻辑

@Component
@Transactional(rollbackFor = Throwable.class)
@RequiredArgsConstructor
public class CreatePolicyHandler implements CommandHandler<CreatePolicyResult, CreatePolicyCommand> {

    private final OfferRepository offerRepository;
    private final PolicyRepository policyRepository;
    private final EventPublisher eventPublisher;

    @Override
    public CreatePolicyResult handle(CreatePolicyCommand command) {
        Offer offer = offerRepository.withNumber(command.getOfferNumber());
        Policy policy = Policy.convertOffer(offer, UUID.randomUUID().toString(), command.getPurchaseDate(), command.getPolicyStartDate());
        policyRepository.add(policy);

        eventPublisher.publish(new PolicyEvents.PolicyCreated(this, policy));

        return new CreatePolicyResult(policy.getNumber());
    }
}

Repository和EventPublisher都属于抽象，可替换实现。

看一下领域对象和Event。

@Entity
@Getter
@NoArgsConstructor
@AllArgsConstructor
public class Policy {

    @Id
    @GeneratedValue
    private UUID id;
    private String number;
    @ManyToOne(optional = false)
    private Product product;
    @OneToMany(cascade = CascadeType.ALL)
    private List<PolicyVersion> versions = new ArrayList<>();
    private LocalDate purchaseDate;

    public Policy(UUID uuid, String policyNumber, Product product, LocalDate purchaseDate) {
        this.id = uuid;
        this.number = policyNumber;
        this.product = product;
        this.purchaseDate = purchaseDate;
    }

    public static Policy convertOffer(
        Offer offer,
        String policyNumber,
        LocalDate purchaseDate,
        LocalDate policyStartDate) {
        if (offer.isConverted()) { throw new BusinessException("Offer already converted"); }

        if (offer.isRejected()) { throw new BusinessException("Offer already rejected"); }

        if (offer.isExpired(purchaseDate)) { throw new BusinessException("Offer expired"); }

        if (offer.isExpired(policyStartDate)) { throw new BusinessException("Offer not valid at policy start date"); }

        Policy
            newPolicy = new Policy(
            UUID.randomUUID(),
            policyNumber,
            offer.getProduct(),
            purchaseDate
        );

        newPolicy.addFirstVersion(offer, purchaseDate, policyStartDate);
        newPolicy.confirmChanges(1);

        return newPolicy;
    }

    public void extendCoverage(LocalDate effectiveDateOfChange, CoverPrice newCover) {
        //preconditions
        if (isTerminated()) { throw new BusinessException("Cannot annex terminated policy"); }

        Optional<PolicyVersion> versionAtEffectiveDate = getPolicyVersions().effectiveAtDate(effectiveDateOfChange);
        if (!versionAtEffectiveDate.isPresent()) { throw new BusinessException("No active version at given date"); }

        PolicyVersion annexVer = addNewVersionBasedOn(versionAtEffectiveDate.get(), effectiveDateOfChange);
        annexVer.addCover(newCover, effectiveDateOfChange, annexVer.getCoverPeriod().getTo());
    }

    private boolean isTerminated() {
        return versions.stream().anyMatch(v -> v.isActive() && PolicyStatus.Terminated.equals(v.getPolicyStatus()));
    }

    public void terminatePolicy(LocalDate effectiveDateOfChange) {
        if (isTerminated()) { throw new BusinessException("Policy already terminated"); }

        Optional<PolicyVersion> versionAtEffectiveDateOpt = getPolicyVersions().effectiveAtDate(effectiveDateOfChange);
        if (!versionAtEffectiveDateOpt.isPresent()) { throw new BusinessException("No active version at given date"); }

        PolicyVersion versionAtEffectiveDate = versionAtEffectiveDateOpt.get();

        if (!versionAtEffectiveDate.getCoverPeriod().contains(effectiveDateOfChange)) {
            throw new BusinessException("Cannot terminate policy at given date as it is not withing cover period");
        }

        PolicyVersion termVer = addNewVersionBasedOn(versionAtEffectiveDate, effectiveDateOfChange);
        termVer.endPolicyOn(effectiveDateOfChange.minusDays(1));
    }

    public void cancelLastAnnex() {
        PolicyVersion lastActiveVer = getPolicyVersions().latestActive();
        if (lastActiveVer == null) { throw new BusinessException("There are no annexed left to cancel"); }

        lastActiveVer.cancel();
    }

    public void confirmChanges(int versionToConfirmNumber) {
        Optional<PolicyVersion> versionToConfirm = getPolicyVersions().withNumber(versionToConfirmNumber);
        if (!versionToConfirm.isPresent()) { throw new BusinessException("Version not found"); }

        versionToConfirm.get().confirm();
    }

    private void addFirstVersion(Offer offer, LocalDate purchaseDate, LocalDate policyStartDate) {
        PolicyVersion
            ver = new PolicyVersion(
            UUID.randomUUID(),
            1,
            PolicyStatus.Active,
            DateRange.between(policyStartDate, policyStartDate.plus(offer.getCoverPeriod())),
            DateRange.between(policyStartDate, policyStartDate.plus(offer.getCoverPeriod())),
            offer.getCustomer().copy(),
            offer.getDriver().copy(),
            offer.getCar().copy(),
            offer.getTotalCost(),
            offer.getCovers()
        );

        versions.add(ver);
    }

    private PolicyVersion addNewVersionBasedOn(
        PolicyVersion versionAtEffectiveDate, LocalDate effectiveDateOfChange) {
        PolicyVersion
            newVersion = new PolicyVersion(
            versionAtEffectiveDate,
            getPolicyVersions().maxVersionNumber() + 1,
            effectiveDateOfChange);

        versions.add(newVersion);
        return newVersion;
    }

    public PolicyVersions getPolicyVersions() {
        return new PolicyVersions(versions);
    }

    public enum PolicyStatus {
        Active,
        Terminated
    }
}

public class PolicyEvents {

    @Getter
    public static class PolicyCreated extends Event {
        private Policy newPolicy;

        public PolicyCreated(Object source, Policy newPolicy) {
            super(source);
            this.newPolicy = newPolicy;
        }
    }

    @Getter
    public static class PolicyAnnexed extends Event {
        private Policy annexedPolicy;
        private PolicyVersion annexVersion;

        public PolicyAnnexed(
            Object source, Policy annexedPolicy, PolicyVersion annexVersion) {
            super(source);
            this.annexedPolicy = annexedPolicy;
            this.annexVersion = annexVersion;
        }
    }

    @Getter
    public static class PolicyTerminated extends Event {
        private Policy terminatedPolicy;
        private PolicyVersion terminatedVersion;

        public PolicyTerminated(
            Object source, Policy terminatedPolicy, PolicyVersion terminatedVersion) {
            super(source);
            this.terminatedPolicy = terminatedPolicy;
            this.terminatedVersion = terminatedVersion;
        }
    }

    @Getter
    public static class PolicyAnnexCancelled extends Event {
        private Policy policy;
        private PolicyVersion cancelledAnnexVersion;
        private PolicyVersion currentVersionAfterAnnexCancellation;

        public PolicyAnnexCancelled(Object source,
            Policy policy,
            PolicyVersion cancelledAnnexVersion,
            PolicyVersion currentVersionAfterAnnexCancellation) {
            super(source);
            this.policy = policy;
            this.cancelledAnnexVersion = cancelledAnnexVersion;
            this.currentVersionAfterAnnexCancellation = currentVersionAfterAnnexCancellation;
        }
    }
}

相应的EventHandler:

@Component
@RequiredArgsConstructor
class PolicyEventsProjectionsHandler {

    private final PolicyInfoDtoProjection policyInfoDtoProjection;
    private final PolicyVersionDtoProjection policyVersionDtoProjection;

    @EventListener
    public void handlePolicyCreated(PolicyEvents.PolicyCreated event) {
        policyInfoDtoProjection.createPolicyInfoDto(event.getNewPolicy());
        policyVersionDtoProjection.createPolicyVersionDto(event.getNewPolicy(),
            event.getNewPolicy().getPolicyVersions().withNumber(1).get());
    }

    @EventListener
    public void handlePolicyTerminated(PolicyEvents.PolicyTerminated event) {
        policyInfoDtoProjection.updatePolicyInfoDto(event.getTerminatedPolicy(), event.getTerminatedVersion());
        policyVersionDtoProjection.createPolicyVersionDto(event.getTerminatedPolicy(), event.getTerminatedVersion());
    }

    @EventListener
    public void handlePolicyAnnexed(PolicyEvents.PolicyAnnexed event) {
        policyInfoDtoProjection.updatePolicyInfoDto(event.getAnnexedPolicy(), event.getAnnexVersion());
        policyVersionDtoProjection.createPolicyVersionDto(event.getAnnexedPolicy(), event.getAnnexVersion());
    }

    @EventListener
    public void handlePolicyAnnexCancelled(PolicyEvents.PolicyAnnexCancelled event) {
        policyInfoDtoProjection.updatePolicyInfoDto(event.getPolicy(), event.getCurrentVersionAfterAnnexCancellation());
        policyVersionDtoProjection.updatePolicyVersionDto(event.getCancelledAnnexVersion());
    }
}

@Component
@Transactional(rollbackFor = Throwable.class)
@RequiredArgsConstructor
public class PolicyInfoDtoProjection {

    private final PolicyInfoDtoRepository policyInfoDtoRepository;

    public void createPolicyInfoDto(Policy policy) {
        PolicyVersion policyVersion = policy.getPolicyVersions().withNumber(1).get();
        PolicyInfoDto policyInfo = buildPolicyInfoDto(policy, policyVersion);
        policyInfoDtoRepository.save(policyInfo);
    }

    public void updatePolicyInfoDto(Policy policy, PolicyVersion currentVersion) {
        PolicyInfoDto policyInfo = buildPolicyInfoDto(policy, currentVersion);
        policyInfoDtoRepository.update(policyInfo);
    }

    private PolicyInfoDto buildPolicyInfoDto(Policy policy, PolicyVersion policyVersion) {
        return new PolicyInfoDto(
            policy.getId(),
            policy.getNumber(),
            policyVersion.getCoverPeriod().getFrom(),
            policyVersion.getCoverPeriod().getTo(),
            policyVersion.getCar().getPlaceNumberWithMake(),
            policyVersion.getPolicyHolder().getFullName(),
            policyVersion.getTotalPremium().getAmount()
        );
    }
}

public interface PolicyInfoDtoRepository extends CrudRepository<PolicyInfoDto, Long> {

    /**
     * update
     *
     * @param policy
     */
    @Modifying
    @Query("UPDATE policy_info_dto " +
        "SET " +
        "cover_from = :policy.coverFrom, " +
        "cover_to = :policy.coverTo, " +
        "vehicle = :policy.vehicle, " +
        "policy_holder = :policy.policyHolder, " +
        "total_premium = :policy.totalPremium " +
        "WHERE " +
        "policy_id = :policy.policyId")
    void update(@Param("policy") PolicyInfoDto policy);

    /**
     * find one
     *
     * @param policyId
     * @return
     */
    @Query("SELECT * FROM policy_info_dto p WHERE p.policy_id = :policyId")
    Optional<PolicyInfoDto> findByPolicyId(@Param("policyId") UUID policyId);

}

再看一下Query:

@Component
@RequiredArgsConstructor
public class GetPolicyVersionDetailsHandler implements QueryHandler<PolicyVersionDto, GetPolicyVersionDetailsQuery> {

    private final PolicyVersionDtoFinder policyVersionDtoFinder;

    @Override
    public PolicyVersionDto handle(GetPolicyVersionDetailsQuery query) {
        return policyVersionDtoFinder.findByPolicyNumberAndVersionNumber(query.getPolicyNumber(), query.getVersionNumber());
    }
}

@Component
@RequiredArgsConstructor
public class PolicyVersionDtoFinder {

    private final PolicyVersionDtoRepository repository;

    public PolicyVersionsListDto findVersionsByPolicyNumber(String policyNumber) {
        return new PolicyVersionsListDto(policyNumber, repository.findVersionsByPolicyNumber(policyNumber));
    }

    public PolicyVersionDto findByPolicyNumberAndVersionNumber(String policyNumber, int versionNumber) {
        PolicyVersionDto dto = repository.findByPolicyNumberAndVersionNumber(policyNumber, versionNumber);
        List<PolicyVersionCoverDto> coversInVersion = repository.getCoversInVersion(dto.getId());
        dto.setCovers(coversInVersion);

        return dto;
    }
}

public interface PolicyVersionDtoRepository extends CrudRepository<PolicyVersionDto, Long> {

    /**
     * update
     *
     * @param versionStatus
     * @param policyVersionId
     */
    @Modifying
    @Query("UPDATE policy_version_dto " +
        "SET " +
        "version_status = :versionStatus " +
        "WHERE " +
        "policy_version_id = :policyVersionId")
    void update(@Param("versionStatus") String versionStatus, @Param("policyVersionId") String policyVersionId);

    /**
     * find one
     *
     * @param policyNumber
     * @param versionNumber
     * @return
     */
    @Query(value = "SELECT " +
        "id, policy_version_id, policy_id, " +
        "policy_number, version_number, " +
        "product_code, " +
        "version_status, policy_status, " +
        "policy_holder, insured, car, " +
        "cover_from, cover_to, version_from, version_to, " +
        "total_premium_amount " +
        "FROM policy_version_dto " +
        "WHERE " +
        "policy_number = :policyNumber " +
        "AND version_number = :versionNumber",
        rowMapperClass = PolicyVersionDto.PolicyVersionDtoRowMapper.class)
    PolicyVersionDto findByPolicyNumberAndVersionNumber(
        @Param("policyNumber") String policyNumber,
        @Param("versionNumber") int versionNumber);

    /**
     * find one
     *
     * @param policyVersionDtoId
     * @return
     */
    @Query("SELECT * " +
        "FROM policy_version_cover_dto " +
        "WHERE " +
        "policy_version_dto = :policyVersionDtoId")
    List<PolicyVersionCoverDto> getCoversInVersion(@Param("policyVersionDtoId") Long policyVersionDtoId);

    /**
     * find one
     *
     * @param policyNumber
     * @return
     */
    @Query(value = "SELECT " +
        "version_number, " +
        "version_from, " +
        "version_to, " +
        "version_status " +
        "FROM policy_version_dto " +
        "WHERE " +
        "policy_number = :policyNumber",
        rowMapperClass = PolicyVersionsListDto.PolicyVersionInfoDtoRowMapper.class)
    List<PolicyVersionsListDto.PolicyVersionInfoDto> findVersionsByPolicyNumber(
        @Param("policyNumber") String policyNumber);
}

@Getter
@AllArgsConstructor
public class PolicyVersionsListDto {
    private String policyNumber;
    private List<PolicyVersionInfoDto> versionsInfo;

    @Getter
    @AllArgsConstructor
    public static class PolicyVersionInfoDto {
        private int number;
        private LocalDate versionFrom;
        private LocalDate versionTo;
        private String versionStatus;
    }

    static class PolicyVersionInfoDtoRowMapper implements RowMapper<PolicyVersionInfoDto> {

        @Override
        public PolicyVersionInfoDto mapRow(ResultSet rs, int i) throws SQLException {
            return new PolicyVersionInfoDto(
                rs.getInt("version_number"),
                rs.getDate("version_from").toLocalDate(),
                rs.getDate("version_to").toLocalDate(),
                rs.getString("version_status")
            );
        }
    }
}

代码分层

最终大体结构如下

image.png
image.png

commands存放CommandHandlers
queries存放QueryHandlers
CommandHandlers触发的Event由eventhandlers包下消费。
domain存放领域对象。

按照DDD分层的话，任何外部端口属于六边形洋葱架构，统一放在infrastructure层适配即可，本例介绍最简单的CQRS实践，就不讨论application、domain、infrastructure、interfaces那种DI分层了。

Event-Sourcing拓展

完整的Event-Sourcing的话，还需要很多细节，回溯需要Event持久化，类似于redis没有重写过的aof文件，可以将Event链路复现，方便分析数据过程，管理版本。

还有数据一致性的问题，需要引入最终一致性和柔性事务，常见的有业务上使用MQ补偿，或者Saga，像Axon Framework等现成的CQRS框架。

如果说接入Event持久化的话，并不复杂，还是Handler那个地方，Transaction注解已经包住了publish前中期的代码，publish event之前落库即可，复杂的是event可视化治理投入。

Saga现在也有现成的框架可以接。

性能方面拓展可以在Command落库以后，binlog同步es、redis、mongodb等，查询端走es，走es这个finder实现也可以随时替换成mongodb等。甚至在封装一层分布式内存缓存，击穿则读es reset。

适合Event Sourcing的场景

• 系统没有大量的CRUD，复杂业务的团队。

• 有DDD经验或者具备DDD素养的团队。

• 关注业务数据产生过程，关注业务流程运维，关注报表等情况的团队。

• 版本管理、版本回退等需求。