CQRS and Event Sourcing in Java – Java中的CQRS和事件源

最后修改: 2020年 5月 20日

中文/混合/英文(键盘快捷键：t)

1. Introduction

1.绪论

In this tutorial, we’ll explore the basic concepts of Command Query Responsibility Segregation (CQRS) and Event Sourcing design patterns.

在本教程中，我们将探讨命令查询责任隔离（CQRS）和事件源设计模式的基本概念。

While often cited as complementary patterns, we’ll try to understand them separately and finally see how they complement each other. There are several tools and frameworks, such as Axon, to help adopt these patterns, but we’ll create a simple application in Java to understand the basics.

虽然经常被引用为互补模式，但我们将尝试分别理解它们，最后看看它们如何互补。有一些工具和框架，例如Axon，可以帮助采用这些模式，但我们将在Java中创建一个简单的应用程序来了解基本原理。

2. Basic Concepts

2.基本概念

We’ll first understand these patterns theoretically before we attempt to implement them. Also, as they stand as individual patterns quite well, we’ll try to understand without mixing them.

我们将首先从理论上理解这些模式，然后再尝试实现它们。另外，由于它们作为单独的模式站得住脚，我们将尝试在不混合它们的情况下进行理解。

Please note that these patterns are often used together in an enterprise application. In this regard, they also benefit from several other enterprise architecture patterns. We’ll discuss some of them as we go along.

请注意，这些模式经常在企业应用中一起使用。在这方面，它们也受益于其他几种企业架构模式。我们将在接下来的讨论中讨论其中一些。

2.1. Event Sourcing

2.1.事件来源

Event Sourcing gives us a new way of persisting application state as an ordered sequence of events. We can selectively query these events and reconstruct the state of the application at any point in time. Of course, to make this work, we need to reimage every change to the state of the application as events:

事件源为我们提供了一种将应用程序状态持久化为有序事件序列的新方法。我们可以有选择地查询这些事件，并在任何时间点上重建应用程序的状态。当然，要做到这一点，我们需要将应用程序状态的每一个变化都重塑为事件。


These events here are facts that have happened and can not be altered — in other words, they must be immutable. Recreating the application state is just a matter of replaying all the events.

这里的这些事件是已经发生并且不能被改变的事实–换句话说，它们必须是不可改变的。重新创建应用程序的状态只是重新播放所有的事件而已。

Note that this also opens up the possibility to replay events selectively, replay some events in reverse, and much more. As a consequence, we can treat the application state itself as a secondary citizen, with the event log as our primary source of truth.

请注意，这也为选择性地重放事件、反向重放一些事件以及更多的事情提供了可能。因此，我们可以把应用状态本身当作次要的公民，而把事件日志作为我们的主要真相来源。

2.2. CQRS

2.2 CQRS

Put simply, CQRS is about segregating the command and query side of the application architecture. CQRS is based on the Command Query Separation (CQS) principle which was suggested by Bertrand Meyer. CQS suggests that we divide the operations on domain objects into two distinct categories: Queries and Commands:

简单地说，CQRS是关于隔离应用程序架构的命令和查询方面的。CQRS是基于Bertrand Meyer提出的命令查询分离（CQS）原则。CQS建议我们将对领域对象的操作分为两个不同的类别。查询和命令。

Queries return a result and do not change the observable state of a system. Commands change the state of the system but do not necessarily return a value.

查询返回一个结果，不改变系统的可观察状态。命令改变系统的状态，但不一定返回一个值。

We achieve this by cleanly separating the Command and Query sides of the domain model. We can take a step further, splitting the write and read side of the data store as well, of course, by introducing a mechanism to keep them in sync.

我们通过干净地分离领域模型的命令和查询端来实现这一点。我们可以更进一步，将数据存储的写和读两方也分开，当然，要引入一种机制来保持它们的同步。

3. A Simple Application

3.一个简单的应用

We’ll begin by describing a simple application in Java that builds a domain model.

我们将首先描述一个简单的Java应用程序，它建立了一个领域模型。

