【架构实战】CQRS架构模式实战

一、CQRS概述

CQRS（Command Query Responsibility Segregation，命令查询职责分离）是一种架构模式：

核心思想：

• 命令（Command）：修改数据的操作
• 查询（Query）：读取数据的操作
• 两者使用不同的模型和存储

为什么需要CQRS：

• 读写负载不均衡
• 读写数据结构差异大
• 需要独立的读写优化

二、CQRS核心概念

1. 基本模型

┌─────────────────────────────────────────────────────────────┐ │ CQRS架构 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ 命令端 │ │ 查询端 │ │ │ │ (写入优化) │──── 同步 ────▶│ (读取优化) │ │ │ └──────┬───────┘ └──────▲───────┘ │ │ │ │ │ │ ▼ │ │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ 命令数据库 │ │ 查询数据库 │ │ │ │ (事务存储) │ │ (只读副本) │ │ │ └──────────────┘ └──────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘

2. 命令端

// 命令接口publicinterfaceCommandHandler<CextendsCommand>{voidhandle(Ccommand);}// 命令基类publicabstractclassCommand{privatefinalStringcommandId;privatefinalLocalDateTimetimestamp;protectedCommand(){this.commandId=UUID.randomUUID().toString();this.timestamp=LocalDateTime.now();}}// 创建订单命令publicclassCreateOrderCommandextendsCommand{privatefinalStringcustomerId;privatefinalList<OrderItemData>items;publicCreateOrderCommand(StringcustomerId,List<OrderItemData>items){this.customerId=customerId;this.items=items;}}// 命令处理器@ServicepublicclassOrderCommandHandlerimplementsCommandHandler<CreateOrderCommand>{@AutowiredprivateOrderRepositoryorderRepository;@AutowiredprivateEventPublishereventPublisher;@Transactional@Overridepublicvoidhandle(CreateOrderCommandcommand){// 创建订单聚合Orderorder=Order.create(OrderId.generate(),CustomerId.of(command.getCustomerId()));// 添加商品for(OrderItemDataitemData:command.getItems()){Productproduct=productRepository.findById(ProductId.of(itemData.getProductId()));order.addItem(product,itemData.getQuantity());}// 提交订单order.submit();// 保存orderRepository.save(order);// 发布事件eventPublisher.publish(newOrderCreatedEvent(order));}}

3. 查询端

// 查询接口publicinterfaceQueryHandler<QextendsQuery,R>{Rhandle(Qquery);}// 查询基类publicabstractclassQuery{// 查询参数}// 订单查询DTO（专为读取优化）publicclassOrderQueryDTO{privateLongorderId;privateStringorderNo;privateStringcustomerName;// 可能需要JOINprivateStringstatusText;// 状态转换privateBigDecimaltotalAmount;privateList<OrderItemQueryDTO>items;privateStringcreateTimeText;// 格式化时间// 允许非常灵活的查询模型}// 查询处理器@ServicepublicclassOrderQueryHandlerimplementsQueryHandler<OrderQuery,List<OrderQueryDTO>>{@AutowiredprivateOrderReadRepositoryreadRepository;@OverridepublicList<OrderQueryDTO>handle(OrderQueryquery){returnreadRepository.findOrders(query);}}

三、数据同步方案

1. 同步复制

┌────────────┐ 同步写入 ┌────────────┐ │ 命令数据库 │ ──────────────▶│ 查询数据库 │ │ (OLTP) │ 实时同步 │ (OLAP) │ └────────────┘ └────────────┘

// 同步复制实现@ServicepublicclassSynchronousReplicationService{@AutowiredprivateJdbcTemplatecommandJdbcTemplate;@AutowiredprivateJdbcTemplatequeryJdbcTemplate;@TransactionalpublicvoidsaveOrder(Orderorder){// 写入命令数据库Stringsql="INSERT INTO orders (id, order_no, customer_id, status, total_amount) "+"VALUES (?, ?, ?, ?, ?)";commandJdbcTemplate.update(sql,order.getId(),order.getOrderNo(),order.getCustomerId(),order.getStatus().name(),order.getTotalAmount());// 同步写入查询数据库StringquerySql="INSERT INTO v_orders (id, order_no, customer_name, status_text, total_amount) "+"VALUES (?, ?, ?, ?, ?)";queryJdbcTemplate.update(querySql,order.getId(),order.getOrderNo(),order.getCustomerName(),// 查询端需要的字段order.getStatus().getText(),order.getTotalAmount());}}

2. 事件驱动复制

