编程事件溯源与CQRS深度实战：从状态存储到事件日志的架构革命

2026-05-09 03:05:57 +0800 CST views 16

事件溯源与CQRS深度实战：从状态存储到事件日志的架构革命

引言：为什么传统CRUD不够用了？

你有没有遇到过这样的问题：用户说"昨天数据明明是对的，今天怎么就变了？"你打开数据库，只看到当前状态，却无法追溯状态是如何一步步变成现在这样的。又或者，业务方要求"给我一个历史报表，统计上个月某一天的库存快照"，你只能在心里默默叹气——因为你的系统只存了"当前状态"，历史数据早就被覆盖了。

这不仅是数据追溯的问题。在高并发场景下，多个用户同时修改同一条记录，乐观锁、悲观锁、数据库行锁……各种锁机制让系统复杂度直线上升。更糟糕的是，当业务越来越复杂，数据库表越来越大，读写性能开始下降，你开始考虑读写分离，却发现主从延迟、数据一致性等问题接踵而来。

事件溯源（Event Sourcing）和CQRS（Command Query Responsibility Segregation）就是为解决这些问题而生的架构模式。它们不是银弹，但在特定场景下，确实能带来传统CRUD无法比拟的优势。本文将深入探讨这两个模式的原理、实现方式、适用场景以及生产级实践。

第一部分：事件溯源的核心思想

1.1 从"状态存储"到"事件日志"

传统CRUD模式的核心是状态存储：数据库中存储的是实体的当前状态，每次更新操作都会覆盖之前的状态。例如，一个订单从"创建"到"已支付"再到"已发货"，数据库中只保留"已发货"这个最终状态。

-- 传统CRUD：覆盖状态
UPDATE orders SET status = 'PAID' WHERE id = 123;
UPDATE orders SET status = 'SHIPPED' WHERE id = 123;
-- 历史状态丢失

事件溯源的核心思想是用事件序列代替当前状态：不存储实体的当前状态，而是存储导致状态变化的所有事件。任何时刻的状态都可以通过重放事件序列来重建。

事件序列：
1. OrderCreated(orderId=123, items=[...], total=99.9)
2. OrderPaid(orderId=123, paymentId=pay_456, paidAt=2026-05-08 10:30:00)
3. OrderShipped(orderId=123, trackingNo=SF123456, shippedAt=2026-05-08 14:00:00)

当前状态：通过重放上述事件计算得出

这种方式的本质转变带来了一系列优势：

完整的审计日志：每个事件都是不可变的事实，天然形成完整的操作历史。

时间旅行能力：可以重建任意历史时刻的系统状态，支持"回放"和"撤销"操作。

业务语义明确：事件名称直接反映业务含义（如OrderCreated、OrderPaid），比单纯的状态字段更有表达力。

并发友好：事件只追加不修改，天然避免了更新冲突。

1.2 事件的本质

在事件溯源中，事件是不可变的事实记录。一个事件通常包含以下信息：

public abstract class DomainEvent {
    private final String eventId;      // 全局唯一ID
    private final String aggregateId;   // 聚合根ID
    private final String eventType;     // 事件类型
    private final Instant timestamp;    // 发生时间
    private final int version;          // 版本号（用于乐观锁）
    private final Map<String, Object> metadata; // 元数据（如操作人、IP等）
}

// 具体事件
public class OrderCreatedEvent extends DomainEvent {
    private final String orderId;
    private final String customerId;
    private final List<OrderItem> items;
    private final BigDecimal totalAmount;
    private final String shippingAddress;
}

事件的关键特性：

不可变性：事件一旦发生，就不能被修改或删除。如果需要"撤销"某个事件，只能发布一个补偿事件。
时序性：事件按时间顺序存储，顺序决定了状态演化的路径。
原子性：每个事件代表一个原子操作，要么发生，要么不发生。
幂等性：相同事件被处理多次，应该产生相同的结果。

1.3 聚合根与事件流

聚合根（Aggregate Root）是领域驱动设计（DDD）中的核心概念，在事件溯源中扮演着事件管理者的角色。每个聚合根管理自己的一组事件，通过应用事件来维护内部状态。

public class OrderAggregate {
    private String orderId;
    private OrderStatus status;
    private List<OrderItem> items;
    private BigDecimal totalAmount;
    private String paymentId;
    private String trackingNo;
    
    // 当前版本号，用于乐观并发控制
    private int version = 0;
    
