200｜CQRS：读写模型分离、数据同步、复杂度与适用场景

// CQS 原则示例
public interface OrderService {
    // Command：改变状态，返回 void
    void placeOrder(PlaceOrderCommand command);

    // Query：返回数据，无副作用
    OrderView getOrder(String orderId);
}

┌──────────────┐     ┌──────────────┐
│  Command API  │     │   Query API  │
│  (写模型)      │     │  (读模型)     │
└──────┬───────┘     └──────┬───────┘
       │                    │
       ▼                    ▼
┌─────────────────────────────────────┐
│         共享数据库 (MySQL)           │
└─────────────────────────────────────┘

// 写模型：使用规范化的实体
@Entity
@Table(name = "orders")
public class Order {
    @Id
    private String orderId;
    private String userId;
    private OrderStatus status;
    private BigDecimal totalAmount;
    // 规范化字段，不含冗余
}

// 读模型：使用非规范化的 DTO / 视图
public class OrderDetailView {
    private String orderId;
    private String userName;      // 冗余：来自 user 表
    private String productName;   // 冗余：来自 product 表
    private BigDecimal totalAmount;
    private List<OrderItemView> items;
    // 专门为前端展示优化
}

// 读模型直接使用 SQL 视图或 JOIN 查询
@Repository
public class OrderQueryRepository {
    @Query("SELECT new com.example.view.OrderDetailView(" +
           "o.orderId, u.name, p.name, o.totalAmount) " +
           "FROM Order o JOIN User u ON o.userId = u.id " +
           "JOIN OrderItem oi ON o.orderId = oi.orderId " +
           "JOIN Product p ON oi.productId = p.id " +
           "WHERE o.orderId = :orderId")
    OrderDetailView findOrderDetail(@Param("orderId") String orderId);
}

┌──────────────┐     ┌──────────────┐
│  Command API  │     │   Query API  │
│  (写服务)      │     │  (读服务)     │
└──────┬───────┘     └──────┬───────┘
       │                    │
       ▼                    ▼
┌──────────────┐     ┌──────────────┐
│  Write DB    │────▶│  Read DB     │
│  (MySQL)     │ 同步 │  (ES/Redis)  │
└──────────────┘     └──────────────┘

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Command API  │────▶│ Event Store  │────▶│  Event Bus   │
│  (写服务)      │     │  (事件存储)   │     │  (消息队列)   │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                 │
                    ┌────────────────────────────┘
                    ▼
            ┌──────────────┐
            │   Projector  │
            │  (事件投影器)  │
            └──────┬───────┘
                   │
                   ▼
            ┌──────────────┐     ┌──────────────┐
            │  Read DB     │◀────│   Query API  │
            │  (查询存储)   │     │  (读服务)     │
            └──────────────┘     └──────────────┘

@Transactional
public void placeOrder(PlaceOrderCommand command) {
    // 1. 写入命令模型（写库）
    Order order = orderRepository.save(command.toOrder());

    // 2. 同步更新查询模型（读库）
    OrderView view = OrderView.from(order);
    orderViewRepository.save(view);  // 写入 ES 或 Redis
}

// 命令处理
@Transactional
public void placeOrder(PlaceOrderCommand command) {
    Order order = orderRepository.save(command.toOrder());
    // 发布领域事件
    eventPublisher.publish(new OrderPlacedEvent(order));
}

// 事件处理器：更新查询模型
@EventHandler
public void on(OrderPlacedEvent event) {
    OrderView view = OrderView.from(event.getOrder());
    orderViewRepository.save(view);
}

MySQL (Write DB) ──▶ binlog ──▶ Canal/Debezium ──▶ Kafka ──▶ Consumer ──▶ ES (Read DB)

// 命令对象：不可变，包含完整的操作意图
@Value  // Lombok：不可变对象
public class PlaceOrderCommand {
    @NotBlank
    String userId;

    @NotEmpty
    List<OrderItemCommand> items;

    String couponCode;

    @Valid
    AddressCommand shippingAddress;
}