// 事件监听同步@ComponentpublicclassOrderEventSynchronizer{@AutowiredprivateOrderReadRepositoryreadRepository;@KafkaListener(topics="order-events")publicvoidhandleOrderEvent(OrderEventevent){if(eventinstanceofOrderCreatedEvent){OrderCreatedEventcreated=(OrderCreatedEvent)event;// 转换为查询模型OrderQueryModelmodel=toQueryModel(created.getOrder());readRepository.save(model);}if(eventinstanceofOrderUpdatedEvent){OrderUpdatedEventupdated=(OrderUpdatedEvent)event;readRepository.update(toQueryModel(updated.getOrder()));}if(eventinstanceofOrderCancelledEvent){OrderCancelledEventcancelled=(OrderCancelledEvent)event;readRepository.delete(cancelled.getOrderId());}}privateOrderQueryModeltoQueryModel(Orderorder){returnOrderQueryModel.builder().id(order.getId()).orderNo(order.getOrderNo()).customerName(getCustomerName(order.getCustomerId())).statusText(order.getStatus().getText()).totalAmount(order.getTotalAmount()).items(order.getItems().stream().map(this::toItemModel).collect(Collectors.toList())).build();}}

3. 最终一致性

┌────────────┐ 事件 ┌────────────┐ 消费 ┌────────────┐ │ 命令端 │ ──────────▶│ 消息队列 │ ───────────▶│ 查询端 │ │ (聚合根) │ │ (Kafka) │ │ (投影) │ └────────────┘ └────────────┘ └────────────┘ │ ▼ ┌────────────┐ │ 事件存储 │ │ (EventStore)│ └────────────┘

四、读写分离优化

1. 命令端优化

// 命令端：事务优先，保证一致性@ServicepublicclassOrderCommandService{@Transactional(isolation=Isolation.REPEATABLE_READ)publicOrderDTOcreateOrder(CreateOrderCommandcommand){// 严格的业务校验validateBusinessRules(command);// 创建聚合Orderorder=orderAggregateFactory.create(command);// 保存到主库orderRepository.save(order);// 发布领域事件eventPublisher.publish(order.getDomainEvents());returntoDTO(order);}privatevoidvalidateBusinessRules(CreateOrderCommandcommand){// 检查库存for(OrderItemDataitem:command.getItems()){if(!inventoryService.checkStock(item.getProductId(),item.getQuantity())){thrownewInsufficientStockException(item.getProductId());}}// 检查客户信用if(!creditService.checkCredit(command.getCustomerId(),command.getTotalAmount())){thrownewInsufficientCreditException(command.getCustomerId());}}}

2. 查询端优化

// 查询端：性能优先，支持各种读取场景@ServicepublicclassOrderQueryService{@AutowiredprivateJdbcTemplatejdbcTemplate;// 场景1：订单列表（分页）publicPage<OrderListDTO>listOrders(OrderListQueryquery){Stringsql=""" SELECT o.*, c.name as customer_name, (SELECT COUNT(*) FROM order_items WHERE order_id = o.id) as item_count FROM orders o LEFT JOIN customers c ON o.customer_id = c.id WHERE o.status = ? ORDER BY o.create_time DESC LIMIT ? OFFSET ? """;// 直接执行优化的查询returnjdbcTemplate.query(sql,(rs,rowNum)->toOrderListDTO(rs),query.getStatus(),query.getPageSize(),query.getOffset());}// 场景2：订单详情（JOIN多表）publicOrderDetailDTOgetOrderDetail(LongorderId){Stringsql=""" SELECT o.*, c.name as customer_name, c.phone as customer_phone, p.name as payment_name, p.method as payment_method FROM orders o LEFT JOIN customers c ON o.customer_id = c.id LEFT JOIN payments p ON o.id = p.order_id WHERE o.id = ? """;returnjdbcTemplate.queryForObject(sql,this::toOrderDetailDTO,orderId);}// 场景3：统计报表publicOrderStatisticsDTOgetStatistics(OrderStatisticsQueryquery){Stringsql=""" SELECT DATE(create_time) as date, COUNT(*) as order_count, SUM(total_amount) as total_amount, AVG(total_amount) as avg_amount FROM orders WHERE create_time BETWEEN ? AND ? GROUP BY DATE(create_time) """;returnjdbcTemplate.query(sql,(rs,rowNum)->toStatisticsDTO(rs),query.getStartDate(),query.getEndDate()).stream().collect(Collectors.groupingBy(OrderStatisticsDTO::getDate)).values().stream().findFirst().orElse(newOrderStatisticsDTO());}}

五、视图模型设计

1. 查询数据库表设计