The application will offer CRUD operations on the domain model and will also feature a persistence for the domain objects. CRUD stands for Create, Read, Update, and Delete, which are basic operations that we can perform on a domain object.

该应用程序将提供对域模型的CRUD操作，并且还将具有域对象的持久性。CRUD代表创建、读取、更新和删除，这是我们可以对领域对象进行的基本操作。

We’ll use the same application to introduce Event Sourcing and CQRS in later sections.

我们将在后面的章节中使用同一个应用程序来介绍事件源和CQRS。

In the process, we’ll leverage some of the concepts from Domain-Driven Design (DDD) in our example.

在这个过程中，我们将在我们的例子中利用一些领域驱动设计（DDD）的概念。

DDD addresses the analysis and design of software that relies on complex domain-specific knowledge. It builds upon the idea that software systems need to be based on a well-developed model of a domain. DDD was first prescribed by Eric Evans as a catalog of patterns. We’ll be using some of these patterns to build our example.

DDD解决了依赖复杂的特定领域知识的软件的分析和设计问题。它建立在软件系统需要建立在一个完善的领域模型的基础上。DDD最早是由Eric Evans规定的，是一个模式的目录。我们将使用这些模式中的一些来构建我们的例子。

3.1. Application Overview

3.1.应用概述

Creating a user profile and managing it is a typical requirement in many applications. We’ll define a simple domain model capturing the user profile along with a persistence:

创建用户配置文件并对其进行管理是许多应用程序中的一个典型需求。我们将定义一个简单的领域模型来捕获用户配置文件以及一个持久化。

As we can see, our domain model is normalized and exposes several CRUD operations. These operations are just for demonstration and can be simple or complex depending upon the requirements. Moreover, the persistence repository here can be in-memory or use a database instead.

正如我们所看到的，我们的领域模型是规范化的，并暴露了几个CRUD操作。这些操作只是为了演示，可以很简单，也可以很复杂，这取决于需求。此外，这里的持久化存储库可以是内存的，也可以使用数据库来代替。

3.2. Application Implementation

3.2.应用实施

First, we’ll have to create Java classes representing our domain model. This is a fairly simple domain model and may not even require the complexities of design patterns like Event Sourcing and CQRS. However, we’ll keep this simple to focus on understanding the basics:

首先，我们要创建代表我们领域模型的Java类。这是一个相当简单的领域模型，甚至可能不需要像事件源和CQRS这样复杂的设计模式。然而，我们将保持简单，以专注于理解基础知识。

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
}

Also, we’ll define a simple in-memory repository for the persistence of our application state. Of course, this does not add any value but suffices for our demonstration later:

另外，我们将定义一个简单的内存存储库，用于持久化我们的应用状态。当然，这并不增加任何价值，但对我们后面的演示来说已经足够了。

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

Now, we’ll define a service to expose typical CRUD operations on our domain model:

现在，我们将定义一个服务，在我们的领域模型上公开典型的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());
    }
}

That’s pretty much what we have to do to set up our simple application. This is far from being production-ready code, but it exposes some of the important points that we’re going to deliberate on later in this tutorial.

这就是我们建立简单应用程序所要做的事情。这还远远不是可用于生产的代码，但它暴露了一些重要的要点，我们将在本教程的后面进行讨论。

3.3. Problems in This Application

3.3.本应用中的问题

Before we proceed any further in our discussion with Event Sourcing and CQRS, it’s worthwhile to discuss the problems with the current solution. After all, we’ll be addressing the same problems by applying these patterns!

在我们进一步讨论事件源和CQRS之前，值得讨论一下当前解决方案的问题。毕竟，我们将通过应用这些模式来解决同样的问题!

Out of many problems that we may notice here, we’ll just like to focus on two of them:

在我们可能注意到的许多问题中，我们只想关注其中的两个问题。

  • Domain Model: The read and write operations are happening over the same domain model. While this is not a problem for a simple domain model like this, it may worsen as the domain model gets complex. We may need to optimize our domain model and the underlying storage for them to suit the individual needs of the read and write operations.
  • Persistence: The persistence we have for our domain objects stores only the latest state of the domain model. While this is sufficient for most situations, it makes some tasks challenging. For instance, if we have to perform a historical audit of how the domain object has changed state, it’s not possible here. We have to supplement our solution with some audit logs to achieve this.

