第9章跨服务查询 - 分布式查询策略 (Cross-Service Queries: Distributed Query Strategies)

一、导读

1.1 本章学习目标

通过本章的学习，读者将能够：

理解微服务架构中跨服务查询的根本挑战和技术限制

掌握 API 组合模式的工作原理、实现方式和性能优化技巧

深入理解 CQRS 模式在跨服务查询中的应用，包括读模型的設計和同步机制

学会使用命令查询模式处理复杂的跨服务业务场景

掌握数据复制策略，包括全量复制、增量复制和按需复制

了解连接表（Join Table）模式的适用场景和实现方式

学会评估不同查询策略的性能特征，做出合理的技术选型

掌握处理分布式查询中数据一致性、延迟和可靠性的最佳实践

1.2 为什么本章如此重要

在单体应用中，跨表查询是日常操作，SQL JOIN 可以轻松地关联多个表获取完整数据。但在微服务架构中，数据被分散存储在多个服务的独立数据库中，传统的 JOIN 操作不再可行。

跨服务查询是微服务架构中最常见也最具挑战性的问题之一。错误的查询策略会导致性能瓶颈（如 N+1 查询问题）、数据不一致或系统复杂度过高。正确选择合适的查询策略，对于构建高性能、可维护的微服务系统至关重要。

本章将系统介绍各种跨服务查询策略，分析它们的优缺点和适用场景，帮助读者在实际项目中做出明智的技术决策。

1.3 核心问题引导

为什么微服务中不能直接使用数据库 JOIN？这带来了哪些挑战？

什么是 API 组合模式？它有哪些变体和优化技巧？

CQRS 如何解决跨服务查询问题？读模型应该如何设计？

命令查询模式与 API 组合有什么区别？何时应该使用命令查询？

数据复制有哪些策略？如何保证复制数据的一致性？

连接表模式在什么场景下适用？它有什么局限性？

如何处理跨服务查询中的 N+1 问题？

不同查询策略的性能特征是什么？如何进行技术选型？

二、核心概念详解

9.1 跨服务查询的根本挑战

数据库隔离原则

微服务架构的核心原则之一是每个服务拥有自己的数据存储。这意味着：

物理隔离：不同服务的数据可能存储在不同的数据库实例甚至不同类型的数据库中

逻辑隔离：服务之间不能直接访问彼此的数据库

自治性：每个服务可以独立选择最适合的数据存储技术

订单服务                用户服务                商品服务
┌─────────────┐        ┌─────────────┐        ┌─────────────┐
│ orders_db   │        │ users_db    │        │ products_db │
│             │        │             │        │             │
│ orders      │        │ users       │        │ products    │
│ order_items │        │ profiles    │        │ categories  │
└─────────────┘        └─────────────┘        └─────────────┘

挑战分析：

无法 JOIN：跨数据库的 JOIN 操作不可用或性能极差

网络开销：每次跨服务调用都涉及网络通信，延迟显著

部分失败：某个服务不可用时，查询可能返回不完整数据

数据一致性：不同服务的数据可能存在短暂的不一致

分页困难：跨服务的数据难以进行统一的分页和排序

典型场景：

假设需要展示订单列表，包含订单信息和对应的用户名称、商品名称：

需求：显示用户的订单列表，每行包含：
- 订单号
- 用户名称
- 商品名称列表
- 订单总金额
- 订单状态

数据分布：
- 订单号和总金额 → 订单服务
- 用户名称 → 用户服务
- 商品名称 → 商品服务

在单体应用中，这只是一个简单的 SQL JOIN。但在微服务中，需要多次跨服务调用。

9.2 API 组合模式

API 组合（API Composition）是最直观的跨服务查询方案。由一个组合器（Composer）调用多个服务的 API，将结果合并后返回给客户端。

基本工作流程：

客户端 → API 组合器 → 订单服务 API
                    → 用户服务 API
                    → 商品服务 API
         ← 合并结果 ←

实现示例：

java

@Service
@RequiredArgsConstructor
public class OrderQueryComposer {
    
    private final OrderServiceClient orderService;
    private final UserServiceClient userService;
    private final ProductServiceClient productService;
    
    /**
     * 获取订单详情，包含用户和商品信息
     */
    public OrderDetailResponse getOrderDetail(String orderId) {
        // 1. 获取订单基本信息
        OrderResponse order = orderService.getOrder(orderId);
        
        // 2. 获取用户信息
        UserResponse user = userService.getUser(order.getCustomerId());
        
        // 3. 获取商品信息
        List<ProductResponse> products = new ArrayList<>();
        for (OrderItemResponse item : order.getItems()) {
            ProductResponse product = productService.getProduct(item.getProductId());
            products.add(product);
        }
        
        // 4. 组合结果
        return buildOrderDetail(order, user, products);
    }
    
