事件源模式与事件存储：构建可审计的分布式系统

老马啸西风2025/8/31大约 8 分钟

事件源模式（Event Sourcing）作为一种创新的架构模式，正在改变我们存储和管理应用程序状态的方式。与传统的状态存储方式不同，事件源模式将系统的状态变化存储为一系列不可变的事件，而不是存储当前状态。这种方法不仅提供了完整的审计轨迹，还支持状态重建、时态查询等强大功能。本文将深入探讨事件源模式的核心概念、实现机制、事件存储的设计，以及在实际应用中的最佳实践。

完整审计轨迹：记录了系统状态变化的完整历史
状态重建：可以通过重放事件重建任何历史状态
时态查询：可以查询系统在任意时间点的状态
业务洞察：通过分析事件序列获得业务洞察

不可变性：一旦创建就不能修改
自包含性：包含处理所需的所有信息
时间戳：记录事件发生的时间
唯一标识：每个事件都有唯一标识符

示例

public final class AccountCreatedEvent {
    private final String accountId;
    private final String owner;
    private final BigDecimal initialBalance;
    private final long timestamp;
    private final String eventId;
    
    // 构造函数和getter方法
}

聚合根（Aggregate Root）

聚合根是事件源模式中的核心概念，代表业务领域中的实体，负责维护事件的一致性。

职责

状态管理：维护聚合的当前状态
业务规则：执行业务规则验证
事件生成：根据命令生成相应的事件
一致性保证：确保聚合内部的一致性

示例

public class BankAccount {
    private String accountId;
    private String owner;
    private BigDecimal balance;
    private List<Event> uncommittedEvents = new ArrayList<>();
    
    // 应用事件更新状态
    public void apply(Event event) {
        if (event instanceof AccountCreatedEvent) {
            apply((AccountCreatedEvent) event);
        } else if (event instanceof MoneyDepositedEvent) {
            apply((MoneyDepositedEvent) event);
        } else if (event instanceof MoneyWithdrawnEvent) {
            apply((MoneyWithdrawnEvent) event);
        }
    }
    
    private void apply(AccountCreatedEvent event) {
        this.accountId = event.getAccountId();
        this.owner = event.getOwner();
        this.balance = event.getInitialBalance();
    }
    
    private void apply(MoneyDepositedEvent event) {
        this.balance = this.balance.add(event.getAmount());
    }
    
    private void apply(MoneyWithdrawnEvent event) {
        this.balance = this.balance.subtract(event.getAmount());
    }
    
    // 业务方法
    public void deposit(BigDecimal amount) {
        // 业务规则验证
        if (amount.compareTo(BigDecimal.ZERO) <= 0) {
            throw new IllegalArgumentException("Amount must be positive");
        }
        
        // 生成事件
        MoneyDepositedEvent event = new MoneyDepositedEvent(
            accountId,
            amount,
            System.currentTimeMillis()
        );
        
        // 应用事件
        apply(event);
        
        // 记录未提交事件
        uncommittedEvents.add(event);
    }
}

事件存储（Event Store）

事件存储是专门用于存储事件的持久化存储系统，通常是一个只追加的存储。

核心特性

只追加：事件一旦写入就不能修改或删除
持久化：确保事件不会因系统故障而丢失
可查询：支持按聚合ID、事件类型、时间等条件查询事件
可回放：支持重放事件序列重建聚合状态

实现方式

public interface EventStore {
    // 保存事件
    void saveEvents(String aggregateId, List<Event> events, int expectedVersion);
    
    // 获取事件
    List<Event> getEventsForAggregate(String aggregateId);
    
    // 获取事件流
    EventStream getEventStream(String aggregateId, int startVersion, int endVersion);
}

快照（Snapshot）

为了提高性能，定期对聚合根的状态进行快照，避免重放过多事件。

作用

性能优化：减少事件重放的数量
启动加速：加快聚合根的初始化速度
存储优化：减少事件存储的压力

实现示例

public class Snapshot {
    private String aggregateId;
    private int version;
    private Object state;
    private long timestamp;
    
    // 构造函数和getter方法
}

public interface SnapshotStore {
    void saveSnapshot(Snapshot snapshot);
    Snapshot getLatestSnapshot(String aggregateId);
}

工作原理

命令处理流程

接收命令：系统接收来自外部的命令
加载聚合：从事件存储中加载聚合的事件并重建状态
执行命令：在聚合上执行命令
生成事件：命令执行成功后生成相应的事件
保存事件：将新生成的事件保存到事件存储中
发布事件：将事件发布给订阅者

状态重建流程

获取快照：从快照存储中获取最新的快照
加载事件：从事件存储中加载快照之后的事件
应用事件：按顺序应用事件更新聚合状态
返回聚合：返回重建后的聚合

public BankAccount loadAccount(String accountId) {
    // 1. 获取最新快照
    Snapshot snapshot = snapshotStore.getLatestSnapshot(accountId);
    BankAccount account;
    
    if (snapshot != null) {
        // 2. 从快照恢复状态
        account = (BankAccount) snapshot.getState();
        // 3. 加载快照之后的事件
        List<Event> events = eventStore.getEventsForAggregate(
            accountId, snapshot.getVersion() + 1);
        // 4. 应用事件
        events.forEach(account::apply);
    } else {
        // 从头开始重建
        account = new BankAccount();
        List<Event> events = eventStore.getEventsForAggregate(accountId);
        events.forEach(account::apply);
    }
    
    return account;
}

CREATE TABLE events (
    event_id UUID PRIMARY KEY,
    aggregate_id VARCHAR(255) NOT NULL,
    aggregate_type VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    event_version INT NOT NULL,
    event_data JSONB NOT NULL,
    occurred_at TIMESTAMP NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_events_aggregate_id ON events(aggregate_id);
CREATE INDEX idx_events_event_type ON events(event_type);
CREATE INDEX idx_events_occurred_at ON events(occurred_at);

快照表结构

CREATE TABLE snapshots (
    aggregate_id VARCHAR(255) PRIMARY KEY,
    aggregate_type VARCHAR(255) NOT NULL,
    snapshot_data JSONB NOT NULL,
    version INT NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);