    // 待发布的新事件列表
    private final List<DomainEvent> pendingEvents = new ArrayList<>();
    
    // 创建订单（命令处理）
    public void create(CreateOrderCommand command) {
        if (status != null) {
            throw new IllegalStateException("Order already exists");
        }
        
        // 验证业务规则
        if (command.getItems().isEmpty()) {
            throw new IllegalArgumentException("Order must have at least one item");
        }
        
        // 发布事件
        applyChange(new OrderCreatedEvent(
            command.getOrderId(),
            command.getCustomerId(),
            command.getItems(),
            calculateTotal(command.getItems()),
            command.getShippingAddress()
        ));
    }
    
    // 支付订单（命令处理）
    public void pay(PayOrderCommand command) {
        if (status != OrderStatus.CREATED) {
            throw new IllegalStateException("Only created order can be paid");
        }
        
        applyChange(new OrderPaidEvent(
            orderId,
            command.getPaymentId(),
            Instant.now()
        ));
    }
    
    // 发货（命令处理）
    public void ship(ShipOrderCommand command) {
        if (status != OrderStatus.PAID) {
            throw new IllegalStateException("Only paid order can be shipped");
        }
        
        applyChange(new OrderShippedEvent(
            orderId,
            command.getTrackingNo(),
            Instant.now()
        ));
    }
    
    // 应用事件（状态变更）
    private void applyChange(DomainEvent event) {
        apply(event);
        pendingEvents.add(event);
    }
    
    // 重放事件（状态重建）
    public void apply(DomainEvent event) {
        if (event instanceof OrderCreatedEvent e) {
            this.orderId = e.getOrderId();
            this.items = e.getItems();
            this.totalAmount = e.getTotalAmount();
            this.status = OrderStatus.CREATED;
        } else if (event instanceof OrderPaidEvent e) {
            this.paymentId = e.getPaymentId();
            this.status = OrderStatus.PAID;
        } else if (event instanceof OrderShippedEvent e) {
            this.trackingNo = e.getTrackingNo();
            this.status = OrderStatus.SHIPPED;
        }
        this.version++;
    }
    
    // 从历史事件重建状态
    public static OrderAggregate fromEvents(List<DomainEvent> events) {
        OrderAggregate aggregate = new OrderAggregate();
        events.forEach(aggregate::apply);
        aggregate.pendingEvents.clear(); // 重建时不产生新事件
        return aggregate;
    }
}

1.4 事件存储（Event Store）

事件存储是专门用于存储事件序列的数据库，与传统数据库有以下不同：

只追加写入：事件只追加，不更新、不删除。
乐观并发控制：使用版本号而非锁机制。
流式组织：事件按聚合根组织成流。

public interface EventStore {
    // 追加事件到指定流
    void append(String streamId, List<DomainEvent> events, int expectedVersion);
    
    // 加载指定流的所有事件
    List<DomainEvent> load(String streamId);
    
    // 从指定版本开始加载事件
    List<DomainEvent> loadFromVersion(String streamId, int version);
    
    // 订阅所有新事件
    void subscribe(Consumer<DomainEvent> subscriber);
}

// 基于数据库的实现
public class JdbcEventStore implements EventStore {
    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;
    
    @Override
    public void append(String streamId, List<DomainEvent> events, int expectedVersion) {
        jdbcTemplate.execute((ConnectionCallback<Void>) connection -> {
            // 检查当前版本（乐观锁）
            Integer currentVersion = getCurrentVersion(connection, streamId);
            if (currentVersion != null && currentVersion != expectedVersion) {
                throw new OptimisticConcurrencyException(
                    "Expected version " + expectedVersion + 
                    " but was " + currentVersion
                );
            }
            
            // 插入事件
            PreparedStatement ps = connection.prepareStatement(
                "INSERT INTO events (event_id, stream_id, event_type, event_data, version, timestamp) " +
                "VALUES (?, ?, ?, ?, ?, ?)"
            );
            
            int version = expectedVersion;
            for (DomainEvent event : events) {
                ps.setString(1, event.getEventId());
                ps.setString(2, streamId);
                ps.setString(3, event.getEventType());
                ps.setString(4, objectMapper.writeValueAsString(event));
                ps.setInt(5, ++version);
                ps.setTimestamp(6, Timestamp.from(event.getTimestamp()));
                ps.addBatch();
            }
            
            ps.executeBatch();
            return null;
        });
    }
    
