通过Java学习与演示CQRS与事件溯源模式

Posted 2022-04-01 FserSuN

1 简介

本文将介绍CQRS与Event Sourcing设计模式的基本概念。我们首先分别学习两种模式，最后学习二者是如何结合使用。此外还有一些支持这些模式的框架，如Axon。但为了学习两种模式本文不会使用框架演示，而是编写一个简单的应用来理解基本概念。

2 基本概念

使用前先需要理解模式，通常这两个模式在企业应用中是结合在一起使用

2.1 事件溯源(Event Sourcing)

Event Sourcing 模式是将应用程序的状态以一系列事件的进行持久化，最终可以在任何时候查询这些状态并将应用恢复到任意时刻的状态下。

这些状态发生后并进行持久化且不可更改。因此重新还原应用状态就是重放事件的过程。

2.2 CQRS

我们将系统领域对象中的操作分为两种 - 查询与命令，CQRS就是在应用架构中将命令与查询进行分离。

查询返回结果，不会改变系统状态。 命令会改变系统状态，但不一定会返回值。通过将领域模型的命令与查询进行分离实现该模式。

此外，也可以将存储层的读写进行分析，并通过同步机制保持数据一致。

3 一个简单的应用

我们将使用Java进行演示，建立应用及领域模型。 应用会提供领域模型上的CRUD操作，此外还会增加持久化特性。 通过这个简单的应用来演示Event Sourcing与CQRS。此外在例子中还会用到一些领域驱动设计（DDD）的概念。

3.1 应用程序概述

用户信息管理是一个常见的需求，我们以此需求为例进行演示。

如图所示，我们建立了领模型模型，对外暴露了CRUD操作。此外通过内存存储来演示数据持久化。

3.2 应用实现

首先定义领域对象。

public class User 
private String userid;
    private String firstName;
    private String lastName;
    private Set<Contact> contacts;
    private Set<Address> addresses;
    // getters and setters


public class Contact 
    private String type;
    private String detail;
    // getters and setters


public class Address 
    private String city;
    private String state;
    private String postcode;
    // getters and setters

接下来定义存储对象，用来进行持久化演示。

public class UserRepository 
    private Map<String, User> store = new HashMap<>();

    public User getUser(String userId) 
        return store.get(userId);
    

    public void addUser(String userId,User user) 
        store.put(userId,user);
    

接下来定义服务，提供领域对象上的CRUD操作。

public class UserService 
    private UserRepository repository;
    public UserService(UserRepository repository) 
        this.repository = repository;
    

    public void createUser(String userId, String firstName, String lastName) 
        User user = new User(userId, firstName, lastName);
        repository.addUser(userId, user);
    

    public void updateUser(String userId, Set<Contact> contacts, Set<Address> addresses) 
        User user = repository.getUser(userId);
        user.setContacts(contacts);
        user.setAddresses(addresses);
        repository.addUser(userId, user);
    

    public Set<Contact> getContactByType(String userId, String contactType) 
        User user = repository.getUser(userId);
        Set<Contact> contacts = user.getContacts();
        return contacts.stream()
                .filter(c -> c.getType().equals(contactType))
                .collect(Collectors.toSet());
    

    public Set<Address> getAddressByRegion(String userId, String state) 
        User user = repository.getUser(userId);
        Set<Address> addresses = user.getAddresses();
        return addresses.stream()
                .filter(a -> a.getState().equals(state))
                .collect(Collectors.toSet());
    

演示代码很简单，无法满足生产的需求，但我们主要是通过简单的代码演示重要的概念。

3.3 应用存在的问题

在使用Event Sourcing和CQRS处理我们讨论的问题前，我们需要先搞清楚存在的问题，及如何利用这两种模式解决当前问题。

读写操作同时在一个领域对象上，简单问题上当然没有问题。但复杂的问题上可能会引入问题。

应用在持久化对象时，只存储了领域对象的最新的状态。某些场景需要追溯历史时，就无法满足需求。

4. 使用CQRS重构

首先我们将通过CQRS解决第三节提到的第一个问题 - 读写未分离。

上图解释了如何将应用程序的读写进行分离，为了实现读写分离，将引入新的组件聚合器与投影器。

• 聚合（Aggregate）/聚合器（Aggregator）：聚合是DDD中引入的一种模式，将不同的实体绑定在一个聚合根上。

• 投影（Projection）/投影器（Projector）：投影也是一种模式，以不同的形态及结构来表示一个对象。

4.1 实现应用的写功能