    private OrderDetailResponse buildOrderDetail(
            OrderResponse order, 
            UserResponse user, 
            List<ProductResponse> products) {
        
        Map<String, ProductResponse> productMap = products.stream()
            .collect(Collectors.toMap(ProductResponse::getId, p -> p));
        
        OrderDetailResponse detail = new OrderDetailResponse();
        detail.setOrderId(order.getId());
        detail.setCustomerName(user.getName());
        detail.setStatus(order.getStatus());
        detail.setTotalAmount(order.getTotalAmount());
        
        List<OrderItemDetail> items = order.getItems().stream()
            .map(item -> {
                ProductResponse product = productMap.get(item.getProductId());
                OrderItemDetail itemDetail = new OrderItemDetail();
                itemDetail.setProductName(product != null ? product.getName() : "未知");
                itemDetail.setQuantity(item.getQuantity());
                itemDetail.setPrice(item.getPrice());
                return itemDetail;
            })
            .collect(Collectors.toList());
        
        detail.setItems(items);
        return detail;
    }
}

性能问题：串行调用的延迟累积

上述实现是串行的，总延迟 = 订单服务延迟 + 用户服务延迟 + N × 商品服务延迟。如果每个服务平均延迟 50ms，且有 5 个商品，总延迟可能达到 300ms 以上。

优化策略一：并行调用

java

@Service
@RequiredArgsConstructor
public class ParallelOrderQueryComposer {
    
    private final OrderServiceClient orderService;
    private final UserServiceClient userService;
    private final ProductServiceClient productService;
    
    public OrderDetailResponse getOrderDetail(String orderId) {
        // 1. 获取订单
        OrderResponse order = orderService.getOrder(orderId);
        
        // 2. 并行获取用户和商品信息
        CompletableFuture<UserResponse> userFuture = CompletableFuture
            .supplyAsync(() -> userService.getUser(order.getCustomerId()));
        
        List<String> productIds = order.getItems().stream()
            .map(OrderItemResponse::getProductId)
            .collect(Collectors.toList());
        
        CompletableFuture<List<ProductResponse>> productsFuture = CompletableFuture
            .supplyAsync(() -> productService.getProducts(productIds));
        
        // 3. 等待所有结果
        UserResponse user = userFuture.join();
        List<ProductResponse> products = productsFuture.join();
        
        // 4. 组合结果
        return buildOrderDetail(order, user, products);
    }
}

并行调用可以将总延迟降低到 max(用户服务延迟, 商品服务延迟) + 订单服务延迟。

优化策略二：批量接口

避免 N+1 查询问题，提供批量获取接口：

java

// 商品服务提供批量接口
@RestController
public class ProductController {
    
    @PostMapping("/products/batch")
    public List<ProductResponse> getProducts(@RequestBody List<String> productIds) {
        return productService.findAllById(productIds);
    }
}

// 组合器使用批量接口
List<String> productIds = order.getItems().stream()
    .map(OrderItemResponse::getProductId)
    .collect(Collectors.toList());

List<ProductResponse> products = productService.getProducts(productIds);

优化策略三：缓存

java

@Service
@RequiredArgsConstructor
public class CachedOrderQueryComposer {
    
    @Cacheable(value = "users", key = "#userId")
    public UserResponse getUser(String userId) {
        return userService.getUser(userId);
    }
    
    @Cacheable(value = "products", key = "#productId")
    public ProductResponse getProduct(String productId) {
        return productService.getProduct(productId);
    }
}

API 组合的优缺点：

优点：

实现简单直观
数据实时性强
不需要额外的数据存储

缺点：

性能受最慢服务影响
服务故障会影响整体可用性
复杂查询难以实现（如跨服务过滤、排序）
N+1 问题需要额外优化

适用场景：

查询涉及的服務数量少（2-3 个）
对实时性要求高
查询频率不高

9.3 CQRS 与读模型

CQRS（Command Query Responsibility Segregation）通过为查询专门设计读模型，有效解决跨服务查询问题。

核心思想：

将写操作和读操作分离，写操作保持领域模型的完整性，读操作使用优化的物化视图。

架构示意：

写侧                          读侧
┌──────────┐                 ┌──────────────┐
│ 订单服务  │── 事件 ───────▶│ 订单查询服务  │
│ (写模型)  │                 │ (读模型)      │
└──────────┘                 └──────────────┘
                                    │
                              ┌───────▼────────┐
                              │  order_view    │
                              │  - order_id    │
                              │  - customer_id │
                              │  - customer_name│
                              │  - product_names│
                              │  - total_amount│
                              │  - status      │
                              └────────────────┘

读模型设计：

java

@Entity
@Table(name = "order_query_view")
public class OrderQueryView {
    