    @Override
    public List<DomainEvent> load(String streamId) {
        return jdbcTemplate.query(
            "SELECT event_data, event_type FROM events WHERE stream_id = ? ORDER BY version",
            (rs, rowNum) -> deserializeEvent(rs.getString("event_type"), rs.getString("event_data")),
            streamId
        );
    }
}

事件表设计：

CREATE TABLE events (
    event_id VARCHAR(36) PRIMARY KEY,
    stream_id VARCHAR(64) NOT NULL,
    event_type VARCHAR(128) NOT NULL,
    event_data JSON NOT NULL,
    version INT NOT NULL,
    timestamp TIMESTAMP(6) NOT NULL,
    metadata JSON,
    INDEX idx_stream_version (stream_id, version)
);

第二部分：CQRS——读写分离的架构模式

2.1 CQRS的核心思想

CQRS（Command Query Responsibility Segregation）的核心思想是将**命令（写操作）和查询（读操作）**分离，使用不同的模型处理。

传统架构中，读和写共用同一个模型：

┌─────────────────────────────────────┐
│            同一个模型               │
│  ┌─────────────────────────────┐    │
│  │     Order Entity            │    │
│  │  - id, status, items...     │    │
│  │  - create() update() find() │    │
│  └─────────────────────────────┘    │
└─────────────────────────────────────┘
        ↑                   ↓
     写操作               读操作

CQRS架构将读写分离：

┌──────────────────────┐     ┌──────────────────────┐
│     写模型（命令端）   │     │     读模型（查询端）   │
│  ┌────────────────┐  │     │  ┌────────────────┐  │
│  │ OrderAggregate │  │────→│  │ OrderSummary   │  │
│  │ - 状态管理      │  │事件  │  │ - 查询优化      │  │
│  │ - 业务规则      │  │同步  │  │ - 投影视图      │  │
│  │ - 事件发布      │  │     │  │ - 缓存支持      │  │
│  └────────────────┘  │     │  └────────────────┘  │
└──────────────────────┘     └──────────────────────┘
        ↑                            ↓
     命令处理                      查询处理

2.2 为什么要分离读写模型？

性能优化的差异：

写操作需要保证事务一致性、业务规则完整性，通常涉及复杂的验证逻辑。读操作则追求快速响应，可能需要不同的数据结构（如JOIN优化、预计算字段）。

// 写模型：保证业务一致性
public class OrderAggregate {
    public void create(CreateOrderCommand cmd) {
        // 验证库存
        validateInventory(cmd.getItems());
        // 验证价格
        validatePricing(cmd.getItems(), cmd.getCoupons());
        // 计算优惠
        BigDecimal discount = calculateDiscount(cmd);
        // 发布事件
        applyChange(new OrderCreatedEvent(...));
    }
}

// 读模型：优化查询性能
@Table("order_summaries")
public class OrderSummary {
    private String orderId;
    private String customerName;    // 冗余存储，避免JOIN
    private String status;
    private Integer itemCount;      // 预计算
    private BigDecimal totalAmount;
    private Instant createdAt;
    // 不需要复杂的业务规则
}

扩展性的差异：

读操作的频率通常远高于写操作（读写比例可能达到10:1甚至更高）。分离后，可以独立扩展读写端：读端可以添加缓存、使用读副本、部署多个查询服务实例。

技术选型的灵活性：

写端可能选择关系型数据库保证事务，读端可能选择NoSQL数据库优化查询。例如：

写端：PostgreSQL + Event Store
读端：Elasticsearch（全文搜索）、Redis（缓存）、MongoDB（文档查询）

2.3 命令（Command）与命令处理器

命令是表达"意图"的对象，它描述了"用户想要做什么"，而非"系统状态如何变化"。

// 命令定义
public record CreateOrderCommand(
    String orderId,
    String customerId,
    List<OrderItem> items,
    String shippingAddress,
    String couponCode
) implements Command {}

public record PayOrderCommand(
    String orderId,
    String paymentId,
    BigDecimal amount
) implements Command {}

// 命令处理器
@Component
public class OrderCommandHandler {
    private final EventStore eventStore;
    private final OrderRepository orderRepository;
    
    @Transactional
    public void handle(CreateOrderCommand command) {
        // 加载聚合根（如果是新建，则为空）
        OrderAggregate order = orderRepository.load(command.orderId())
            .orElse(new OrderAggregate());
        
        // 执行命令
        order.create(command);
        
        // 持久化事件
        eventStore.append(
            "order-" + command.orderId(),
            order.getPendingEvents(),
            order.getVersion() - order.getPendingEvents().size()
        );
        
        // 发布事件到消息队列（异步处理）
        order.getPendingEvents().forEach(eventBus::publish);
    }
    
