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

一、CQRS概述CQRSCommand Query Responsibility Segregation命令查询职责分离是一种架构模式核心思想命令Command修改数据的操作查询Query读取数据的操作两者使用不同的模型和存储为什么需要CQRS读写负载不均衡读写数据结构差异大需要独立的读写优化二、CQRS核心概念1. 基本模型┌─────────────────────────────────────────────────────────────┐ │ CQRS架构 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ 命令端 │ │ 查询端 │ │ │ │ (写入优化) │──── 同步 ────▶│ (读取优化) │ │ │ └──────┬───────┘ └──────▲───────┘ │ │ │ │ │ │ ▼ │ │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ 命令数据库 │ │ 查询数据库 │ │ │ │ (事务存储) │ │ (只读副本) │ │ │ └──────────────┘ └──────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘2. 命令端// 命令接口publicinterfaceCommandHandlerCextendsCommand{voidhandle(Ccommand);}// 命令基类publicabstractclassCommand{privatefinalStringcommandId;privatefinalLocalDateTimetimestamp;protectedCommand(){this.commandIdUUID.randomUUID().toString();this.timestampLocalDateTime.now();}}// 创建订单命令publicclassCreateOrderCommandextendsCommand{privatefinalStringcustomerId;privatefinalListOrderItemDataitems;publicCreateOrderCommand(StringcustomerId,ListOrderItemDataitems){this.customerIdcustomerId;this.itemsitems;}}// 命令处理器ServicepublicclassOrderCommandHandlerimplementsCommandHandlerCreateOrderCommand{AutowiredprivateOrderRepositoryorderRepository;AutowiredprivateEventPublishereventPublisher;TransactionalOverridepublicvoidhandle(CreateOrderCommandcommand){// 创建订单聚合OrderorderOrder.create(OrderId.generate(),CustomerId.of(command.getCustomerId()));// 添加商品for(OrderItemDataitemData:command.getItems()){ProductproductproductRepository.findById(ProductId.of(itemData.getProductId()));order.addItem(product,itemData.getQuantity());}// 提交订单order.submit();// 保存orderRepository.save(order);// 发布事件eventPublisher.publish(newOrderCreatedEvent(order));}}3. 查询端// 查询接口publicinterfaceQueryHandlerQextendsQuery,R{Rhandle(Qquery);}// 查询基类publicabstractclassQuery{// 查询参数}// 订单查询DTO专为读取优化publicclassOrderQueryDTO{privateLongorderId;privateStringorderNo;privateStringcustomerName;// 可能需要JOINprivateStringstatusText;// 状态转换privateBigDecimaltotalAmount;privateListOrderItemQueryDTOitems;privateStringcreateTimeText;// 格式化时间// 允许非常灵活的查询模型}// 查询处理器ServicepublicclassOrderQueryHandlerimplementsQueryHandlerOrderQuery,ListOrderQueryDTO{AutowiredprivateOrderReadRepositoryreadRepository;OverridepublicListOrderQueryDTOhandle(OrderQueryquery){returnreadRepository.findOrders(query);}}三、数据同步方案1. 同步复制┌────────────┐ 同步写入 ┌────────────┐ │ 命令数据库 │ ──────────────▶│ 查询数据库 │ │ (OLTP) │ 实时同步 │ (OLAP) │ └────────────┘ └────────────┘// 同步复制实现ServicepublicclassSynchronousReplicationService{AutowiredprivateJdbcTemplatecommandJdbcTemplate;AutowiredprivateJdbcTemplatequeryJdbcTemplate;TransactionalpublicvoidsaveOrder(Orderorder){// 写入命令数据库StringsqlINSERT 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());// 同步写入查询数据库StringquerySqlINSERT 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(topicsorder-events)publicvoidhandleOrderEvent(OrderEventevent){if(eventinstanceofOrderCreatedEvent){OrderCreatedEventcreated(OrderCreatedEvent)event;// 转换为查询模型OrderQueryModelmodeltoQueryModel(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(isolationIsolation.REPEATABLE_READ)publicOrderDTOcreateOrder(CreateOrderCommandcommand){// 严格的业务校验validateBusinessRules(command);// 创建聚合OrderorderorderAggregateFactory.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订单列表分页publicPageOrderListDTOlistOrders(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),-- 冗余的客户信息避免JOINcustomer_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(indexNameorders)publicclassOrderIndexModel{IdprivateStringid;Field(typeFieldType.Keyword)privateStringorderNo;Field(typeFieldType.Long)privateLongcustomerId;Field(typeFieldType.Text,analyzerik_max_word)privateStringcustomerName;Field(typeFieldType.Keyword)privateStringstatus;Field(typeFieldType.Text)privateStringstatusText;Field(typeFieldType.Double)privateBigDecimaltotalAmount;Field(typeFieldType.Nested)privateListOrderItemIndexitems;Field(typeFieldType.Date)privateLocalDateTimecreateTime;Field(typeFieldType.Text)privateStringcreateTimeText;// 支持全文搜索Field(typeFieldType.Text,analyzerik_max_word)privateStringsearchText;// orderNo customerName productNames}六、CQRS实现框架1. Axon Framework// Axon Framework实现CQRSSpringBootApplicationEnableAxonFrameworkpublicclassOrderApplication{}AggregatepublicclassOrderAggregate{AggregateIdentifierprivateStringorderId;CommandHandlerpublicOrderAggregate(CreateOrderCommandcommand){apply(newOrderCreatedEvent(command.getOrderId(),command.getCustomerId()));}EventSourcingHandlerpublicvoidon(OrderCreatedEventevent){this.orderIdevent.getOrderId();}CommandHandlerpublicvoidhandle(AddItemCommandcommand){apply(newItemAddedEvent(orderId,command.getProductId(),command.getQuantity()));}}ComponentpublicclassOrderEventHandler{EventHandlerpublicvoidon(OrderCreatedEventevent){// 更新查询端OrderProjectionprojectionOrderProjection.builder().orderId(event.getOrderId()).status(CREATED).build();orderProjectionRepository.save(projection);}}2. Spring CQRS示例// 命令端ServiceRequiredArgsConstructorpublicclassOrderCommandService{privatefinalCommandGatewaycommandGateway;publicStringcreateOrder(CreateOrderCommandcommand){returncommandGateway.send(command);}}// 查询端ServiceRequiredArgsConstructorpublicclassOrderQueryService{privatefinalJdbcTemplatejdbcTemplate;publicListOrderDTOlistOrders(LongcustomerId){returnjdbcTemplate.query(SELECT * FROM orders WHERE customer_id ?,(rs,rowNum)-toDTO(rs),customerId);}}七、CQRS最佳实践1. 何时使用CQRS场景建议简单CRUD不需要CQRS读写负载差异大考虑CQRS复杂业务逻辑考虑CQRS需要高并发读取适合CQRS报表和分析需求非常适合CQRS2. 注意事项1. 避免过度设计 - 小型应用不需要CQRS 2. 处理好最终一致性 - 命令端和查询端可能短暂不一致 3. 选择合适的数据同步方式 - 同步延迟低但影响写入性能 - 异步写入快但存在延迟 4. 保持命令和查询的独立性 - 不要在命令端直接查询3. 与其他模式结合CQRS DDD - 命令端使用DDD设计聚合根 - 查询端使用投影构建视图模型 CQRS Event Sourcing - 命令端存储事件 - 查询端订阅事件构建投影 CQRS 微服务 - 每个服务独立使用CQRS - 通过事件总线同步数据八、总结CQRS是一种强大的架构模式命令端专注业务逻辑保证一致性查询端专注读取性能支持灵活查询数据同步同步或异步根据场景选择适用场景读写负载不均、复杂业务、需要高并发最佳实践优先考虑简单架构根据实际需求决定是否使用CQRS处理好一致性问题做好监控和告警个人观点仅供参考