4. Introducing CQRS

4.引入CQRS

We’ll begin addressing the first problem we discussed in the last section by introducing the CQRS pattern in our application. As part of this, we’ll separate the domain model and its persistence to handle write and read operations. Let’s see how CQRS pattern restructures our application:

我们将通过在我们的应用程序中引入CQRS模式，开始解决我们在上一节中讨论的第一个问题。作为其中的一部分，我们将分离领域模型及其持久化，以处理写和读操作。让我们看看CQRS模式是如何重组我们的应用程序的。

The diagram here explains how we intend to cleanly separate our application architecture to write and read sides. However, we have introduced quite a few new components here that we must understand better. Please note that these are not strictly related to CQRS, but CQRS greatly benefits from them:

这里的图表解释了我们打算如何将我们的应用架构干净地分离到写和读两边。然而，我们在这里引入了相当多的新组件，我们必须更好地理解。请注意，这些与CQRS没有严格的关系，但CQRS从它们中大大受益。

  • Aggregate/Aggregator:

Aggregate is a pattern described in Domain-Driven Design (DDD) that logically groups different entities by binding entities to an aggregate root. The aggregate pattern provides transactional consistency between the entities.

聚合是领域驱动设计（DDD）中描述的一种模式，它通过将实体绑定到聚合根上，从逻辑上将不同的实体分组。聚合模式在实体之间提供了事务性的一致性。

CQRS naturally benefits from the aggregate pattern, which groups the write domain model, providing transactional guarantees. Aggregates normally hold a cached state for better performance but can work perfectly without it.

CQRS自然受益于聚合模式，它将写域模型分组，提供事务性保证。聚合体通常持有一个缓存状态，以获得更好的性能，但没有缓存也可以完美地工作。

  • Projection/Projector:

Projection is another important pattern which greatly benefits CQRS. Projection essentially means representing domain objects in different shapes and structures.

投影是另一个重要的模式，对CQRS大有裨益。投影本质上意味着以不同的形状和结构代表领域对象。

These projections of original data are read-only and highly optimized to provide an enhanced read experience. We may again decide to cache projections for better performance, but that’s not a necessity.

这些原始数据的投影是只读的，并且高度优化，以提供一个增强的阅读体验。我们可能会再次决定对投影进行缓存以获得更好的性能，但这并不是必须的。

4.1. Implementing Write Side of Application

4.1.实现应用程序的写入端

Let’s first implement the write side of the application.

让我们首先实现应用程序的写入端。

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.

我们将从定义所需的命令开始。一个命令是一个突变领域模型状态的意图。它是否成功，取决于我们配置的业务规则。

Let’s see our commands:

让我们看看我们的命令。

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;
}

These are pretty simple classes that hold the data we intend to mutate.

这些都是相当简单的类，用来保存我们打算变异的数据。

Next, we define an aggregate that’s responsible for taking commands and handling them. Aggregates may accept or reject a command:

接下来，我们定义一个聚合体，负责接受命令并处理它们。聚合体可以接受或拒绝一个命令。

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;
    }
}

The aggregate uses a repository to retrieve the current state and persist any changes to it. Moreover, it may store the current state locally to avoid the round-trip cost to a repository while processing every command.

聚合体使用存储库来检索当前的状态，并坚持对它的任何改变。此外，它可以在本地存储当前状态，以避免在处理每条命令时到存储库的往返费用。

Finally, we need a repository to hold the state of the domain model. This will typically be a database or other durable store, but here we’ll simply replace them with an in-memory data structure:

最后，我们需要一个存储库来保存领域模型的状态。这通常是一个数据库或其他持久性存储，但在这里我们将简单地用一个内存数据结构代替它们。

public class UserWriteRepository {
    private Map<String, User> store = new HashMap<>();
    // accessors and mutators
}

This concludes the write side of our application.