    @Transactional
    public void handle(PayOrderCommand command) {
        // 从事件存储重建聚合根
        List<DomainEvent> events = eventStore.load("order-" + command.orderId());
        OrderAggregate order = OrderAggregate.fromEvents(events);
        
        // 执行命令
        order.pay(command);
        
        // 持久化新事件
        eventStore.append(
            "order-" + command.orderId(),
            order.getPendingEvents(),
            order.getVersion() - order.getPendingEvents().size()
        );
    }
}

2.4 投影（Projection）与读模型

投影是将事件流转换为读模型的过程。每当有新事件产生，投影处理器就会更新对应的读模型。

@Component
public class OrderSummaryProjection {
    private final OrderSummaryRepository repository;
    
    @EventHandler
    public void on(OrderCreatedEvent event) {
        OrderSummary summary = new OrderSummary();
        summary.setOrderId(event.getOrderId());
        summary.setCustomerId(event.getCustomerId());
        summary.setStatus("CREATED");
        summary.setItemCount(event.getItems().size());
        summary.setTotalAmount(event.getTotalAmount());
        summary.setCreatedAt(event.getTimestamp());
        
        repository.save(summary);
    }
    
    @EventHandler
    public void on(OrderPaidEvent event) {
        repository.findByOrderId(event.getOrderId())
            .ifPresent(summary -> {
                summary.setStatus("PAID");
                summary.setPaidAt(event.getPaidAt());
                repository.save(summary);
            });
    }
    
    @EventHandler
    public void on(OrderShippedEvent event) {
        repository.findByOrderId(event.getOrderId())
            .ifPresent(summary -> {
                summary.setStatus("SHIPPED");
                summary.setTrackingNo(event.getTrackingNo());
                summary.setShippedAt(event.getShippedAt());
                repository.save(summary);
            });
    }
    
    @EventHandler
    public void on(OrderCancelledEvent event) {
        repository.findByOrderId(event.getOrderId())
            .ifPresent(summary -> {
                summary.setStatus("CANCELLED");
                summary.setCancelledAt(event.getCancelledAt());
                summary.setCancelReason(event.getReason());
                repository.save(summary);
            });
    }
}

2.5 最终一致性

CQRS的一个重要特点是读写模型的最终一致性。命令端发布事件后，读模型可能需要几毫秒到几秒的时间才能完成更新。这意味着：

用户提交命令后立即查询，可能看不到最新状态。
需要在UI层处理这种情况（如显示"处理中"状态，或使用WebSocket推送更新）。

// 处理最终一致性的策略

// 策略1：命令返回后等待读模型更新
@PostMapping("/orders")
public ResponseEntity<OrderResponse> createOrder(@RequestBody CreateOrderRequest request) {
    String orderId = UUID.randomUUID().toString();
    
    // 发送命令
    commandBus.send(new CreateOrderCommand(orderId, ...));
    
    // 等待读模型更新（最多等待5秒）
    Awaitility.await()
        .atMost(5, TimeUnit.SECONDS)
        .until(() -> orderQueryService.findById(orderId).isPresent());
    
    return ResponseEntity.ok(orderQueryService.findById(orderId).orElseThrow());
}

// 策略2：返回命令ID，客户端轮询
@PostMapping("/orders")
public ResponseEntity<CommandResult> createOrder(@RequestBody CreateOrderRequest request) {
    String orderId = UUID.randomUUID().toString();
    String commandId = UUID.randomUUID().toString();
    
    // 异步处理命令
    commandBus.sendAsync(new CreateOrderCommand(orderId, ..., commandId));
    
    return ResponseEntity.accepted()
        .body(new CommandResult(commandId, orderId, "PROCESSING"));
}

// 策略3：WebSocket推送更新
@EventHandler
public void on(OrderCreatedEvent event) {
    // 更新读模型
    updateReadModel(event);
    
    // 推送给客户端
    webSocketService.sendToUser(
        event.getCustomerId(),
        "order:created",
        new OrderNotification(event.getOrderId(), "CREATED")
    );
}

第三部分：事件溯源与CQRS的联合应用

3.1 为什么事件溯源常与CQRS结合？