-- 命令端：规范化设计CREATETABLEorders(idBIGINTPRIMARYKEY,order_noVARCHAR(32),customer_idBIGINT,statusVARCHAR(20),total_amountDECIMAL(12,2),create_timeTIMESTAMP);CREATETABLEorder_items(idBIGINTPRIMARYKEY,order_idBIGINT,product_idBIGINT,quantityINT,priceDECIMAL(10,2));-- 查询端：反规范化设计，冗余常用字段CREATETABLEv_orders(idBIGINTPRIMARYKEY,order_noVARCHAR(32),-- 冗余的客户信息（避免JOIN）customer_idBIGINT,customer_nameVARCHAR(100),customer_phoneVARCHAR(20),customer_addressVARCHAR(200),-- 状态文本（避免转换）statusVARCHAR(20),status_textVARCHAR(50),status_colorVARCHAR(20),-- 预计算的金额total_amountDECIMAL(12,2),discount_amountDECIMAL(12,2),final_amountDECIMAL(12,2),-- 预格式化的时间create_timeTIMESTAMP,create_time_textVARCHAR(50),create_time_dateDATE,-- 冗余的商品数量（避免子查询）item_countINT,item_namesTEXT,INDEXidx_customer(customer_id),INDEXidx_status(status),INDEXidx_create_time(create_time_date));

2. ES查询模型

// Elasticsearch视图模型@Document(indexName="orders")publicclassOrderIndexModel{@IdprivateStringid;@Field(type=FieldType.Keyword)privateStringorderNo;@Field(type=FieldType.Long)privateLongcustomerId;@Field(type=FieldType.Text,analyzer="ik_max_word")privateStringcustomerName;@Field(type=FieldType.Keyword)privateStringstatus;@Field(type=FieldType.Text)privateStringstatusText;@Field(type=FieldType.Double)privateBigDecimaltotalAmount;@Field(type=FieldType.Nested)privateList<OrderItemIndex>items;@Field(type=FieldType.Date)privateLocalDateTimecreateTime;@Field(type=FieldType.Text)privateStringcreateTimeText;// 支持全文搜索@Field(type=FieldType.Text,analyzer="ik_max_word")privateStringsearchText;// orderNo + customerName + productNames}

六、CQRS实现框架

1. Axon Framework

// Axon Framework实现CQRS@SpringBootApplication@EnableAxonFrameworkpublicclassOrderApplication{}@AggregatepublicclassOrderAggregate{@AggregateIdentifierprivateStringorderId;@CommandHandlerpublicOrderAggregate(CreateOrderCommandcommand){apply(newOrderCreatedEvent(command.getOrderId(),command.getCustomerId()));}@EventSourcingHandlerpublicvoidon(OrderCreatedEventevent){this.orderId=event.getOrderId();}@CommandHandlerpublicvoidhandle(AddItemCommandcommand){apply(newItemAddedEvent(orderId,command.getProductId(),command.getQuantity()));}}@ComponentpublicclassOrderEventHandler{@EventHandlerpublicvoidon(OrderCreatedEventevent){// 更新查询端OrderProjectionprojection=OrderProjection.builder().orderId(event.getOrderId()).status("CREATED").build();orderProjectionRepository.save(projection);}}

2. Spring CQRS示例

// 命令端@Service@RequiredArgsConstructorpublicclassOrderCommandService{privatefinalCommandGatewaycommandGateway;publicStringcreateOrder(CreateOrderCommandcommand){returncommandGateway.send(command);}}// 查询端@Service@RequiredArgsConstructorpublicclassOrderQueryService{privatefinalJdbcTemplatejdbcTemplate;publicList<OrderDTO>listOrders(LongcustomerId){returnjdbcTemplate.query("SELECT * FROM orders WHERE customer_id = ?",(rs,rowNum)->toDTO(rs),customerId);}}

七、CQRS最佳实践

1. 何时使用CQRS

2. 注意事项

1. 避免过度设计 - 小型应用不需要CQRS 2. 处理好最终一致性 - 命令端和查询端可能短暂不一致 3. 选择合适的数据同步方式 - 同步：延迟低，但影响写入性能 - 异步：写入快，但存在延迟 4. 保持命令和查询的独立性 - 不要在命令端直接查询

3. 与其他模式结合

CQRS + DDD： - 命令端使用DDD设计聚合根 - 查询端使用投影构建视图模型 CQRS + Event Sourcing： - 命令端存储事件 - 查询端订阅事件构建投影 CQRS + 微服务： - 每个服务独立使用CQRS - 通过事件总线同步数据

八、总结

CQRS是一种强大的架构模式：

• 命令端：专注业务逻辑，保证一致性
• 查询端：专注读取性能，支持灵活查询
• 数据同步：同步或异步，根据场景选择
• 适用场景：读写负载不均、复杂业务、需要高并发

最佳实践：

1. 优先考虑简单架构
2. 根据实际需求决定是否使用CQRS
3. 处理好一致性问题
4. 做好监控和告警

个人观点，仅供参考