    @Id
    private String orderId;
    
    private String customerId;
    private String customerName;
    private String customerEmail;
    
    @Column(columnDefinition = "JSON")
    private String itemsJson;  // [{"productId": "...", "productName": "...", "quantity": 1, "price": 99.9}]
    
    private BigDecimal totalAmount;
    private String status;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
    
    // Getters and Setters
}

@Repository
public interface OrderQueryRepository extends JpaRepository<OrderQueryView, String> {
    
    Page<OrderQueryView> findByCustomerId(String customerId, Pageable pageable);
    
    Page<OrderQueryView> findByStatus(String status, Pageable pageable);
    
    @Query("SELECT o FROM OrderQueryView o WHERE o.customerName LIKE %:keyword% OR o.orderId LIKE %:keyword%")
    Page<OrderQueryView> search(@Param("keyword") String keyword, Pageable pageable);
}

事件同步：

java

@Component
@RequiredArgsConstructor
@Slf4j
public class OrderReadModelSyncHandler {
    
    private final OrderQueryRepository queryRepository;
    private final CustomerServiceClient customerService;
    private final ProductServiceClient productService;
    
    @KafkaListener(topics = "order-events", groupId = "read-model-sync")
    public void handleOrderCreated(OrderCreatedEvent event, Acknowledgment ack) {
        try {
            // 获取补充信息
            CustomerInfo customer = customerService.getCustomer(event.getCustomerId());
            
            List<String> productIds = event.getItems().stream()
                .map(OrderItemData::getProductId)
                .collect(Collectors.toList());
            Map<String, ProductInfo> products = productService.getProducts(productIds);
            
            // 构建读模型
            OrderQueryView view = new OrderQueryView();
            view.setOrderId(event.getOrderId());
            view.setCustomerId(event.getCustomerId());
            view.setCustomerName(customer.getName());
            view.setCustomerEmail(customer.getEmail());
            
            // 构建订单项 JSON
            List<Map<String, Object>> items = event.getItems().stream()
                .map(item -> {
                    Map<String, Object> itemMap = new HashMap<>();
                    itemMap.put("productId", item.getProductId());
                    ProductInfo product = products.get(item.getProductId());
                    itemMap.put("productName", product != null ? product.getName() : "未知商品");
                    itemMap.put("quantity", item.getQuantity());
                    itemMap.put("price", item.getPrice());
                    return itemMap;
                })
                .collect(Collectors.toList());
            
            view.setItemsJson(JsonUtils.toJson(items));
            view.setTotalAmount(event.getTotalAmount());
            view.setStatus("CREATED");
            view.setCreatedAt(event.getCreatedAt());
            view.setUpdatedAt(LocalDateTime.now());
            
            queryRepository.save(view);
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("同步读模型失败", e);
            throw e;  // 触发重试
        }
    }
    
    @KafkaListener(topics = "order-events", groupId = "read-model-sync")
    public void handleOrderStatusChanged(OrderStatusChangedEvent event, Acknowledgment ack) {
        OrderQueryView view = queryRepository.findById(event.getOrderId())
            .orElseThrow(() -> new OrderNotFoundException(event.getOrderId()));
        
        view.setStatus(event.getNewStatus());
        view.setUpdatedAt(LocalDateTime.now());
        
        queryRepository.save(view);
        ack.acknowledge();
    }
}

查询服务：

java

@Service
@RequiredArgsConstructor
public class OrderQueryService {
    
    private final OrderQueryRepository queryRepository;
    
    public OrderDetailResponse getOrderDetail(String orderId) {
        OrderQueryView view = queryRepository.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));
        
        OrderDetailResponse response = new OrderDetailResponse();
        response.setOrderId(view.getOrderId());
        response.setCustomerName(view.getCustomerName());
        response.setCustomerEmail(view.getCustomerEmail());
        response.setStatus(view.getStatus());
        response.setTotalAmount(view.getTotalAmount());
        response.setCreatedAt(view.getCreatedAt());
        
        // 解析 JSON
        List<OrderItemResponse> items = JsonUtils.fromJson(
            view.getItemsJson(), 
            new TypeReference<List<OrderItemResponse>>() {}
        );
        response.setItems(items);
        
        return response;
    }
    
    public Page<OrderSummaryResponse> listCustomerOrders(String customerId, Pageable pageable) {
        return queryRepository.findByCustomerId(customerId, pageable)
            .map(this::toSummaryResponse);
    }
    
    private OrderSummaryResponse toSummaryResponse(OrderQueryView view) {
        OrderSummaryResponse response = new OrderSummaryResponse();
        response.setOrderId(view.getOrderId());
        response.setStatus(view.getStatus());
        response.setTotalAmount(view.getTotalAmount());
        response.setCreatedAt(view.getCreatedAt());
        return response;
    }
}