事件溯源和CQRS是两个独立的模式，但它们经常一起使用，原因如下：

事件溯源产生事件流，CQRS需要事件流：事件溯源的核心输出是事件序列，而CQRS的投影处理器需要订阅事件来更新读模型。两者天然契合。
事件溯源优化写端，CQRS优化读端：事件溯源让写端通过追加事件获得高性能，CQRS让读端通过专用模型获得查询灵活性。
分离关注点，降低复杂度：写端专注于业务规则和状态一致性，读端专注于查询性能和用户体验。

3.2 完整架构设计

┌──────────────────────────────────────────────────────────────────────┐
│                            客户端层                                   │
│  ┌─────────────┐   ┌─────────────┐   ┌─────────────────────────────┐│
│  │  Web App    │   │  Mobile App │   │  第三方集成                  ││
│  └─────────────┘   └─────────────┘   └─────────────────────────────┘│
└───────────────────────────────┬──────────────────────────────────────┘
                                │
┌───────────────────────────────┴──────────────────────────────────────┐
│                            API网关层                                  │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │  REST API: POST /commands, GET /queries                         ││
│  │  WebSocket: 订阅事件更新                                         ││
│  └─────────────────────────────────────────────────────────────────┘│
└───────────────┬─────────────────────────────────────┬────────────────┘
                │ 命令                                 │ 查询
                ↓                                     ↓
┌───────────────────────────────┐   ┌────────────────────────────────────┐
│       命令端（写模型）          │   │         查询端（读模型）            │
│  ┌─────────────────────────┐  │   │  ┌────────────────────────────┐   │
│  │   Command Handler       │  │   │  │   Query Service            │   │
│  │   - 接收命令             │  │   │  │   - 处理查询请求            │   │
│  │   - 加载聚合根           │  │   │  │   - 返回投影视图            │   │
│  │   - 执行业务规则         │  │   │  └────────────────────────────┘   │
│  │   - 发布事件             │  │   │                ↑                  │
│  └─────────────────────────┘  │   │                │                  │
│                ↓               │   │  ┌────────────────────────────┐   │
│  ┌─────────────────────────┐  │   │  │   Read Model Store         │   │
│  │   Event Store           │  │   │  │   - PostgreSQL (查询优化)   │   │
│  │   - 追加事件             │──┼──→│  │   - Redis (缓存)           │   │
│  │   - 乐观并发控制         │  │   │  │   - Elasticsearch (搜索)   │   │
│  │   - 事件流管理           │  │   │  └────────────────────────────┘   │
│  └─────────────────────────┘  │   │                                    │
│                ↓               │   │  ┌────────────────────────────┐   │
│  ┌─────────────────────────┐  │   │  │   Projection Handlers      │   │
│  │   Event Publisher       │  │   │  │   - 订阅事件                │   │
│  │   - 发布到消息队列       │  │   │  │   - 更新读模型              │   │
│  └─────────────────────────┘  │   │  └────────────────────────────┘   │
└───────────────────────────────┘   └────────────────────────────────────┘

3.3 Spring Boot实现示例

项目结构：

src/main/java/com/example/ordersystem/
├── command/                          # 命令端
│   ├── api/
│   │   └── OrderCommandController.java
│   ├── application/
│   │   └── OrderCommandService.java
│   ├── domain/
│   │   ├── OrderAggregate.java
│   │   ├── events/
│   │   │   ├── OrderCreatedEvent.java
│   │   │   ├── OrderPaidEvent.java
│   │   │   └── OrderShippedEvent.java
│   │   └── commands/
│   │       ├── CreateOrderCommand.java
│   │       └── PayOrderCommand.java
│   └── infrastructure/
│       ├── EventStore.java
│       └── JdbcEventStore.java
├── query/                            # 查询端
│   ├── api/
│   │   └── OrderQueryController.java
│   ├── application/
│   │   └── OrderQueryService.java
│   ├── domain/
│   │   └── OrderSummary.java
│   └── infrastructure/
│       ├── OrderSummaryRepository.java
│       └── OrderSummaryProjection.java
└── shared/
    └── EventBus.java

命令端实现：

// OrderCommandController.java
@RestController
@RequestMapping("/api/orders")
public class OrderCommandController {
    private final OrderCommandService commandService;
    