至此，我们的应用程序的写入部分结束。

4.2. Implementing Read Side of Application

4.2.实现应用程序的读取端

Let’s switch over to the read side of the application now. We’ll begin by defining the read side of the domain model:

现在让我们切换到应用程序的读取端。我们将从定义领域模型的读取端开始。

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

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

If we recall our read operations, it’s not difficult to see that these classes map perfectly well to handle them. That is the beauty of creating a domain model centered around queries we have.

如果我们回忆一下我们的读操作，不难发现这些类完全可以映射处理它们。这就是围绕我们所拥有的查询创建领域模型的魅力所在。

Next, we’ll define the read repository. Again, we’ll just use an in-memory data structure, even though this will be a more durable data store in real applications:

接下来，我们将定义读取存储库。同样，我们将只是使用一个内存数据结构，尽管在实际应用中这将是一个更持久的数据存储。

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

Now, we’ll define the required queries we have to support. A query is an intent to get data — it may not necessarily result in data.

现在，我们将定义我们必须支持的必要查询。查询是一种获取数据的意图–它不一定会产生数据。

Let’s see our queries:

让我们看看我们的查询。

public class ContactByTypeQuery {
    private String userId;
    private String contactType;
}

public class AddressByRegionQuery {
    private String userId;
    private String state;
}

Again, these are simple Java classes holding the data to define a query.

同样，这些都是简单的Java类，持有定义查询的数据。

What we need now is a projection that can handle these queries:

我们现在需要的是一个能够处理这些查询的投影。

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());
    }
}

The projection here uses the read repository we defined earlier to address the queries we have. This pretty much concludes the read side of our application as well.

这里的投影使用我们之前定义的读取库来解决我们的查询。这也基本结束了我们的应用程序的读取方面的工作。

4.3. Synchronizing Read and Write Data

4.3.同步读取和写入数据

One piece of this puzzle is still unsolved: there’s nothing to synchronize our write and read repositories.

这个难题的一个部分仍未解决：没有任何东西可以同步我们的写和读存储库。

This is where we’ll need something known as a projector. A projector has the logic to project the write domain model into the read domain model.

这时我们就需要一个被称为投影仪的东西。一个投影仪具有将写域模型投影到读域模型的逻辑。

There are much more sophisticated ways to handle this, but we’ll keep it relatively simple:

有很多更复杂的方法来处理这个问题，但我们将保持相对简单。

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);
    }
}

This is rather a very crude way of doing this but gives us enough insight into what is needed for CQRS to work. Moreover, it’s not necessary to have the read and write repositories sitting in different physical stores. A distributed system has its own share of problems!

这是一个非常粗略的方法，但让我们对CQRS的工作需要有足够的了解。此外，没有必要把读写库放在不同的物理存储中。分布式系统有它自己的问题!

Please note that it’s not convenient to project the current state of the write domain into different read domain models. The example we have taken here is fairly simple, hence, we do not see the problem.

请注意，将写域的当前状态投射到不同的读域模型中是不方便的。我们在这里所举的例子相当简单，因此，我们没有看到这个问题。

However, as the write and read models get more complex, it’ll get increasingly difficult to project. We can address this through event-based projection instead of state-based projection with Event Sourcing. We’ll see how to achieve this later in the tutorial.

然而，随着写和读的模型越来越复杂，推算起来会越来越困难。我们可以通过基于事件的投射而不是基于状态的投射来解决这个问题，用事件源。我们将在本教程的后面看到如何实现这一点。

4.4. Benefits and Drawbacks of CQRS

4.4.CQRS的好处和坏处

We discussed the CQRS pattern and learned how to introduce it in a typical application. We’ve categorically tried to address the issue related to the rigidity of the domain model in handling both read and write.

我们讨论了CQRS模式，并学习了如何在一个典型的应用程序中引入它。我们已经断然尝试解决与处理读写两方面的领域模型的僵化有关的问题。

Let’s now discuss some of the other benefits that CQRS brings to an application architecture:

