基于Event Sourcing和CQRS的微服务架构设计与实战

业务场景描述

在电商系统中，订单的高并发写入与复杂的状态流转（下单、支付、发货、退货等）给传统的CRUD模型带来了挑战：

数据一致性难保证：跨服务事务处理复杂，分布式事务开销大。
写放大问题：频繁更新导致热点写入及性能瓶颈。
审计和追溯需求：需要完整的订单状态变更历史。

针对上述痛点，我们引入Event Sourcing（事件溯源）与CQRS（命令查询职责分离）来构建高可用、可追溯、易扩展的订单微服务。

技术选型过程

Event Sourcing：将状态变化记录为不可变事件，完整保留历史。优点是天然可审计、可回溯，但事件存储和重播需要额外设计。
CQRS：将写模型（Command）与读模型（Query）分离，写入事件后异步同步或投影至专门的查询存储，提高读写性能。缺点是最终一致性带来的复杂性。
消息中间件：选择Kafka作为事件总线，提供高吞吐与持久保证。
存储：事件存储使用关系型数据库（PostgreSQL + EventStore表），查询存储使用Elasticsearch，以满足复杂搜索与报表需求。

综合考虑，系统采用：Spring Boot + Spring Cloud 构建微服务；Event Sourcing + CQRS；Kafka 事件总线；PostgreSQL 事件表；Elasticsearch 查询库。

实现方案详解

项目结构（简化）

order-service/
├── cmd-api/           // Command 侧 REST 接口
├── cmd-impl/          // Command 处理、Event Sourcing 模块
├── query-service/     // Query 侧服务（Spring Data + ES）
├── common/            // 共享模型和工具包
└── config/            // 配置中心、Spring Cloud Config

1. 事件定义

// OrderCreatedEvent.java
public class OrderCreatedEvent {private String orderId;private BigDecimal amount;private LocalDateTime createdTime;// getter/setter
}// OrderStatusChangedEvent.java
public class OrderStatusChangedEvent {private String orderId;private String fromStatus;private String toStatus;private LocalDateTime occurredTime;// getter/setter
}

2. 聚合与Command处理

@Service
public class OrderAggregate {@Aggregateprivate String orderId;private String status;@CommandHandlerpublic OrderAggregate(CreateOrderCommand cmd) {// 校验if (cmd.getAmount().compareTo(BigDecimal.ZERO) <= 0) {throw new IllegalArgumentException("订单金额必须大于0");}// 发布事件apply(new OrderCreatedEvent(cmd.getOrderId(), cmd.getAmount(), LocalDateTime.now()));}@CommandHandlerpublic void handle(ChangeOrderStatusCommand cmd) {apply(new OrderStatusChangedEvent(orderId, this.status, cmd.getNewStatus(), LocalDateTime.now()));}@EventSourcingHandlerpublic void on(OrderCreatedEvent evt) {this.orderId = evt.getOrderId();this.status = "CREATED";}@EventSourcingHandlerpublic void on(OrderStatusChangedEvent evt) {this.status = evt.getToStatus();}
}

3. Kafka配置（application.yml）

spring:kafka:bootstrap-servers: ${KAFKA_SERVERS}producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.springframework.kafka.support.serializer.JsonSerializerconsumer:key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerproperties:spring.json.trusted.packages: "*"

4. 读模型投影

@Service
public class OrderProjection {@EventListenerpublic void handle(OrderCreatedEvent evt) {OrderIndex idx = new OrderIndex(evt.getOrderId(), evt.getAmount(), evt.getCreatedTime(), "CREATED");orderIndexRepository.save(idx);}@EventListenerpublic void handle(OrderStatusChangedEvent evt) {OrderIndex idx = orderIndexRepository.findById(evt.getOrderId()).orElseThrow();idx.setStatus(evt.getToStatus());orderIndexRepository.save(idx);}
}

Elasticsearch实体：

@Document(indexName = "order_index")
public class OrderIndex {@Id private String orderId;private BigDecimal amount;private LocalDateTime createdTime;private String status;// constructor/getter/setter
}

5. API示例

// 创建订单
@PostMapping("/orders")
public ResponseEntity<String> create(@RequestBody CreateOrderDTO dto) {commandGateway.send(new CreateOrderCommand(dto.getOrderId(), dto.getAmount()));return ResponseEntity.accepted().body("创建成功");
}// 查询订单
@GetMapping("/orders/{id}")
public Mono<OrderIndex> get(@PathVariable String id) {return orderIndexRepository.findById(id);
}