We’ll begin by defining the required commands. A command is an intent to mutate the state of the domain model. Whether it succeeds or not depends on the business rules that we configure.

我们将定义一些命令，通过命令调整领域模型的状态。状态调整成功与否以来我们配置的业务规则。

下面是定义的两个命令，命令内维护了我们将要调整的数据。

public class CreateUserCommand 
    private String userId;
    private String firstName;
    private String lastName;


public class UpdateUserCommand 
    private String userId;
    private Set<Address> addresses;
    private Set<Contact> contacts;

接着定义聚合来接收并处理命令，聚合可以接收或拒绝一个命令。聚合使用仓储对象来存获取、持久化对象状态。因此我们创建一个仓储对象。

public class UserWriteRepository 
    private Map<String, User> store = new HashMap<>();


    public User getUser(String userId) 
        return store.get(userId);
    

    public void addUser(String userId,User user) 
        store.put(userId,user);
    

4.2 实现应用的读功能

首先定义读功能用到的领域模型。

public class UserAddress 
    private Map<String, Set<Address>> addressByRegion = new HashMap<>();


public class UserContact 
    private Map<String, Set<Contact>> contactByType = new HashMap<>();

接着我们定义读仓储对象

public class UserReadRepository 
    private Map<String, UserAddress> userAddress = new HashMap<>();
    private Map<String, UserContact> userContact = new HashMap<>();
    // accessors and mutators

接着我们定义需要支持的查询条件对象

public class ContactByTypeQuery 
    private String userId;
    private String contactType;


public class AddressByRegionQuery 
    private String userId;
    private String state;

最后我们通过投影对象（Projection)来处理查询

public class UserProjection 
    private UserReadRepository readRepository;
    public UserProjection(UserReadRepository readRepository) 
        this.readRepository = readRepository;
    

    public Set<Contact> handle(ContactByTypeQuery query) 
        UserContact userContact = readRepository.getUserContact(query.getUserId());
        return userContact.getContactByType()
          .get(query.getContactType());
    

    public Set<Address> handle(AddressByRegionQuery query) 
        UserAddress userAddress = readRepository.getUserAddress(query.getUserId());
        return userAddress.getAddressByRegion()
          .get(query.getState());
    

投影对象（projection）使用读仓储来执行查询。

4.3 读写分离后数据同步

通过前面两个例子看，读和写操作分离执行，使用不同的存储。因此需要对两侧仓储进行数据同步。这里创建投影器（projector）对象完成。投影器对象能将写领域模型的数据同步到读领域模型的逻辑。

public class UserProjector 
    UserReadRepository readRepository = new UserReadRepository();
    public UserProjector(UserReadRepository readRepository) 
        this.readRepository = readRepository;
    

    public void project(User user) 
        UserContact userContact = Optional.ofNullable(
          readRepository.getUserContact(user.getUserid()))
            .orElse(new UserContact());
        Map<String, Set<Contact>> contactByType = new HashMap<>();
        for (Contact contact : user.getContacts()) 
            Set<Contact> contacts = Optional.ofNullable(
              contactByType.get(contact.getType()))
                .orElse(new HashSet<>());
            contacts.add(contact);
            contactByType.put(contact.getType(), contacts);
        
        userContact.setContactByType(contactByType);
        readRepository.addUserContact(user.getUserid(), userContact);

        UserAddress userAddress = Optional.ofNullable(
          readRepository.getUserAddress(user.getUserid()))
            .orElse(new UserAddress());
        Map<String, Set<Address>> addressByRegion = new HashMap<>();
        for (Address address : user.getAddresses()) 
            Set<Address> addresses = Optional.ofNullable(
              addressByRegion.get(address.getState()))
                .orElse(new HashSet<>());
            addresses.add(address);
            addressByRegion.put(address.getState(), addresses);
        
        userAddress.setAddressByRegion(addressByRegion);
        readRepository.addUserAddress(user.getUserid(), userAddress);
    

上面进行同步数据的代码很粗陋，但能够演示CQRS概念。当然在实际的分布式环境中还有很多复杂的问题要处理。

4.4 CQRS的优缺点

优点

• 将读写分离在不同的领域模型上，避免集中在一个领域模型上，使模型变得复杂。
• 将读写存储进行分离，这样可以针对读写QPS及RT(响应时间）的不同，提供更好的性能
• 在分布式架构下与基于事件的编程完美结合

缺点

• 使模型与系统更复杂，仅在复杂的系统中会带来收益。简单系统会增加不必要的复杂度。
• 读写分离的存储会带来一致性问题
• 数据上及代码上会产生冗余