    @PostMapping
    public ResponseEntity<CommandResponse> createOrder(
            @RequestBody @Valid CreateOrderRequest request) {
        String orderId = UUID.randomUUID().toString();
        commandService.createOrder(new CreateOrderCommand(
            orderId,
            request.getCustomerId(),
            request.getItems(),
            request.getShippingAddress(),
            request.getCouponCode()
        ));
        return ResponseEntity.accepted()
            .body(new CommandResponse(orderId, "CREATED"));
    }
    
    @PostMapping("/{orderId}/pay")
    public ResponseEntity<CommandResponse> payOrder(
            @PathVariable String orderId,
            @RequestBody @Valid PayOrderRequest request) {
        commandService.payOrder(new PayOrderCommand(
            orderId,
            request.getPaymentId(),
            request.getAmount()
        ));
        return ResponseEntity.accepted()
            .body(new CommandResponse(orderId, "PAID"));
    }
}

// OrderCommandService.java
@Service
@Transactional
public class OrderCommandService {
    private final EventStore eventStore;
    private final EventBus eventBus;
    
    public void createOrder(CreateOrderCommand command) {
        OrderAggregate order = new OrderAggregate();
        order.create(command);
        
        String streamId = "order-" + command.getOrderId();
        eventStore.append(streamId, order.getPendingEvents(), 0);
        
        order.getPendingEvents().forEach(eventBus::publish);
    }
    
    public void payOrder(PayOrderCommand command) {
        String streamId = "order-" + command.getOrderId();
        List<DomainEvent> events = eventStore.load(streamId);
        
        OrderAggregate order = OrderAggregate.fromEvents(events);
        order.pay(command);
        
        eventStore.append(streamId, order.getPendingEvents(), order.getVersion() - order.getPendingEvents().size());
        order.getPendingEvents().forEach(eventBus::publish);
    }
}

查询端实现：

// OrderQueryController.java
@RestController
@RequestMapping("/api/orders")
public class OrderQueryController {
    private final OrderQueryService queryService;
    
    @GetMapping("/{orderId}")
    public ResponseEntity<OrderSummary> getOrder(@PathVariable String orderId) {
        return queryService.findById(orderId)
            .map(ResponseEntity::ok)
            .orElse(ResponseEntity.notFound().build());
    }
    
    @GetMapping
    public Page<OrderSummary> listOrders(
            @RequestParam(required = false) String customerId,
            @RequestParam(required = false) String status,
            Pageable pageable) {
        return queryService.findByCriteria(customerId, status, pageable);
    }
    
    @GetMapping("/search")
    public List<OrderSummary> searchOrders(@RequestParam String keyword) {
        return queryService.search(keyword);
    }
}

// OrderSummaryProjection.java
@Component
public class OrderSummaryProjection {
    private final OrderSummaryRepository repository;
    private final ElasticsearchTemplate elasticsearchTemplate;
    
    @EventListener
    @Async
    public void on(OrderCreatedEvent event) {
        // 保存到PostgreSQL
        OrderSummary summary = OrderSummary.builder()
            .orderId(event.getOrderId())
            .customerId(event.getCustomerId())
            .status("CREATED")
            .itemCount(event.getItems().size())
            .totalAmount(event.getTotalAmount())
            .createdAt(event.getTimestamp())
            .build();
        
        repository.save(summary);
        
        // 同步到Elasticsearch（用于搜索）
        elasticsearchTemplate.save(buildSearchDocument(event));
    }
    
    @EventListener
    @Async
    public void on(OrderPaidEvent event) {
        repository.findByOrderId(event.getOrderId())
            .ifPresent(summary -> {
                summary.setStatus("PAID");
                summary.setPaidAt(event.getPaidAt());
                summary.setPaymentId(event.getPaymentId());
                repository.save(summary);
            });
    }
    
    @EventListener
    @Async
    public void on(OrderShippedEvent event) {
        repository.findByOrderId(event.getOrderId())
            .ifPresent(summary -> {
                summary.setStatus("SHIPPED");
                summary.setTrackingNo(event.getTrackingNo());
                summary.setShippedAt(event.getShippedAt());
                repository.save(summary);
            });
    }
}

第四部分：高级话题与生产实践

4.1 快照（Snapshot）优化

当聚合根的事件数量非常大时，每次重建状态都需要重放所有事件，性能会下降。快照是一种优化手段：定期保存聚合根的完整状态，重建时只需加载最近的快照，然后重放快照之后的事件。

public class Snapshot {
    private String aggregateId;
    private String aggregateType;
    private int version;
    private String state;  // JSON序列化的状态
    private Instant timestamp;
}