// 命令处理器：无返回值
public interface CommandHandler<C> {
    void handle(C command);
}

@Component
public class PlaceOrderHandler implements CommandHandler<PlaceOrderCommand> {

    private final OrderRepository orderRepository;
    private final InventoryService inventoryService;
    private final EventPublisher eventPublisher;

    @Override
    @Transactional
    public void handle(PlaceOrderCommand command) {
        // 1. 业务规则校验
        inventoryService.validate(command.getItems());

        // 2. 创建聚合根
        Order order = Order.create(command);

        // 3. 持久化
        orderRepository.save(order);

        // 4. 发布领域事件
        eventPublisher.publish(order.getDomainEvents());
    }
}

// 不同的查询场景，使用不同的视图模型
public class OrderListView {        // 订单列表页
    private String orderId;
    private String productImage;    // 冗余：首张商品图
    private String productName;     // 冗余：首件商品名
    private String orderStatus;     // 冗余：状态文本
    private BigDecimal totalAmount;
    private LocalDateTime createdTime;
}

public class OrderDetailView {      // 订单详情页
    private String orderId;
    private String userName;
    private String userPhone;
    private AddressView shippingAddress;
    private List<OrderItemDetail> items;  // 含商品图片、规格
    private PaymentInfoView paymentInfo;
    private LogisticsView logistics;       // 物流轨迹
    private List<OrderEventView> timeline; // 操作时间线
}

// 命令总线接口
public interface CommandBus {
    <C> void dispatch(C command);
}

// Spring 实现
@Component
public class SpringCommandBus implements CommandBus {

    private final Map<Class<?>, CommandHandler<?>> handlers;

    public SpringCommandBus(List<CommandHandler<?>> handlerList) {
        this.handlers = handlerList.stream()
            .collect(Collectors.toMap(
                h -> resolveCommandType(h.getClass()),
                h -> h
            ));
    }

    @Override
    @SuppressWarnings("unchecked")
    public <C> void dispatch(C command) {
        CommandHandler<C> handler = (CommandHandler<C>)
            handlers.get(command.getClass());
        if (handler == null) {
            throw new NoHandlerException(command.getClass());
        }
        handler.handle(command);
    }
}

// 使用
@RestController
public class OrderController {
    @Autowired
    private CommandBus commandBus;

    @PostMapping("/orders")
    public ResponseEntity<Void> placeOrder(@Valid @RequestBody PlaceOrderCommand command) {
        commandBus.dispatch(command);
        return ResponseEntity.accepted().build();  // 返回 202 Accepted
    }
}

// Event Sourcing 中的聚合根
public class Order extends EventSourcedAggregate {
    private String orderId;
    private OrderStatus status;

    // 命令：不直接修改状态，而是产生事件
    public void place(PlaceOrderCommand cmd) {
        apply(new OrderPlacedEvent(cmd.getUserId(), cmd.getItems()));
    }

    public void cancel(CancelOrderCommand cmd) {
        if (status != OrderStatus.PAID) {
            throw new IllegalStateException("只有已支付订单可取消");
        }
        apply(new OrderCancelledEvent(cmd.getReason()));
    }

    // 事件应用：重建状态
    private void on(OrderPlacedEvent event) {
        this.orderId = event.getOrderId();
        this.status = OrderStatus.PLACED;
    }

    private void on(OrderCancelledEvent event) {
        this.status = OrderStatus.CANCELLED;
    }
}

时间线：
T0: 用户下单 ──▶ 写库已保存
T1: 用户查看订单列表 ──▶ 读库可能还没有（数据不一致窗口）
T2: 事件消费完成 ──▶ 读库已更新

是否需要 CQRS？
    │
    ├── 读写模型是否天然不同？ ── 否 ──▶ 使用传统 CRUD
    │   是
    ├── 是否需要独立扩展读写能力？ ── 否 ──▶ 使用"层次一"（仅模型分离）
    │   是
    ├── 是否需要 Event Sourcing？ ── 是 ──▶ 使用"层次三"（CQRS + ES）
    │   否
    └── 使用"层次二"（模型 + 数据库分离）