现在让我们来讨论一下CQRS给应用程序架构带来的一些其他好处。

  • CQRS provides us a convenient way to select separate domain models appropriate for write and read operations; we don’t have to create a complex domain model supporting both
  • It helps us to select repositories that are individually suited for handling the complexities of the read and write operations, like high throughput for writing and low latency for reading
  • It naturally complements event-based programming models in a distributed architecture by providing a separation of concerns as well as simpler domain models

However, this does not come for free. As is evident from this simple example, CQRS adds considerable complexity to the architecture. It may not be suitable or worth the pain in many scenarios:

然而，这并不是免费的。从这个简单的例子可以看出，CQRS给架构增加了相当的复杂性。在许多情况下，它可能不适合或不值得这样做。

  • Only a complex domain model can benefit from the added complexity of this pattern; a simple domain model can be managed without all this
  • Naturally leads to code duplication to some extent, which is an acceptable evil compared to the gain it leads us to; however, individual judgment is advised
  • Separate repositories lead to problems of consistency, and it’s difficult to keep the write and read repositories in perfect sync always; we often have to settle for eventual consistency

5. Introducing Event Sourcing

5.引入事件采购

Next, we’ll address the second problem we discussed in our simple application. If we recall, it was related to our persistence repository.

接下来，我们将解决我们在简单应用中讨论的第二个问题。如果我们记得，它与我们的持久性存储库有关。

We’ll introduce Event Sourcing to address this problem. Event Sourcing dramatically changes the way we think of the application state storage.

我们将引入事件源来解决这个问题。事件源极大地改变了我们对应用程序状态存储的思考方式。

Let’s see how it changes our repository:

让我们看看它是如何改变我们的资源库的。

Here, we’ve structured our repository to store an ordered list of domain events. Every change to the domain object is considered an event. How coarse- or fine-grained an event should be is a matter of domain design. The important things to consider here are that events have a temporal order and are immutable.

在这里，我们把我们的资源库结构化，以存储一个有序的域事件列表。对领域对象的每一个改变都被认为是一个事件。一个事件应该有多粗或多细是领域设计的问题。这里需要考虑的重要事项是：事件有一个时间顺序，并且是不可改变的。

5.1. Implementing Events and Event Store

5.1.实现事件和事件存储

The fundamental objects in event-driven applications are events, and event sourcing is no different. As we’ve seen earlier, events represent a specific change in the state of the domain model at a specific point of time. So, we’ll begin by defining the base event for our simple application:

事件驱动应用程序的基本对象是事件，而事件源也不例外。正如我们前面所看到的，事件代表了领域模型的状态在某一特定时间点的具体变化。因此，我们将首先为我们的简单应用定义基础事件。

public abstract class Event {
    public final UUID id = UUID.randomUUID();
    public final Date created = new Date();
}

This just ensures that every event we generate in our application gets a unique identification and the timestamp of creation. These are necessary to process them further.

这只是确保我们在应用程序中生成的每个事件都有一个独特的标识和创建的时间戳。这些是进一步处理它们所必需的。

Of course, there can be several other attributes that may interest us, like an attribute to establish the provenance of an event.

当然，还可以有其他几个可能让我们感兴趣的属性，比如建立一个事件的出处的属性。

Next, let’s create some domain-specific events inheriting from this base 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;
}

These are simple POJOs in Java containing the details of the domain event. However, the important thing to note here is the granularity of events.

这些是Java中简单的POJO，包含了领域事件的细节。然而，这里需要注意的是事件的颗粒度。

We could’ve created a single event for user updates, but instead, we decided to create separate events for addition and removal of address and contact. The choice is mapped to what makes it more efficient to work with the domain model.

我们本可以为用户更新创建一个单一的事件，但我们决定为地址和联系人的添加和删除创建单独的事件。这个选择被映射到什么使它更有效地与领域模型一起工作。

Now, naturally, we need a repository to hold our domain events:

现在，我们自然需要一个资源库来保存我们的领域事件。

public class EventStore {
    private Map<String, List<Event>> store = new HashMap<>();
}

