CQRS 与事件溯源在 Java 中的应用

最后修改:2024年5月11日

by Kumar Chandrakant

1. 引言

本文将深入探讨命令查询职责分离（CQRS）和事件溯源（Event Sourcing）这两种设计模式的核心概念。

尽管它们常被并列提及、相辅相成，但我们会先分别理解各自的原理，再探讨它们如何协同工作。虽然有 Axon 等框架可以简化实现，但本文将从零开始构建一个简单的 Java 应用，帮助你真正掌握底层机制。

2. 基础概念

在动手编码前，我们先从理论层面理解这两个模式。它们各自独立成立，因此我们先分开讨论。当然，在企业级应用中，它们常常结合使用，并受益于其他架构模式的支持。

2.1. 事件溯源（Event Sourcing）

事件溯源提供了一种全新的方式来持久化应用状态：将状态变化记录为一系列有序的事件。通过选择性地查询这些事件，我们可以重建应用在任意时间点的状态。

关键在于：每一次状态变更都被视为一个不可变的“事实”事件。这意味着事件一旦发生就不能被修改。

Event Sourcing

重建应用状态，本质上就是“重放”所有事件的过程。这带来了巨大的灵活性：你可以选择性重放、反向重放，甚至进行“时间旅行”调试。最终，事件日志成为你系统的“唯一事实来源”（Single Source of Truth），而应用状态反而成了衍生品。

2.2. CQRS（命令查询职责分离）

简单来说，CQRS 就是将应用架构的“写”（命令）和“读”（查询）职责清晰地分离开。它基于 Bertrand Meyer 提出的“命令查询分离”（CQS）原则：

CQRS

  • ✅ 查询（Query）：返回结果，但不改变系统的可观测状态。
  • ✅ 命令（Command）：改变系统状态，通常不返回值（或只返回状态码）。

CQRS 将这一原则贯彻到架构层面，不仅分离了业务逻辑，还可以进一步将数据存储的读写两边也拆开，当然，这需要一个同步机制来保证两边数据的一致性（通常是最终一致性）。

3. 一个简单的应用

我们将从一个典型的 CRUD 应用入手，逐步引入 CQRS 和事件溯源。

该应用提供对领域模型的增删改查操作，并包含数据持久化。我们也会借鉴一些领域驱动设计（DDD）的概念。

3.1. 应用概览

管理用户档案是一个常见需求。我们的领域模型如下：

CRUD Application 3

如图所示，这是一个规范化的模型，暴露了多个 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<>();
}

最后，服务层暴露 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. 当前应用的痛点

这个 CRUD 模型看似正常，但在复杂场景下会踩坑：

  • ⚠️ 领域模型僵化：读写操作共用同一套模型。简单场景没问题，但复杂模型下，读写需求差异巨大，很难同时优化。
  • ⚠️ 持久化信息丢失：仓库只存最新状态。想做历史审计、状态回滚？没门！除非额外加日志，但这就成了补丁。

4. 引入 CQRS

我们先用 CQRS 解决第一个问题：分离读写模型。

CQRS in Application 3

4.1. 实现写端（Command Side）

写端负责处理状态变更。

定义命令（Command）：代表一个状态变更意图。

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

定义聚合根（Aggregate Root）：DDD 中的核心概念，是写模型的入口，负责业务规则校验和状态变更。

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

写仓库（Write Repository）：持久化写模型状态。

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

4.2. 实现读端（Query Side）

读端负责高效响应查询。

定义读模型：针对查询场景优化的数据结构。

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

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

读仓库（Read Repository）：存储优化后的读模型。

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

定义查询（Query）：

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

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

定义投影器（Projector）：处理查询，返回结果。

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

4.3. 同步读写数据

CQRS 的核心挑战：如何保持读写模型的数据同步？

我们引入一个 UserProjector，它监听写模型的变更，并更新读模型：

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

    public void project(User user) {
        // ... 将 User 的 contacts 映射到 UserContact ...
        // ... 将 User 的 addresses 映射到 UserAddress ...
        // 代码略，见原文
    }
}