public class OrderAggregate {
    // 从快照+增量事件重建
    public static OrderAggregate fromSnapshotAndEvents(
            Snapshot snapshot, 
            List<DomainEvent> events) {
        // 反序列化快照状态
        OrderAggregate aggregate = objectMapper.readValue(
            snapshot.getState(), 
            OrderAggregate.class
        );
        // 应用增量事件
        events.forEach(aggregate::apply);
        return aggregate;
    }
}

// 快照策略
@Component
public class SnapshotStrategy {
    private static final int SNAPSHOT_THRESHOLD = 100; // 每100个事件生成一次快照
    
    public boolean shouldCreateSnapshot(List<DomainEvent> events, int currentVersion) {
        return events.size() >= SNAPSHOT_THRESHOLD;
    }
    
    public Snapshot createSnapshot(OrderAggregate aggregate) {
        return new Snapshot(
            aggregate.getOrderId(),
            "Order",
            aggregate.getVersion(),
            objectMapper.writeValueAsString(aggregate),
            Instant.now()
        );
    }
}

4.2 Saga模式与分布式事务

在微服务架构中，一个业务操作可能涉及多个服务的事件溯源聚合根。Saga模式通过事件编排来协调跨服务的事务。

// 订单创建Saga
public class CreateOrderSaga {
    private final EventBus eventBus;
    
    public void start(CreateOrderCommand command) {
        // 步骤1：创建订单
        eventBus.publish(new OrderCreationStarted(command));
        
        // 步骤2：预留库存（监听OrderCreated事件后执行）
        // 步骤3：处理支付（监听InventoryReserved事件后执行）
        // 步骤4：确认订单（监听PaymentProcessed事件后执行）
    }
    
    // 补偿逻辑
    @EventHandler
    public void on(PaymentFailedEvent event) {
        // 补偿：释放库存
        eventBus.publish(new ReleaseInventory(event.getOrderId()));
        // 补偿：取消订单
        eventBus.publish(new CancelOrder(event.getOrderId(), "Payment failed"));
    }
}

// Saga状态机
public enum SagaState {
    STARTED,
    ORDER_CREATED,
    INVENTORY_RESERVED,
    PAYMENT_PROCESSING,
    COMPLETED,
    COMPENSATING,
    FAILED
}

4.3 事件版本兼容性

随着业务演进，事件的schema可能发生变化。需要处理新旧版本事件的兼容性问题。

策略1：向上兼容

新版本事件保留所有旧字段，只添加新字段。

// V1事件
public class OrderCreatedEventV1 {
    private String orderId;
    private List<OrderItem> items;
}

// V2事件（兼容V1）
public class OrderCreatedEventV2 {
    private String orderId;
    private List<OrderItem> items;
    private String couponCode;  // 新增字段
    private BigDecimal discount; // 新增字段
}

策略2：事件升级器

在加载事件时，自动将旧版本事件转换为新版本。

@Component
public class EventUpgrader {
    private final ObjectMapper objectMapper;
    
    public DomainEvent upgrade(String eventType, String eventData, int version) {
        return switch (eventType) {
            case "OrderCreatedEvent" -> {
                if (version == 1) {
                    OrderCreatedEventV1 v1 = objectMapper.readValue(eventData, OrderCreatedEventV1.class);
                    yield new OrderCreatedEventV2(
                        v1.getOrderId(),
                        v1.getItems(),
                        null,  // 默认值
                        BigDecimal.ZERO
                    );
                }
                yield objectMapper.readValue(eventData, OrderCreatedEventV2.class);
            }
            default -> objectMapper.readValue(eventData, DomainEvent.class);
        };
    }
}

4.4 事件重放与系统恢复

事件溯源系统的一个重要能力是可以重建整个系统状态。这在以下场景非常有用：

引入新的读模型投影
修复读模型的bug
系统迁移或灾备恢复

@Service
public class EventReplayService {
    private final EventStore eventStore;
    private final List<ProjectionHandler> projections;
    
    public void replayAll() {
        // 清空所有读模型
        projections.forEach(ProjectionHandler::clear);
        
        // 重放所有事件
        eventStore.subscribeAll(event -> {
            projections.forEach(p -> p.handle(event));
        });
    }
    
    public void replayFrom(Instant from) {
        eventStore.subscribeFrom(from, event -> {
            projections.forEach(p -> p.handle(event));
        });
    }
    