This is a simple in-memory data structure to hold our domain events. In reality, there are several solutions specially created to handle event data like Apache Druid. There are many general-purpose distributed data stores capable of handling event sourcing including Kafka and Cassandra.

这是一个简单的内存数据结构，用来保存我们的领域事件。在现实中，有几个专门为处理事件数据而创建的解决方案，如Apache Druid。有许多通用的分布式数据存储能够处理事件源，包括Kafka和Cassandra。

5.2. Generating and Consuming Events

5.2.生成和消耗事件

So, now our service that handled all CRUD operations will change. Now, instead of updating a moving domain state, it will append domain events. It will also use the same domain events to respond to queries.

所以，现在我们处理所有CRUD操作的服务将改变。现在，它将追加领域事件，而不是更新一个移动的领域状态。它还将使用相同的域事件来响应查询。

Let’s see how we can achieve this:

让我们看看如何实现这一目标。

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))
          .forEach(c -> repository.addEvent(
            userId, new UserContactAddedEvent(c.getType(), c.getDetail())));
        user.getAddresses().stream()
          .filter(a -> !addresses.contains(a))
          .forEach(a -> repository.addEvent(
            userId, new UserAddressRemovedEvent(a.getCity(), a.getState(), a.getPostcode())));
        addresses.stream()
          .filter(a -> !user.getAddresses().contains(a))
          .forEach(a -> repository.addEvent(
            userId, new UserAddressAddedEvent(a.getCity(), a.getState(), a.getPostcode())));
    }

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

    public Set<Address> getAddressByRegion(String userId, String state) throws Exception {
        User user = UserUtility.recreateUserState(repository, userId);
        return user.getAddresses().stream()
          .filter(a -> a.getState().equals(state))
          .collect(Collectors.toSet());
    }
}

Please note that we’re generating several events as part of handling the update user operation here. Also, it’s interesting to note how we are generating the current state of the domain model by replaying all the domain events generated so far.

请注意，作为处理更新用户操作的一部分，我们在这里产生了几个事件。另外，值得注意的是，我们是如何通过重放到目前为止产生的所有领域事件来生成领域模型的当前状态的。

Of course, in a real application, this is not a feasible strategy, and we’ll have to maintain a local cache to avoid generating the state every time. There are other strategies like snapshots and roll-up in the event repository that can speed up the process.

当然，在实际应用中，这并不是一个可行的策略，我们必须维护一个本地缓存，以避免每次都生成状态。还有其他的策略，比如事件库中的快照和滚动，可以加快这个过程。

This concludes our effort to introduce event sourcing in our simple application.

至此，我们在简单的应用程序中引入事件源的努力结束了。

5.3. Benefits and Drawbacks of Event Sourcing

5.3.事件采购的好处和坏处

Now we’ve successfully adopted an alternate way of storing domain objects using event sourcing. Event sourcing is a powerful pattern and brings a lot of benefits to an application architecture if used appropriately:

现在我们已经成功地采用了另一种使用事件源的方式来存储领域对象。事件源是一种强大的模式，如果使用得当，会给应用程序架构带来很多好处。

  • Makes write operations much faster as there is no read, update, and write required; write is merely appending an event to a log
  • Removes the object-relational impedance and, hence, the need for complex mapping tools; of course, we still need to recreate the objects back
  • Happens to provide an audit log as a by-product, which is completely reliable; we can debug exactly how the state of a domain model has changed
  • It makes it possible to support temporal queries and achieve time-travel (the domain state at a point in the past)!
  • It’s a natural fit for designing loosely coupled components in a microservices architecture that communicate asynchronously by exchanging messages

However, as always, even event sourcing is not a silver bullet. It does force us to adopt a dramatically different way to store data. This may not prove to be useful in several cases:

然而，像往常一样，即使事件源也不是银弹。它确实迫使我们采用一种截然不同的方式来存储数据。在一些情况下，这可能不会被证明是有用的。

  • There’s a learning curve associated and a shift in mindset required to adopt event sourcing; it’s not intuitive, to begin with
  • It makes it rather difficult to handle typical queries as we need to recreate the state unless we keep the state in the local cache
  • Although it can be applied to any domain model, it’s more appropriate for the event-based model in an event-driven architecture