⚠️ 这种“状态投影”方式简单粗暴，但在复杂模型下会非常难维护。更好的方式是“事件投影”——这正是事件溯源的用武之地。

4.4. CQRS 的利与弊

✅ 优点：

  • 读写模型分离，各自优化，架构更灵活。
  • 可选用最适合的存储（如写用 Kafka，读用 Elasticsearch）。
  • 天然适合事件驱动、微服务架构。

❌ 缺点：

  • 复杂度飙升：对简单应用是过度设计。
  • 代码重复：读写模型难免有重复逻辑。
  • 一致性难题：只能做到最终一致性，强一致性成本极高。

5. 引入事件溯源

现在解决第二个痛点：持久化只存最新状态。

事件溯源将仓库改为存储事件流：

ES in Application 3

5.1. 实现事件与事件仓库

定义基础事件：

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

定义领域事件：

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

// ... 其他事件：UserContactRemovedEvent, UserAddressAddedEvent 等

事件仓库（Event Store）：

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

5.2. 生成与消费事件

服务层不再直接修改状态，而是发布事件：

public class UserService {
    private EventStore 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);
        // 计算差异，发布增删事件
        // ... 略 ...
    }

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

关键点：recreateUserState() 通过重放所有事件来重建当前状态。

5.3. 事件溯源的利与弊

✅ 优点：

  • 写操作极快（追加日志）。
  • 天然审计日志，可追溯任何状态变更。
  • 支持“时间旅行”查询历史状态。
  • 消除 ORM 阻抗。

❌ 缺点：

  • 学习成本高，思维转变大。
  • 查询性能差（需重放事件），通常需配合快照（Snapshot）。
  • 并非所有场景都适用。

6. CQRS 与事件溯源结合

当 CQRS 遇上事件溯源，威力倍增。事件流成为连接读写两边的“桥梁”。

ES CQRS in Application 3

6.1. 两者结合的实现

核心改动：

  1. 聚合根不再直接修改状态，而是发布事件：
public class UserAggregate {
    private EventStore writeRepository;

    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) {
        // ... 重建状态，计算差异，发布多个事件 ...
        return events; // 返回所有生成的事件
    }
}
  1. 投影器监听事件，而非状态：
public class UserProjector {
    public void project(String userId, List<Event> events) {
        for (Event event : events) {
            if (event instanceof UserAddressAddedEvent)
                apply(userId, (UserAddressAddedEvent) event);
            // ... 处理其他事件类型
        }
    }

    public void apply(String userId, UserAddressAddedEvent event) {
        // 直接将事件应用到读模型
        UserAddress userAddress = readRepository.getUserAddress(userId) or new UserAddress();
        userAddress.getAddressByRegion().computeIfAbsent(event.getState(), k -> new HashSet<>()).add(new Address(...));
        readRepository.save(userAddress);
    }
}

✅ 这才是正确的打开方式！事件驱动的投影（Event-based Projection）比状态驱动的投影（State-based Projection）更清晰、更可靠、更容易扩展。

7. 总结

本文从零构建了一个应用，逐步引入了 CQRS 和事件溯源。我们看到了：

  • CQRS 解决了读写模型的分离与优化。
  • 事件溯源解决了状态持久化的可追溯性与性能。
  • 两者结合，通过事件流作为“粘合剂”，构建出高内聚、低耦合、可审计、易扩展的系统。

⚠️ 重要提醒：这些模式是“大杀器”，不是“万金油”。它们会显著增加系统复杂度。只有在领域模型足够复杂、读写需求差异巨大、且需要审计追溯时，才值得引入。对于简单的 CRUD 应用，老老实实写代码才是正道。

文中所有示例代码均可在 GitHub 仓库 找到。

原始标题:CQRS and Event Sourcing in Java | Baeldung

« 上一篇： @PropertySource 与 YAML 文件在 Spring Boot 中的使用
» 下一篇： Java 中 Exchanger 的使用详解