CQRS 的优缺点：

优点：

查询性能高，单次查询即可获取完整数据
支持复杂的过滤、排序和分页
减轻写侧服务的负载
可以针对不同查询需求设计多个读模型

缺点：

数据存在延迟（最终一致性）
需要额外的存储和维护成本
实现复杂度较高
需要同步机制处理数据更新

适用场景：

查询频率远高于写入频率
需要复杂的查询和报表
对查询性能要求高
可以接受短暂的数据延迟

9.4 命令查询模式

命令查询模式（Command Query）是一种特殊的跨服务查询方式，它将查询请求作为命令发送给目标服务，由目标服务负责组装数据并返回。

与 API 组合的区别：

API 组合：由组合器调用多个服务，在组合器端合并数据
命令查询：由一个服务接收查询命令，该服务内部调用其他服务获取数据

工作流程：

客户端 → 订单服务（查询命令）→ 订单服务内部调用用户服务、商品服务
         ← 完整结果 ←

实现示例：

java

// 订单服务中的查询处理器
@Service
@RequiredArgsConstructor
public class OrderQueryHandler {
    
    private final OrderRepository orderRepository;
    private final UserServiceClient userService;
    private final ProductServiceClient productService;
    
    /**
     * 处理复杂查询命令
     */
    public OrderDetailResponse handleGetOrderDetail(GetOrderDetailQuery query) {
        // 1. 从本地数据库获取订单
        Order order = orderRepository.findById(query.getOrderId())
            .orElseThrow(() -> new OrderNotFoundException(query.getOrderId()));
        
        // 2. 调用用户服务获取用户信息
        UserResponse user = userService.getUser(order.getCustomerId());
        
        // 3. 调用商品服务获取商品信息
        List<String> productIds = order.getItems().stream()
            .map(OrderItem::getProductId)
            .collect(Collectors.toList());
        List<ProductResponse> products = productService.getProducts(productIds);
        
        Map<String, ProductResponse> productMap = products.stream()
            .collect(Collectors.toMap(ProductResponse::getId, p -> p));
        
        // 4. 组装完整结果
        return buildResponse(order, user, productMap);
    }
}

// 查询 DTO
public record GetOrderDetailQuery(
    String orderId,
    boolean includeCustomer,
    boolean includeProducts
) {}

命令查询的优势：

封装性：查询逻辑封装在服务内部，客户端无需知道数据分布

灵活性：服务内部可以灵活优化查询策略（缓存、并行等）

演进性：数据分布变化时，只需修改服务内部实现

注意事项：

避免在服务间形成循环依赖
注意超时设置，防止级联故障
考虑缓存策略，减少重复调用

9.5 数据复制策略

数据复制是将其他服务的数据复制到本地，使跨服务查询变为本地查询。

复制策略分类：

全量复制：复制所有相关数据

增量复制：只复制变化的数据

按需复制：根据需要复制特定数据

全量复制示例：

订单服务定期从用户服务同步所有用户数据：

java

@Component
@RequiredArgsConstructor
@Slf4j
public class UserDataSyncScheduler {
    
    private final UserServiceClient userService;
    private final LocalUserRepository localUserRepository;
    
    @Scheduled(cron = "0 0 2 * * ?")  // 每天凌晨2点执行
    public void syncAllUsers() {
        log.info("开始同步用户数据");
        
        int page = 0;
        int pageSize = 1000;
        
        while (true) {
            Page<UserResponse> userPage = userService.getAllUsers(page, pageSize);
            
            if (userPage.isEmpty()) {
                break;
            }
            
            List<LocalUser> localUsers = userPage.getContent().stream()
                .map(this::convertToLocalUser)
                .collect(Collectors.toList());
            
            localUserRepository.saveAll(localUsers);
            
            log.info("已同步 {} 条用户数据", localUsers.size());
            page++;
        }
        
        log.info("用户数据同步完成");
    }
    
    private LocalUser convertToLocalUser(UserResponse user) {
        LocalUser localUser = new LocalUser();
        localUser.setId(user.getId());
        localUser.setName(user.getName());
        localUser.setEmail(user.getEmail());
        localUser.setSyncedAt(LocalDateTime.now());
        return localUser;
    }
}

增量复制（基于事件）：

java

@Component
@RequiredArgsConstructor
public class UserEventSyncHandler {
    
    private final LocalUserRepository localUserRepository;
    
    @KafkaListener(topics = "user-events", groupId = "order-service-sync")
    public void handleUserCreated(UserCreatedEvent event, Acknowledgment ack) {
        LocalUser user = new LocalUser();
        user.setId(event.getUserId());
        user.setName(event.getName());
        user.setEmail(event.getEmail());
        user.setSyncedAt(LocalDateTime.now());
        
        localUserRepository.save(user);
        ack.acknowledge();
    }
    
