通过Java学习与演示CQRS与事件溯源模式
Posted FserSuN
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了通过Java学习与演示CQRS与事件溯源模式相关的知识，希望对你有一定的参考价值。
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解决第三节提到的第一个问题 —— 读写未分离。我们通过CQRS模式改造程序，最终完成我们的CRUD功能。
上图解释了如何将应用程序的读写进行分离，为了实现读写分离，将引入新的组件聚合器与投影器。
-
聚合（Aggregate）/聚合器（Aggregator）：聚合是DDD中引入的一种模式，将不同的实体绑定在一个聚合根上。
-
投影（Projection）/投影器（Projector）：投影也是一种模式，以不同的形态及结构来表示一个对象。即一个对象的不同表示形式。也可以参考物理上的概念理解，即光线照射物体，在某个平面上产生的物体的影响。
4.1 实现应用的写功能
我们将定义一些命令，通过命令调整领域模型的状态。状态调整成功与否以来我们配置的业务规则。
下面是定义的两个命令，命令内维护了我们将要调整的数据。
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;
接着定义聚合来接收并处理命令，下面的UserAggregate中负责维护User对象的更新和写入。
public class UserAggregate
private UserWriteRepository writeRepository;
public UserAggregate(UserWriteRepository repository)
this.writeRepository = repository;
public User handleCreateUserCommand(CreateUserCommand command)
User user = new User(command.getUserId(), command.getFirstName(), command.getLastName());
writeRepository.addUser(user.getUserid(), user);
return user;
public User handleUpdateUserCommand(UpdateUserCommand command)
User user = writeRepository.getUser(command.getUserId());
user.setAddresses(command.getAddresses());
user.setContacts(command.getContacts());
writeRepository.addUser(user.getUserid(), user);
return user;
聚合使用仓储对象来存获取、持久化对象状态。因此我们创建一个仓储对象。
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);
通过UserAggregate完成数据写入，写入同时通过UserProjector完成数据同步，将数据同步到读库中。 当产生查询操作时通过UserProjection对外提供查询功能。这样就实现了读写分离及数据同步。当查询条件叫复杂时，可以在读模型上单独建立复杂的查询场景。
当然在实际的分布式环境中还有很多复杂的问题要处理，上面进行同步数据的代码很简单，但能够演示CQRS概念。
4.4 CQRS的优缺点
优点
- 将读写分离在不同的领域模型上，避免集中在一个领域模型上，使模型变得复杂。
- 将读写存储进行分离，这样可以针对读写QPS及RT(响应时间）的不同，提供更好的性能
- 在分布式架构下与基于事件的编程完美结合
缺点
- 使模型与系统更复杂，仅在复杂的系统中会带来收益。简单系统会增加不必要的复杂度。
- 读写分离的存储会带来一致性问题
- 数据上及代码上会产生冗余
5. 事件溯源
接着，我们将解决我们演示程序中遇到的第二个问题 —— 仓储持久化问题。 这种方式改变了我们对应用程序状态存储的思考方式。
我们改造了仓储来按顺序存储领域事件列表。领域时间对象的任何改变都看作是一次事件。注意，事件是序且持久化后不可变的。
5.1 实现事件与事件仓储
事件描述了领域模型在特定的时间点领域状态发生的状态变化，所以我们开始需要为程序定义事件对象。
public abstract class Event
public final UUID id = UUID.randomUUID();
public final Date created = new Date();
首先我们定义了一个抽象事件对象Event，成员ID保证后续生成的事件对象都有唯一的标识和创建时间戳。
此外事件还有其它的一些常见属性，如事件来源。其余属性根据需求去选择即可。
接着我们定义我们的领域事件，这些事件继承与抽象类Event。
public class UserCreatedEvent extends Event
private String userId;
private String firstName;
private String lastName;
public class UserContactAddedEvent extends Event
private String contactType;
private String contactDetails;
public class UserContactRemovedEvent extends Event
private String contactType;
private String contactDetails;
public class UserAddressAddedEvent extends Event
private String city;
private String state;
private String postCode;
public class UserAddressRemovedEvent extends Event
private String city;
private String state;
private String postCode;
我们为用户更新操作只创建了事件对象。我们决定为地址和联系方式的添加与删除在创建独立的事件。如何选择创建哪些事件取决我们的需求，最终保证系统更有效地通过领域模型实现相关功能。
最终我们创建事件存储对象还存储我们的领域事件。
public class EventStore
private Map<String, List<Event>> store = new HashMap<>();
本例中的存储对象很简单。实际中，有多种解决方案来处理领域事件，例如Apache Druid。还有很多通用的分布式数据存储方案能处理事件溯源，包括kafka与cassandra。
5.2 事件的生成与消费
现在，当系统产生CRUD操作后，我们不在改变系统的状态。我们将产生响应的领域事件。
public class UserService
private EventStore repository;
public UserService(EventStore repository)
this.repository = repository;
public void createUser(String userId, String firstName, String lastName)
repository.addEvent(userId, new UserCreatedEvent(userId, firstName, lastName));
public void updateUser(String userId, Set<Contact> contacts, Set<Address> addresses)
User user = UserUtility.recreateUserState(repository, userId);
user.getContacts().stream()
.filter(c -> !contacts.contains(c))
.forEach(c -> repository.addEvent(
userId, new UserContactRemovedEvent(c.getType(), c.getDetail())));
contacts.stream()
.filter(c -> !user.getContacts().contains(c以上是关于通过Java学习与演示CQRS与事件溯源模式的主要内容，如果未能解决你的问题，请参考以下文章