6. CQRS with Event Sourcing

6.带有事件源的CQRS

Now that we have seen how to individually introduce Event Sourcing and CQRS to our simple application, it’s time to bring them together. It should be fairly intuitive now that these patterns can greatly benefit from each other. However, we’ll make it more explicit in this section.

现在我们已经看到了如何将事件源和CQRS单独引入到我们的简单应用中，现在是时候将它们结合起来。现在应该是相当直观的了，这些模式可以从彼此中大大受益。然而，我们将在本节中更明确地说明这一点。

Let’s first see how the application architecture brings them together:

让我们先看看应用架构是如何将它们结合起来的。

This should not be any surprise by now. We’ve replaced the write side of the repository to be an event store, while the read side of the repository continues to be the same.

这一点现在应该不足为奇。我们已经将版本库的写端替换成了事件存储，而版本库的读端则继续保持不变。

Please note that this is not the only way to use Event Sourcing and CQRS in the application architecture. We can be quite innovative and use these patterns together with other patterns and come up with several architecture options.

请注意，这并不是在应用程序架构中使用事件源和CQRS的唯一方法。我们可以相当创新，将这些模式与其他模式一起使用，并提出若干架构方案。

What’s important here is to ensure that we use them to manage the complexity, not to simply increase the complexities further!

这里重要的是要确保我们用它们来管理复杂性，而不是简单地进一步增加复杂性！”。

6.1. Bringing CQRS and Event Sourcing Together

6.1.将CQRS和事件源结合起来

Having implemented Event Sourcing and CQRS individually, it should not be that difficult to understand how we can bring them together.

在单独实施了Event Sourcing和CQRS之后，要理解我们如何将它们结合在一起应该不是那么困难。

We’ll begin with the application where we introduced CQRS and just make relevant changes to bring event sourcing into the fold. We’ll also leverage the same events and event store that we defined in our application where we introduced event sourcing.

我们将从我们引入CQRS的应用程序开始，只是做一些相关的修改，将事件源引入其中。我们还将利用我们在引入事件源的应用程序中定义的相同事件和事件存储。

There are just a few changes. We’ll begin by changing the aggregate to generate events instead of updating state:

只是有一些变化。我们首先要把聚合改为生成事件而不是更新状态。

public class UserAggregate {
    private EventStore writeRepository;
    public UserAggregate(EventStore repository) {
        this.writeRepository = repository;
    }

    public List<Event> handleCreateUserCommand(CreateUserCommand command) {
        UserCreatedEvent event = new UserCreatedEvent(command.getUserId(), 
          command.getFirstName(), command.getLastName());
        writeRepository.addEvent(command.getUserId(), event);
        return Arrays.asList(event);
    }

    public List<Event> handleUpdateUserCommand(UpdateUserCommand command) {
        User user = UserUtility.recreateUserState(writeRepository, command.getUserId());
        List<Event> events = new ArrayList<>();

        List<Contact> contactsToRemove = user.getContacts().stream()
          .filter(c -> !command.getContacts().contains(c))
          .collect(Collectors.toList());
        for (Contact contact : contactsToRemove) {
            UserContactRemovedEvent contactRemovedEvent = new UserContactRemovedEvent(contact.getType(), 
              contact.getDetail());
            events.add(contactRemovedEvent);
            writeRepository.addEvent(command.getUserId(), contactRemovedEvent);
        }
        List<Contact> contactsToAdd = command.getContacts().stream()
          .filter(c -> !user.getContacts().contains(c))
          .collect(Collectors.toList());
        for (Contact contact : contactsToAdd) {
            UserContactAddedEvent contactAddedEvent = new UserContactAddedEvent(contact.getType(), 
              contact.getDetail());
            events.add(contactAddedEvent);
            writeRepository.addEvent(command.getUserId(), contactAddedEvent);
        }

        // similarly process addressesToRemove
        // similarly process addressesToAdd

        return events;
    }
}