    @KafkaListener(topics = "user-events", groupId = "order-service-sync")
    public void handleUserUpdated(UserUpdatedEvent event, Acknowledgment ack) {
        LocalUser user = localUserRepository.findById(event.getUserId())
            .orElse(new LocalUser());
        
        user.setId(event.getUserId());
        user.setName(event.getName());
        user.setEmail(event.getEmail());
        user.setSyncedAt(LocalDateTime.now());
        
        localUserRepository.save(user);
        ack.acknowledge();
    }
    
    @KafkaListener(topics = "user-events", groupId = "order-service-sync")
    public void handleUserDeleted(UserDeletedEvent event, Acknowledgment ack) {
        localUserRepository.deleteById(event.getUserId());
        ack.acknowledge();
    }
}

按需复制：

只在需要时复制特定数据：

java

@Service
@RequiredArgsConstructor
public class OnDemandUserDataSync {
    
    private final UserServiceClient userService;
    private final LocalUserRepository localUserRepository;
    
    public LocalUser getOrSyncUser(String userId) {
        // 先查本地
        Optional<LocalUser> localUser = localUserRepository.findById(userId);
        if (localUser.isPresent()) {
            return localUser.get();
        }
        
        // 本地没有，从远程获取并保存
        UserResponse remoteUser = userService.getUser(userId);
        LocalUser newUser = convertToLocalUser(remoteUser);
        localUserRepository.save(newUser);
        
        return newUser;
    }
}

数据复制的优缺点：

优点：

查询性能高，完全是本地操作
不依赖其他服务的可用性
可以实现复杂的本地查询

缺点：

数据冗余，占用额外存储
数据存在延迟，可能不一致
需要同步机制，增加复杂度
数据结构变化时需要协调

适用场景：

被复制的数据变化频率低
查询频率高
对其他服务的数据只需要子集
可以接受数据延迟

9.6 连接表模式

连接表（Join Table）模式是在一个服务中维护一个包含跨服务数据的宽表，用于高效查询。

与 CQRS 读模型的区别：

CQRS 读模型通常是一个独立的查询服务
连接表是某个服务内部的辅助表

实现示例：

订单服务维护一个包含用户信息的订单宽表：

sql

-- 主订单表
CREATE TABLE orders (
    id VARCHAR(36) PRIMARY KEY,
    customer_id VARCHAR(36) NOT NULL,
    status VARCHAR(20) NOT NULL,
    total_amount DECIMAL(10, 2),
    created_at TIMESTAMP
);

-- 订单宽表（连接表）
CREATE TABLE order_with_customer (
    order_id VARCHAR(36) PRIMARY KEY,
    customer_id VARCHAR(36) NOT NULL,
    customer_name VARCHAR(100),
    customer_email VARCHAR(100),
    status VARCHAR(20) NOT NULL,
    total_amount DECIMAL(10, 2),
    created_at TIMESTAMP,
    INDEX idx_customer (customer_id),
    INDEX idx_status (status)
);

同步逻辑：

java

@Component
@RequiredArgsConstructor
public class OrderWideTableSync {
    
    private final JdbcTemplate jdbcTemplate;
    private final UserServiceClient userService;
    
    @EventListener
    @Transactional
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 获取用户信息
        UserResponse user = userService.getUser(event.getCustomerId());
        