    public void rebuildProjection(String projectionName) {
        ProjectionHandler projection = projections.stream()
            .filter(p -> p.getName().equals(projectionName))
            .findFirst()
            .orElseThrow();
        
        projection.clear();
        eventStore.subscribeAll(projection::handle);
    }
}

4.5 性能优化实践

事件存储优化：

-- 分区表（按时间分区）
CREATE TABLE events (
    event_id VARCHAR(36) PRIMARY KEY,
    stream_id VARCHAR(64) NOT NULL,
    event_type VARCHAR(128) NOT NULL,
    event_data JSON NOT NULL,
    version INT NOT NULL,
    timestamp TIMESTAMP(6) NOT NULL,
    metadata JSON
) PARTITION BY RANGE (timestamp);

-- 创建分区
CREATE TABLE events_2026_01 PARTITION OF events
    FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');

CREATE TABLE events_2026_02 PARTITION OF events
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');

-- 索引优化
CREATE INDEX idx_stream_version ON events (stream_id, version);
CREATE INDEX idx_timestamp ON events (timestamp);

读模型缓存策略：

@Service
public class CachedOrderQueryService {
    private final RedisTemplate<String, OrderSummary> redisTemplate;
    private final OrderSummaryRepository repository;
    
    @Cacheable(value = "order-summary", key = "#orderId")
    public Optional<OrderSummary> findById(String orderId) {
        return repository.findByOrderId(orderId);
    }
    
    @CacheEvict(value = "order-summary", key = "#event.orderId")
    @EventListener
    public void on(OrderUpdatedEvent event) {
        // 事件更新时自动清除缓存
    }
    
    // 预热热点数据
    @Scheduled(fixedRate = 300000) // 每5分钟
    public void warmUpCache() {
        List<String> hotOrderIds = findHotOrders();
        hotOrderIds.forEach(this::findById);
    }
}

第五部分：适用场景与决策指南

5.1 适合事件溯源+CQRS的场景

场景	原因
金融交易系统	需要完整的审计日志，任何状态变化都必须可追溯
订单履约系统	状态流转复杂，需要支持回滚、补偿等操作
协作编辑系统	需要支持撤销/重做、版本对比、冲突解决
审计合规系统	法规要求保留完整历史记录
IoT数据处理	高写入频率，读查询模式多样化
事件驱动架构	已有事件基础设施，容易集成

5.2 不适合的场景

场景	原因
简单CRUD应用	引入事件溯源会增加复杂度，收益不大
报表查询为主	读操作远多于写操作，但不需要复杂的状态管理
团队不熟悉DDD	学习曲线陡峭，容易误用
实时性要求极高	最终一致性可能导致用户体验问题
数据量极大的单实体	事件数量过多会影响重建性能

5.3 引入决策流程

开始
  │
  ├─ 是否需要完整审计日志？
  │     ├─ 是 → 考虑事件溯源
  │     └─ 否 → 继续
  │
  ├─ 是否需要时间旅行（重建历史状态）？
  │     ├─ 是 → 事件溯源
  │     └─ 否 → 继续
  │
  ├─ 读写比例是否显著不平衡（>10:1）？
  │     ├─ 是 → 考虑CQRS
  │     └─ 否 → 传统架构可能足够
  │
  ├─ 是否需要独立的读写扩展？
  │     ├─ 是 → CQRS
  │     └─ 否 → 继续
  │
  └─ 业务复杂度是否高（多聚合根协作）？
        ├─ 是 → DDD + 事件溯源
        └─ 否 → 传统CRUD

结语：架构的本质是权衡

事件溯源和CQRS不是银弹，它们解决了一类问题，同时也引入了新的复杂度。选择它们意味着：

获得的好处：

完整的审计日志
时间旅行能力
读写独立扩展
更好的业务表达力
更容易实现事件驱动架构

付出的代价：

学习曲线陡峭
最终一致性的处理
事件版本管理的复杂度
调试和排障难度增加
需要额外的基础设施（事件存储、消息队列）

在决定采用之前，请认真评估你的业务场景、团队能力和运维能力。好的架构不是最先进的架构，而是最适合当前阶段的架构。

参考资料：

Martin Fowler: Event Sourcing
Greg Young: CQRS Documents
Vaughn Vernon: Domain-Driven Design
EventStoreDB Documentation
Axon Framework Reference

编程 事件溯源与CQRS深度实战：从状态存储到事件日志的架构革命