The only other change required is in the projector, which now needs to process events instead of domain object states:

唯一需要改变的是投影仪，它现在需要处理事件而不是域对象状态。

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

    public void project(String userId, List<Event> events) {
        for (Event event : events) {
            if (event instanceof UserAddressAddedEvent)
                apply(userId, (UserAddressAddedEvent) event);
            if (event instanceof UserAddressRemovedEvent)
                apply(userId, (UserAddressRemovedEvent) event);
            if (event instanceof UserContactAddedEvent)
                apply(userId, (UserContactAddedEvent) event);
            if (event instanceof UserContactRemovedEvent)
                apply(userId, (UserContactRemovedEvent) event);
        }
    }

    public void apply(String userId, UserAddressAddedEvent event) {
        Address address = new Address(
          event.getCity(), event.getState(), event.getPostCode());
        UserAddress userAddress = Optional.ofNullable(
          readRepository.getUserAddress(userId))
            .orElse(new UserAddress());
        Set<Address> addresses = Optional.ofNullable(userAddress.getAddressByRegion()
          .get(address.getState()))
          .orElse(new HashSet<>());
        addresses.add(address);
        userAddress.getAddressByRegion()
          .put(address.getState(), addresses);
        readRepository.addUserAddress(userId, userAddress);
    }

    public void apply(String userId, UserAddressRemovedEvent event) {
        Address address = new Address(
          event.getCity(), event.getState(), event.getPostCode());
        UserAddress userAddress = readRepository.getUserAddress(userId);
        if (userAddress != null) {
            Set<Address> addresses = userAddress.getAddressByRegion()
              .get(address.getState());
            if (addresses != null)
                addresses.remove(address);
            readRepository.addUserAddress(userId, userAddress);
        }
    }

    public void apply(String userId, UserContactAddedEvent event) {
        // Similarly handle UserContactAddedEvent event
    }

    public void apply(String userId, UserContactRemovedEvent event) {
        // Similarly handle UserContactRemovedEvent event
    }
}

If we recall the problems we discussed while handling state-based projection, this is a potential solution to that.

如果我们回顾一下我们在处理基于状态的投影时讨论的问题，这就是一个潜在的解决方案。

The event-based projection is rather convenient and easier to implement. All we have to do is process all occurring domain events and apply them to all read domain models. Typically, in an event-based application, the projector would listen to domain events it’s interested in and would not rely on someone calling it directly.

基于事件的投影是相当方便和容易实现的。我们所要做的就是处理所有发生的领域事件，并将其应用于所有读取的领域模型。通常，在一个基于事件的应用程序中，投影仪会监听它感兴趣的领域事件，而不依赖于有人直接调用它。

This is pretty much all we have to do to bring Event Sourcing and CQRS together in our simple application.

这就是我们在简单的应用中把事件源和CQRS结合在一起所要做的几乎所有事情。

7. Conclusion

7.结语

In this tutorial, we discussed the basics of Event Sourcing and CQRS design patterns. We developed a simple application and applied these patterns individually to it.

在本教程中，我们讨论了事件源和CQRS设计模式的基础知识。我们开发了一个简单的应用程序，并将这些模式分别应用于其中。

In the process, we understood the advantages they bring and the drawbacks they present. Finally, we understood why and how to incorporate both of these patterns together in our application.

在这个过程中，我们理解了它们带来的优势和它们的缺点。最后，我们理解了为什么以及如何将这两种模式一起纳入我们的应用。

The simple application we’ve discussed in this tutorial does not even come close to justifying the need for CQRS and Event Sourcing. Our focus was to understand the basic concepts, hence, the example was trivial. But as mentioned before, the benefit of these patterns can only be realized in applications that have a reasonably complex domain model.

我们在本教程中讨论的简单应用甚至不能证明对CQRS和事件源的需求。我们的重点是理解基本概念，因此，这个例子是微不足道的。但如前所述，这些模式的好处只有在具有相当复杂的领域模型的应用中才能实现。

As usual, the source code for this article can be found over on GitHub.

像往常一样，本文的源代码可以在GitHub上找到超过。