        // 插入宽表
        jdbcTemplate.update("""
            INSERT INTO order_with_customer 
            (order_id, customer_id, customer_name, customer_email, status, total_amount, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            event.getOrderId(),
            event.getCustomerId(),
            user.getName(),
            user.getEmail(),
            "CREATED",
            event.getTotalAmount(),
            event.getCreatedAt()
        );
    }
    
    @EventListener
    @Transactional
    public void handleUserUpdated(UserUpdatedEvent event) {
        // 更新宽表中的用户信息
        jdbcTemplate.update("""
            UPDATE order_with_customer 
            SET customer_name = ?, customer_email = ?
            WHERE customer_id = ?
            """,
            event.getName(),
            event.getEmail(),
            event.getUserId()
        );
    }
}

连接表的优缺点：

优点：

查询效率高
实现相对简单
适合固定模式的查询

缺点：

数据冗余
维护成本高
灵活性差，新增字段需要修改表结构

适用场景：

查询模式相对固定
关联的服务数量有限
数据变化频率不高

9.7 查询策略对比与选型

策略	实时性	性能	复杂度	一致性	适用场景
API 组合	高	中	低	强	简单查询，服务少
CQRS	中	高	高	最终	复杂查询，高频读取
命令查询	高	中	中	强	封装查询逻辑
数据复制	中	高	中	最终	低频变化，高频查询
连接表	中	高	中	最终	固定模式查询

选型建议：

从简单开始：优先尝试 API 组合，只有在性能不满足时才引入更复杂的方案

考虑读写比例：读多写少适合 CQRS 或数据复制

评估一致性要求：强一致性要求适合 API 组合或命令查询

考虑团队能力：CQRS 和事件驱动需要较高的技术能力

渐进式演进：可以从 API 组合逐步演进到 CQRS

三、程序员的视角

9.1 在实际项目中的应用

调试技巧：

使用分布式追踪工具（如 Jaeger、Zipkin）追踪跨服务调用链
记录详细的日志，包括请求 ID 和服务调用时间
使用 Postman 或 curl 单独测试每个服务的 API

性能优化：

始终使用批量接口避免 N+1 问题
合理设置缓存 TTL，平衡一致性和性能
监控各服务的 P95/P99 延迟，识别瓶颈

9.2 常见陷阱与误区

陷阱一：忽视 N+1 问题

在循环中逐个调用服务 API，导致大量网络请求。

陷阱二：过度使用 CQRS

在简单场景下强行使用 CQRS，增加不必要的复杂度。

陷阱三：忽视错误处理

某个服务失败时没有降级策略，导致整个查询失败。

陷阱四：数据复制不同步

没有正确处理数据变更事件，导致复制数据过时。

9.3 与其他层级的联系

跨服务查询影响：

API 设计：需要提供高效的批量接口
监控：需要追踪跨服务调用的性能和成功率
测试：需要模拟其他服务的响应进行测试

四、代码实战

9.1 完整的 API 组合实现

java

@Service
@RequiredArgsConstructor
@Slf4j
public class EcommerceQueryComposer {
    
    private final OrderServiceClient orderService;
    private final UserServiceClient userService;
    private final ProductServiceClient productService;
    private final ReviewServiceClient reviewService;
    
    private final ExecutorService executor = Executors.newFixedThreadPool(10);
    
    /**
     * 获取订单详情页所需的全部数据
     */
    public OrderPageData getOrderPageData(String orderId) {
        // 1. 获取订单（必须首先获取）
        OrderResponse order = orderService.getOrder(orderId);
        
        // 2. 并行获取其他数据
        CompletableFuture<UserResponse> userFuture = CompletableFuture
            .supplyAsync(() -> userService.getUser(order.getCustomerId()), executor);
        
        CompletableFuture<List<ProductResponse>> productsFuture = CompletableFuture
            .supplyAsync(() -> {
                List<String> productIds = order.getItems().stream()
                    .map(OrderItemResponse::getProductId)
                    .collect(Collectors.toList());
                return productService.getProducts(productIds);
            }, executor);
        
        CompletableFuture<List<ReviewResponse>> reviewsFuture = CompletableFuture
            .supplyAsync(() -> {
                List<String> productIds = order.getItems().stream()
                    .map(OrderItemResponse::getProductId)
                    .collect(Collectors.toList());
                return reviewService.getReviewsForProducts(productIds);
            }, executor);
        
        // 3. 等待所有结果
        UserResponse user = userFuture.join();
        List<ProductResponse> products = productsFuture.join();
        List<ReviewResponse> reviews = reviewsFuture.join();
        
        // 4. 组合
        return buildOrderPageData(order, user, products, reviews);
    }
    
    /**
     * 获取用户订单列表页
     */
    public UserOrderListData getUserOrderList(String customerId, int page, int size) {
        // 获取订单列表
        Page<OrderSummaryResponse> orders = orderService.getCustomerOrders(customerId, page, size);
        
        // 批量获取用户信息（虽然已知 customerId，但可能需要用户名）
        UserResponse user = userService.getUser(customerId);
        
        // 收集所有商品 ID
        Set<String> allProductIds = orders.getContent().stream()
            .flatMap(order -> order.getItemProductIds().stream())
            .collect(Collectors.toSet());
        
        // 批量获取商品信息
        Map<String, ProductResponse> productMap = productService.getProducts(new ArrayList<>(allProductIds))
            .stream()
            .collect(Collectors.toMap(ProductResponse::getId, p -> p));
        
        // 组合
        return buildUserOrderListData(orders, user, productMap);
    }
}

9.2 CQRS 读模型完整实现

java

// 读模型实体
@Entity
@Table(name = "product_search_view")
@Index(name = "idx_category", columnList = "categoryId")
@Index(name = "idx_price", columnList = "price")
@Index(name = "idx_name", columnList = "productName")
public class ProductSearchView {
    
    @Id
    private String productId;
    
    private String productName;
    private String productDescription;
    private String categoryId;
    private String categoryName;
    
    private BigDecimal price;
    private Integer stockQuantity;
    
    private Double averageRating;
    private Integer reviewCount;
    
    private String imageUrl;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
    
    // Getters and Setters
}

// Repository
@Repository
public interface ProductSearchRepository extends JpaRepository<ProductSearchView, String>, 
                                                  JpaSpecificationExecutor<ProductSearchView> {
    
    Page<ProductSearchView> findByCategoryId(String categoryId, Pageable pageable);
    
    Page<ProductSearchView> findByProductNameContainingIgnoreCase(String keyword, Pageable pageable);
    
    @Query("SELECT p FROM ProductSearchView p WHERE p.price BETWEEN :min AND :max ORDER BY p.price ASC")
    Page<ProductSearchView> findByPriceRange(@Param("min") BigDecimal min, 
                                              @Param("max") BigDecimal max, 
                                              Pageable pageable);
}

// 查询服务
@Service
@RequiredArgsConstructor
public class ProductSearchService {
    
    private final ProductSearchRepository searchRepository;
    
    public ProductDetailResponse getProductDetail(String productId) {
        ProductSearchView view = searchRepository.findById(productId)
            .orElseThrow(() -> new ProductNotFoundException(productId));
        
        return convertToDetailResponse(view);
    }
    
    public Page<ProductSummaryResponse> searchProducts(ProductSearchCriteria criteria, Pageable pageable) {
        Specification<ProductSearchView> spec = buildSpecification(criteria);
        
        return searchRepository.findAll(spec, pageable)
            .map(this::convertToSummaryResponse);
    }
    
    private Specification<ProductSearchView> buildSpecification(ProductSearchCriteria criteria) {
        return (root, query, cb) -> {
            List<Predicate> predicates = new ArrayList<>();
            
            if (criteria.getKeyword() != null && !criteria.getKeyword().isEmpty()) {
                predicates.add(cb.like(cb.lower(root.get("productName")), 
                    "%" + criteria.getKeyword().toLowerCase() + "%"));
            }
            
            if (criteria.getCategoryId() != null) {
                predicates.add(cb.equal(root.get("categoryId"), criteria.getCategoryId()));
            }
            
            if (criteria.getMinPrice() != null) {
                predicates.add(cb.greaterThanOrEqualTo(root.get("price"), criteria.getMinPrice()));
            }
            
            if (criteria.getMaxPrice() != null) {
                predicates.add(cb.lessThanOrEqualTo(root.get("price"), criteria.getMaxPrice()));
            }
            
            if (criteria.getMinRating() != null) {
                predicates.add(cb.greaterThanOrEqualTo(root.get("averageRating"), criteria.getMinRating()));
            }
            
            return cb.and(predicates.toArray(new Predicate[0]));
        };
    }
}

// 事件同步处理器
@Component
@RequiredArgsConstructor
@Slf4j
public class ProductReadModelSync {
    
    private final ProductSearchRepository searchRepository;
    private final CategoryServiceClient categoryService;
    private final ReviewServiceClient reviewService;
    
    @KafkaListener(topics = "product-events", groupId = "search-sync")
    public void handleProductCreated(ProductCreatedEvent event, Acknowledgment ack) {
        try {
            CategoryInfo category = categoryService.getCategory(event.getCategoryId());
            
            ProductSearchView view = new ProductSearchView();
            view.setProductId(event.getProductId());
            view.setProductName(event.getName());
            view.setProductDescription(event.getDescription());
            view.setCategoryId(event.getCategoryId());
            view.setCategoryName(category != null ? category.getName() : "");
            view.setPrice(event.getPrice());
            view.setStockQuantity(event.getStockQuantity());
            view.setAverageRating(0.0);
            view.setReviewCount(0);
            view.setImageUrl(event.getImageUrl());
            view.setCreatedAt(LocalDateTime.now());
            view.setUpdatedAt(LocalDateTime.now());
            
            searchRepository.save(view);
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("同步产品读模型失败", e);
            throw e;
        }
    }
    
    @KafkaListener(topics = "review-events", groupId = "search-sync")
    public void handleReviewAdded(ReviewAddedEvent event, Acknowledgment ack) {
        // 更新产品的平均评分和评论数
        ReviewStats stats = reviewService.getProductReviewStats(event.getProductId());
        
        searchRepository.findById(event.getProductId()).ifPresent(view -> {
            view.setAverageRating(stats.getAverageRating());
            view.setReviewCount(stats.getReviewCount());
            view.setUpdatedAt(LocalDateTime.now());
            searchRepository.save(view);
        });
        
        ack.acknowledge();
    }
    
    @KafkaListener(topics = "inventory-events", groupId = "search-sync")
    public void handleStockChanged(StockChangedEvent event, Acknowledgment ack) {
        searchRepository.findById(event.getProductId()).ifPresent(view -> {
            view.setStockQuantity(event.getNewQuantity());
            view.setUpdatedAt(LocalDateTime.now());
            searchRepository.save(view);
        });
        
        ack.acknowledge();
    }
}

五、深入思考

5.1 历史背景

跨服务查询的问题随着 SOA 和微服务架构的出现而凸显。早期的解决方案主要是 API 组合，但随着系统规模扩大，人们开始寻求更高效的查询方式。

CQRS 模式由 Greg Young 提出，最初用于解决复杂领域模型的读写冲突问题，后来被发现非常适合解决微服务的查询问题。

数据复制是传统分布式数据库的经典技术，在微服务中得到了新的应用。

5.2 前沿方向

GraphQL Federation：通过 GraphQL 实现声明式的跨服务数据聚合

Data Mesh：去中心化的数据架构，将数据视为产品

Change Data Capture (CDC)：基于数据库日志的实时数据同步

Vector Databases：用于 AI 驱动的语义搜索

5.3 延伸阅读

Chris Richardson. "Microservices Patterns". Manning, 2018
Sam Newman. "Building Microservices". O'Reilly, 2015
Martin Fowler. "CQRS". martinfowler.com

六、实验笔记

实验一：API 组合性能优化

目标：对比串行和并行 API 组合的性能差异。

步骤：

实现串行的 API 组合器

实现并行的 API 组合器

使用 JMeter 进行压力测试

对比两种实现的 P95/P99 延迟

添加缓存后再次测试

预期结果：并行实现显著降低延迟，缓存进一步提升性能。

实验二：CQRS 读模型实现

目标：为一个博客系统实现 CQRS 读模型。

步骤：

设计写模型（文章、评论聚合）

设计读模型（文章列表视图、文章详情视图）

实现事件同步处理器

实现查询服务

测试数据同步的延迟和一致性

预期结果：成功实现 CQRS，理解读模型的设计和同步机制。

七、本章小结

本章系统介绍了微服务架构中的跨服务查询策略。我们首先分析了跨服务查询的根本挑战：数据库隔离导致无法使用 JOIN，网络通信带来延迟和可靠性问题。

API 组合是最直观的解决方案，通过调用多个服务的 API 并合并结果来实现跨服务查询。并行调用和批量接口可以显著优化性能，但 API 组合仍然受限于网络延迟和服务可用性。

CQRS 通过为查询专门设计读模型，将跨服务查询转化为本地查询，大幅提升了查询性能。读模型通过事件驱动的方式异步同步，实现了最终一致性。

数据复制是另一种有效策略，通过将其他服务的数据复制到本地，避免了跨服务调用。基于事件的增量复制可以减少同步延迟和数据冗余。

连接表模式在单个服务内部维护宽表，适合固定模式的查询需求。

每种策略都有其优缺点和适用场景。在实际项目中，往往需要组合使用多种策略，根据具体需求做出权衡。

八、习题

策略对比：比较 API 组合、CQRS 和数据复制三种策略的优缺点，说明各自适用的场景。

API 组合优化：为一个电商首页设计 API 组合方案，包括并行调用和缓存策略。

读模型设计：为一个社交媒体系统设计 CQRS 读模型，支持动态 feed、用户主页和搜索功能。

数据复制实现：设计一个基于事件的用户数据复制方案，包括全量初始化和增量同步。

N+1 问题解决：分析一个存在 N+1 问题的代码，重构为批量调用方式。

一致性处理：设计一个方案来处理 CQRS 读写模型之间的数据延迟，确保用户体验。

性能测试：设计实验对比不同查询策略在不同负载下的性能表现。

错误恢复：设计事件同步失败的恢复机制，包括重试、死信队列和人工干预。

架构演进：为一个现有系统规划从 API 组合到 CQRS 的演进路径。

案例分析：研究 LinkedIn 或 Twitter 的数据查询架构，总结其设计原则和经验教训。

第9章 跨服务查询 - 分布式查询策略 (Cross-Service Queries: Distributed Query Strategies)

一、导读

1.1 本章学习目标

1.2 为什么本章如此重要

1.3 核心问题引导

二、核心概念详解

9.1 跨服务查询的根本挑战

9.2 API 组合模式

9.3 CQRS 与读模型

9.4 命令查询模式

9.5 数据复制策略

9.6 连接表模式

9.7 查询策略对比与选型

三、程序员的视角

9.1 在实际项目中的应用

9.2 常见陷阱与误区

9.3 与其他层级的联系

四、代码实战

9.1 完整的 API 组合实现

9.2 CQRS 读模型完整实现

五、深入思考

5.1 历史背景

5.2 前沿方向

5.3 延伸阅读

六、实验笔记

实验一：API 组合性能优化

实验二：CQRS 读模型实现

七、本章小结

八、习题

第9章跨服务查询 - 分布式查询策略 (Cross-Service Queries: Distributed Query Strategies)