第276集MongoDB微服务跨库查询，如何解决架构实战：MongoDB跨库查询策略、数据聚合方案与企业级MongoDB微服务数据查询解决方案

/**
 * MongoDB聚合服务
 */
@Service
public class MongoAggregationService {

    @Autowired
    private MongoTemplate userMongoTemplate;

    @Autowired
    private MongoTemplate orderMongoTemplate;

    @Autowired
    private MongoTemplate productMongoTemplate;

    @Autowired
    private MongoTemplate paymentMongoTemplate;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private final String AGGREGATION_CACHE_PREFIX = "mongo_aggregation_cache:";
    private final long AGGREGATION_CACHE_EXPIRE = 1800; // 30分钟

    /**
     * 跨库聚合查询用户订单信息
     */
    public UserOrderAggregationResult aggregateUserOrderInfo(Long userId) {
        try {
            // 从缓存获取
            String cacheKey = AGGREGATION_CACHE_PREFIX + "user_order:" + userId;
            UserOrderAggregationResult cachedResult = (UserOrderAggregationResult) redisTemplate.opsForValue().get(cacheKey);

            if (cachedResult != null) {
                return cachedResult;
            }

            // 1. 获取用户信息
            UserInfo userInfo = getUserInfo(userId);

            // 2. 获取用户订单聚合信息
            OrderAggregationInfo orderAggregation = getOrderAggregationInfo(userId);

            // 3. 获取订单商品聚合信息
            ProductAggregationInfo productAggregation = getProductAggregationInfo(userId);

            // 4. 聚合数据
            UserOrderAggregationResult result = aggregateUserOrderData(userInfo, orderAggregation, productAggregation);

            // 缓存结果
            redisTemplate.opsForValue().set(cacheKey, result, Duration.ofSeconds(AGGREGATION_CACHE_EXPIRE));

            return result;

        } catch (Exception e) {
            log.error("跨库聚合查询用户订单信息失败", e);
            throw new MongoCrossDatabaseQueryException("跨库聚合查询用户订单信息失败", e);
        }
    }

    /**
     * 获取用户信息
     */
    private UserInfo getUserInfo(Long userId) {
        try {
            Query query = new Query(Criteria.where("_id").is(userId));
            return userMongoTemplate.findOne(query, UserInfo.class, "users");
        } catch (Exception e) {
            log.error("获取用户信息失败", e);
            throw new MongoCrossDatabaseQueryException("获取用户信息失败", e);
        }
    }

    /**
     * 获取订单聚合信息
     */
    private OrderAggregationInfo getOrderAggregationInfo(Long userId) {
        try {
            // 构建聚合管道
            Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.match(Criteria.where("userId").is(userId)),
                Aggregation.group("status")
                    .count().as("count")
                    .sum("amount").as("totalAmount")
                    .avg("amount").as("avgAmount"),
                Aggregation.sort(Sort.Direction.DESC, "count")
            );

            AggregationResults<OrderAggregationInfo> results = orderMongoTemplate.aggregate(
                    aggregation, "orders", OrderAggregationInfo.class);

            List<OrderAggregationInfo> aggregationList = results.getMappedResults();

            // 构建聚合信息
            OrderAggregationInfo aggregationInfo = new OrderAggregationInfo();
            aggregationInfo.setUserId(userId);
            aggregationInfo.setTotalOrders(aggregationList.stream().mapToInt(OrderAggregationInfo::getCount).sum());
            aggregationInfo.setTotalAmount(aggregationList.stream().mapToDouble(OrderAggregationInfo::getTotalAmount).sum());
            aggregationInfo.setAvgAmount(aggregationList.stream().mapToDouble(OrderAggregationInfo::getAvgAmount).average().orElse(0.0));
            aggregationInfo.setStatusAggregation(aggregationList);

            return aggregationInfo;

        } catch (Exception e) {
            log.error("获取订单聚合信息失败", e);
            throw new MongoCrossDatabaseQueryException("获取订单聚合信息失败", e);
        }
    }

    /**
     * 获取商品聚合信息
     */
    private ProductAggregationInfo getProductAggregationInfo(Long userId) {
        try {
            // 构建聚合管道
            Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.lookup("orders", "_id", "productId", "orders"),
                Aggregation.unwind("orders"),
                Aggregation.match(Criteria.where("orders.userId").is(userId)),
                Aggregation.group("_id")
                    .first("name").as("productName")
                    .first("price").as("productPrice")
                    .count().as("orderCount")
                    .sum("orders.quantity").as("totalQuantity")
                    .sum("orders.amount").as("totalAmount"),
                Aggregation.sort(Sort.Direction.DESC, "totalAmount")
            );

            AggregationResults<ProductAggregationInfo> results = productMongoTemplate.aggregate(
                    aggregation, "products", ProductAggregationInfo.class);

            List<ProductAggregationInfo> aggregationList = results.getMappedResults();

            // 构建聚合信息
            ProductAggregationInfo aggregationInfo = new ProductAggregationInfo();
            aggregationInfo.setUserId(userId);
            aggregationInfo.setTotalProducts(aggregationList.size());
            aggregationInfo.setTotalQuantity(aggregationList.stream().mapToInt(ProductAggregationInfo::getTotalQuantity).sum());
            aggregationInfo.setTotalAmount(aggregationList.stream().mapToDouble(ProductAggregationInfo::getTotalAmount).sum());
            aggregationInfo.setProductAggregation(aggregationList);

            return aggregationInfo;

        } catch (Exception e) {
            log.error("获取商品聚合信息失败", e);
            throw new MongoCrossDatabaseQueryException("获取商品聚合信息失败", e);
        }
    }

    /**
     * 聚合用户订单数据
     */
    private UserOrderAggregationResult aggregateUserOrderData(UserInfo userInfo, 
            OrderAggregationInfo orderAggregation, ProductAggregationInfo productAggregation) {
        try {
            UserOrderAggregationResult result = new UserOrderAggregationResult();
            result.setUserId(userInfo.getId());
            result.setUserName(userInfo.getName());
            result.setUserEmail(userInfo.getEmail());
            result.setUserPhone(userInfo.getPhone());

            // 聚合订单信息
            result.setTotalOrders(orderAggregation.getTotalOrders());
            result.setTotalOrderAmount(orderAggregation.getTotalAmount());
            result.setAvgOrderAmount(orderAggregation.getAvgAmount());
            result.setOrderStatusAggregation(orderAggregation.getStatusAggregation());

            // 聚合商品信息
            result.setTotalProducts(productAggregation.getTotalProducts());
            result.setTotalQuantity(productAggregation.getTotalQuantity());
            result.setTotalProductAmount(productAggregation.getTotalAmount());
            result.setProductAggregation(productAggregation.getProductAggregation());

            // 计算统计信息
            calculateUserOrderStatistics(result);

            return result;

        } catch (Exception e) {
            log.error("聚合用户订单数据失败", e);
            throw new MongoCrossDatabaseQueryException("聚合用户订单数据失败", e);
        }
    }

    /**
     * 计算用户订单统计信息
     */
    private void calculateUserOrderStatistics(UserOrderAggregationResult result) {
        try {
            UserOrderStatistics statistics = new UserOrderStatistics();

            // 计算订单统计
            statistics.setOrderCount(result.getTotalOrders());
            statistics.setOrderAmount(result.getTotalOrderAmount());
            statistics.setAvgOrderAmount(result.getAvgOrderAmount());

            // 计算商品统计
            statistics.setProductCount(result.getTotalProducts());
            statistics.setProductQuantity(result.getTotalQuantity());
            statistics.setProductAmount(result.getTotalProductAmount());

            // 计算综合统计
            statistics.setTotalAmount(result.getTotalOrderAmount() + result.getTotalProductAmount());
            statistics.setAvgProductPrice(result.getTotalQuantity() > 0 ? 
                    result.getTotalProductAmount() / result.getTotalQuantity() : 0.0);

            result.setStatistics(statistics);

        } catch (Exception e) {
            log.error("计算用户订单统计信息失败", e);
        }
    }

    /**
     * 并行聚合查询多个集合
     */
    public CompletableFuture<MultiCollectionAggregationResult> aggregateMultipleCollectionsAsync(
            MultiCollectionAggregationRequest request) {
        try {
            CompletableFuture<UserInfo> userFuture = CompletableFuture.supplyAsync(() -> {
                return getUserInfo(request.getUserId());
            });

            CompletableFuture<OrderAggregationInfo> orderFuture = CompletableFuture.supplyAsync(() -> {
                return getOrderAggregationInfo(request.getUserId());
            });

            CompletableFuture<ProductAggregationInfo> productFuture = CompletableFuture.supplyAsync(() -> {
                return getProductAggregationInfo(request.getUserId());
            });

            CompletableFuture<PaymentAggregationInfo> paymentFuture = CompletableFuture.supplyAsync(() -> {
                return getPaymentAggregationInfo(request.getUserId());
            });

            // 等待所有查询完成
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                    userFuture, orderFuture, productFuture, paymentFuture);

            return allFutures.thenApply(v -> {
                MultiCollectionAggregationResult result = new MultiCollectionAggregationResult();
                try {
                    result.setUserInfo(userFuture.get());
                    result.setOrderAggregation(orderFuture.get());
                    result.setProductAggregation(productFuture.get());
                    result.setPaymentAggregation(paymentFuture.get());
                    result.setStatus(AggregationStatus.SUCCESS);
                } catch (Exception e) {
                    result.setStatus(AggregationStatus.FAILED);
                    result.setErrorMessage(e.getMessage());
                }
                return result;
            });

        } catch (Exception e) {
            log.error("并行聚合查询多个集合失败", e);
            CompletableFuture<MultiCollectionAggregationResult> failedFuture = new CompletableFuture<>();
            failedFuture.completeExceptionally(e);
            return failedFuture;
        }
    }

    /**
     * 获取支付聚合信息
     */
    private PaymentAggregationInfo getPaymentAggregationInfo(Long userId) {
        try {
            // 构建聚合管道
            Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.match(Criteria.where("userId").is(userId)),
                Aggregation.group("paymentMethod")
                    .count().as("count")
                    .sum("amount").as("totalAmount")
                    .avg("amount").as("avgAmount"),
                Aggregation.sort(Sort.Direction.DESC, "totalAmount")
            );

            AggregationResults<PaymentAggregationInfo> results = paymentMongoTemplate.aggregate(
                    aggregation, "payments", PaymentAggregationInfo.class);

            List<PaymentAggregationInfo> aggregationList = results.getMappedResults();

            // 构建聚合信息
            PaymentAggregationInfo aggregationInfo = new PaymentAggregationInfo();
            aggregationInfo.setUserId(userId);
            aggregationInfo.setTotalPayments(aggregationList.stream().mapToInt(PaymentAggregationInfo::getCount).sum());
            aggregationInfo.setTotalAmount(aggregationList.stream().mapToDouble(PaymentAggregationInfo::getTotalAmount).sum());
            aggregationInfo.setAvgAmount(aggregationList.stream().mapToDouble(PaymentAggregationInfo::getAvgAmount).average().orElse(0.0));
            aggregationInfo.setPaymentMethodAggregation(aggregationList);

            return aggregationInfo;

        } catch (Exception e) {
            log.error("获取支付聚合信息失败", e);
            throw new MongoCrossDatabaseQueryException("获取支付聚合信息失败", e);
        }
    }

    /**
     * 使用MongoDB聚合管道进行复杂查询
     */
    public ComplexAggregationResult executeComplexAggregation(ComplexAggregationRequest request) {
        try {
            // 构建复杂的聚合管道
            List<AggregationOperation> operations = new ArrayList<>();

            // 1. 匹配阶段
            if (request.getMatchCriteria() != null) {
                operations.add(Aggregation.match(request.getMatchCriteria()));
            }

            // 2. 关联阶段
            if (request.getLookupOperations() != null) {
                for (LookupOperation lookup : request.getLookupOperations()) {
                    operations.add(Aggregation.lookup(lookup.getFrom(), lookup.getLocalField(), 
                            lookup.getForeignField(), lookup.getAs()));
                }
            }

            // 3. 展开阶段
            if (request.isUnwind()) {
                operations.add(Aggregation.unwind(request.getUnwindField()));
            }

            // 4. 分组阶段
            if (request.getGroupOperations() != null) {
                for (GroupOperation group : request.getGroupOperations()) {
                    operations.add(Aggregation.group(group.getGroupBy())
                            .count().as("count")
                            .sum(group.getSumField()).as("total")
                            .avg(group.getAvgField()).as("average"));
                }
            }

            // 5. 排序阶段
            if (request.getSortOperations() != null) {
                for (SortOperation sort : request.getSortOperations()) {
                    operations.add(Aggregation.sort(sort.getDirection(), sort.getFields()));
                }
            }

            // 6. 限制阶段
            if (request.getLimit() > 0) {
                operations.add(Aggregation.limit(request.getLimit()));
            }

            // 执行聚合
            Aggregation aggregation = Aggregation.newAggregation(operations);
            AggregationResults<ComplexAggregationResult> results = userMongoTemplate.aggregate(
                    aggregation, request.getCollection(), ComplexAggregationResult.class);

            List<ComplexAggregationResult> resultList = results.getMappedResults();

            // 构建结果
            ComplexAggregationResult result = new ComplexAggregationResult();
            result.setTotalCount(resultList.size());
            result.setResults(resultList);
            result.setExecutionTime(System.currentTimeMillis() - request.getStartTime());

            return result;

        } catch (Exception e) {
            log.error("执行复杂聚合查询失败", e);
            throw new MongoCrossDatabaseQueryException("执行复杂聚合查询失败", e);
        }
    }
}

/**
 * MongoDB跨库查询服务
 */
@Service
public class MongoCrossDatabaseQueryService {

    @Autowired
    private MongoTemplate userMongoTemplate;

    @Autowired
    private MongoTemplate orderMongoTemplate;

    @Autowired
    private MongoTemplate productMongoTemplate;

    @Autowired
    private MongoTemplate paymentMongoTemplate;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private final String CROSS_DB_CACHE_PREFIX = "cross_db_cache:";
    private final long CROSS_DB_CACHE_EXPIRE = 1800; // 30分钟

    /**
     * 跨库查询用户完整信息
     */
    public UserCompleteInfo crossDatabaseQueryUserCompleteInfo(Long userId) {
        try {
            // 从缓存获取
            String cacheKey = CROSS_DB_CACHE_PREFIX + "user_complete:" + userId;
            UserCompleteInfo cachedInfo = (UserCompleteInfo) redisTemplate.opsForValue().get(cacheKey);

            if (cachedInfo != null) {
                return cachedInfo;
            }

            // 并行查询多个数据库
            CompletableFuture<UserInfo> userFuture = CompletableFuture.supplyAsync(() -> {
                return getUserInfoFromUserDB(userId);
            });

            CompletableFuture<List<OrderInfo>> ordersFuture = CompletableFuture.supplyAsync(() -> {
                return getOrdersFromOrderDB(userId);
            });

            CompletableFuture<List<ProductInfo>> productsFuture = CompletableFuture.supplyAsync(() -> {
                return getProductsFromProductDB(userId);
            });

            CompletableFuture<List<PaymentInfo>> paymentsFuture = CompletableFuture.supplyAsync(() -> {
                return getPaymentsFromPaymentDB(userId);
            });

            // 等待所有查询完成
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                    userFuture, ordersFuture, productsFuture, paymentsFuture);

            allFutures.get(); // 等待完成

            // 聚合数据
            UserCompleteInfo userCompleteInfo = new UserCompleteInfo();
            userCompleteInfo.setUserId(userId);
            userCompleteInfo.setUserInfo(userFuture.get());
            userCompleteInfo.setOrders(ordersFuture.get());
            userCompleteInfo.setProducts(productsFuture.get());
            userCompleteInfo.setPayments(paymentsFuture.get());

            // 计算统计信息
            calculateUserCompleteStatistics(userCompleteInfo);

            // 缓存结果
            redisTemplate.opsForValue().set(cacheKey, userCompleteInfo, 
                    Duration.ofSeconds(CROSS_DB_CACHE_EXPIRE));

            return userCompleteInfo;

        } catch (Exception e) {
            log.error("跨库查询用户完整信息失败", e);
            throw new MongoCrossDatabaseQueryException("跨库查询用户完整信息失败", e);
        }
    }

    /**
     * 从用户数据库获取用户信息
     */
    private UserInfo getUserInfoFromUserDB(Long userId) {
        try {
            Query query = new Query(Criteria.where("_id").is(userId));
            return userMongoTemplate.findOne(query, UserInfo.class, "users");
        } catch (Exception e) {
            log.error("从用户数据库获取用户信息失败", e);
            throw new MongoCrossDatabaseQueryException("从用户数据库获取用户信息失败", e);
        }
    }

    /**
     * 从订单数据库获取订单信息
     */
    private List<OrderInfo> getOrdersFromOrderDB(Long userId) {
        try {
            Query query = new Query(Criteria.where("userId").is(userId));
            query.with(Sort.by(Sort.Direction.DESC, "createTime"));
            return orderMongoTemplate.find(query, OrderInfo.class, "orders");
        } catch (Exception e) {
            log.error("从订单数据库获取订单信息失败", e);
            throw new MongoCrossDatabaseQueryException("从订单数据库获取订单信息失败", e);
        }
    }

    /**
     * 从商品数据库获取商品信息
     */
    private List<ProductInfo> getProductsFromProductDB(Long userId) {
        try {
            // 先获取用户购买的商品ID
            Query orderQuery = new Query(Criteria.where("userId").is(userId));
            orderQuery.fields().include("productId");
            List<OrderInfo> orders = orderMongoTemplate.find(orderQuery, OrderInfo.class, "orders");

            if (orders.isEmpty()) {
                return new ArrayList<>();
            }

            // 提取商品ID
            List<Long> productIds = orders.stream()
                    .map(OrderInfo::getProductId)
                    .distinct()
                    .collect(Collectors.toList());

            // 查询商品信息
            Query productQuery = new Query(Criteria.where("_id").in(productIds));
            return productMongoTemplate.find(productQuery, ProductInfo.class, "products");

        } catch (Exception e) {
            log.error("从商品数据库获取商品信息失败", e);
            throw new MongoCrossDatabaseQueryException("从商品数据库获取商品信息失败", e);
        }
    }

    /**
     * 从支付数据库获取支付信息
     */
    private List<PaymentInfo> getPaymentsFromPaymentDB(Long userId) {
        try {
            Query query = new Query(Criteria.where("userId").is(userId));
            query.with(Sort.by(Sort.Direction.DESC, "createTime"));
            return paymentMongoTemplate.find(query, PaymentInfo.class, "payments");
        } catch (Exception e) {
            log.error("从支付数据库获取支付信息失败", e);
            throw new MongoCrossDatabaseQueryException("从支付数据库获取支付信息失败", e);
        }
    }

    /**
     * 计算用户完整统计信息
     */
    private void calculateUserCompleteStatistics(UserCompleteInfo userCompleteInfo) {
        try {
            UserCompleteStatistics statistics = new UserCompleteStatistics();

            // 订单统计
            List<OrderInfo> orders = userCompleteInfo.getOrders();
            if (orders != null && !orders.isEmpty()) {
                statistics.setTotalOrders(orders.size());
                statistics.setTotalOrderAmount(orders.stream()
                        .mapToDouble(OrderInfo::getAmount)
                        .sum());

                // 按状态统计
                Map<String, Long> statusCount = orders.stream()
                        .collect(Collectors.groupingBy(OrderInfo::getStatus, Collectors.counting()));
                statistics.setOrderStatusCount(statusCount);

                // 按时间统计
                Map<String, Long> timeCount = orders.stream()
                        .collect(Collectors.groupingBy(order -> 
                                order.getCreateTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDate().toString(), 
                                Collectors.counting()));
                statistics.setOrderTimeCount(timeCount);
            }

            // 商品统计
            List<ProductInfo> products = userCompleteInfo.getProducts();
            if (products != null && !products.isEmpty()) {
                statistics.setTotalProducts(products.size());
                statistics.setTotalProductAmount(products.stream()
                        .mapToDouble(ProductInfo::getPrice)
                        .sum());

                // 按分类统计
                Map<String, Long> categoryCount = products.stream()
                        .collect(Collectors.groupingBy(ProductInfo::getCategory, Collectors.counting()));
                statistics.setProductCategoryCount(categoryCount);
            }

            // 支付统计
            List<PaymentInfo> payments = userCompleteInfo.getPayments();
            if (payments != null && !payments.isEmpty()) {
                statistics.setTotalPayments(payments.size());
                statistics.setTotalPaymentAmount(payments.stream()
                        .mapToDouble(PaymentInfo::getAmount)
                        .sum());

                // 按支付方式统计
                Map<String, Long> paymentMethodCount = payments.stream()
                        .collect(Collectors.groupingBy(PaymentInfo::getPaymentMethod, Collectors.counting()));
                statistics.setPaymentMethodCount(paymentMethodCount);
            }

            userCompleteInfo.setStatistics(statistics);

        } catch (Exception e) {
            log.error("计算用户完整统计信息失败", e);
        }
    }

    /**
     * 跨库查询订单详细信息
     */
    public OrderDetailInfo crossDatabaseQueryOrderDetailInfo(Long orderId) {
        try {
            // 从缓存获取
            String cacheKey = CROSS_DB_CACHE_PREFIX + "order_detail:" + orderId;
            OrderDetailInfo cachedInfo = (OrderDetailInfo) redisTemplate.opsForValue().get(cacheKey);

            if (cachedInfo != null) {
                return cachedInfo;
            }

            // 并行查询
            CompletableFuture<OrderInfo> orderFuture = CompletableFuture.supplyAsync(() -> {
                return getOrderInfoFromOrderDB(orderId);
            });

            CompletableFuture<List<ProductInfo>> productsFuture = CompletableFuture.supplyAsync(() -> {
                OrderInfo order = orderFuture.join();
                return getProductsByOrderIdFromProductDB(order.getProductId());
            });

            CompletableFuture<UserInfo> userFuture = CompletableFuture.supplyAsync(() -> {
                OrderInfo order = orderFuture.join();
                return getUserInfoFromUserDB(order.getUserId());
            });

            CompletableFuture<List<PaymentInfo>> paymentsFuture = CompletableFuture.supplyAsync(() -> {
                OrderInfo order = orderFuture.join();
                return getPaymentsByOrderIdFromPaymentDB(orderId);
            });

            // 等待完成
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                    orderFuture, productsFuture, userFuture, paymentsFuture);
            allFutures.get();

            // 聚合数据
            OrderDetailInfo orderDetailInfo = new OrderDetailInfo();
            orderDetailInfo.setOrderInfo(orderFuture.get());
            orderDetailInfo.setProducts(productsFuture.get());
            orderDetailInfo.setUserInfo(userFuture.get());
            orderDetailInfo.setPayments(paymentsFuture.get());

            // 缓存结果
            redisTemplate.opsForValue().set(cacheKey, orderDetailInfo, 
                    Duration.ofSeconds(CROSS_DB_CACHE_EXPIRE));

            return orderDetailInfo;

        } catch (Exception e) {
            log.error("跨库查询订单详细信息失败", e);
            throw new MongoCrossDatabaseQueryException("跨库查询订单详细信息失败", e);
        }
    }

    /**
     * 从订单数据库获取订单信息
     */
    private OrderInfo getOrderInfoFromOrderDB(Long orderId) {
        try {
            Query query = new Query(Criteria.where("_id").is(orderId));
            return orderMongoTemplate.findOne(query, OrderInfo.class, "orders");
        } catch (Exception e) {
            log.error("从订单数据库获取订单信息失败", e);
            throw new MongoCrossDatabaseQueryException("从订单数据库获取订单信息失败", e);
        }
    }

    /**
     * 根据订单ID从商品数据库获取商品信息
     */
    private List<ProductInfo> getProductsByOrderIdFromProductDB(Long productId) {
        try {
            Query query = new Query(Criteria.where("_id").is(productId));
            return productMongoTemplate.find(query, ProductInfo.class, "products");
        } catch (Exception e) {
            log.error("根据订单ID从商品数据库获取商品信息失败", e);
            throw new MongoCrossDatabaseQueryException("根据订单ID从商品数据库获取商品信息失败", e);
        }
    }

    /**
     * 根据订单ID从支付数据库获取支付信息
     */
    private List<PaymentInfo> getPaymentsByOrderIdFromPaymentDB(Long orderId) {
        try {
            Query query = new Query(Criteria.where("orderId").is(orderId));
            return paymentMongoTemplate.find(query, PaymentInfo.class, "payments");
        } catch (Exception e) {
            log.error("根据订单ID从支付数据库获取支付信息失败", e);
            throw new MongoCrossDatabaseQueryException("根据订单ID从支付数据库获取支付信息失败", e);
        }
    }

    /**
     * 跨库查询商品销售统计
     */
    public ProductSalesStatistics crossDatabaseQueryProductSalesStatistics(Long productId, Date startTime, Date endTime) {
        try {
            // 从缓存获取
            String cacheKey = CROSS_DB_CACHE_PREFIX + "product_sales:" + productId + ":" + 
                    startTime.getTime() + ":" + endTime.getTime();
            ProductSalesStatistics cachedStats = (ProductSalesStatistics) redisTemplate.opsForValue().get(cacheKey);

            if (cachedStats != null) {
                return cachedStats;
            }

            // 并行查询
            CompletableFuture<ProductInfo> productFuture = CompletableFuture.supplyAsync(() -> {
                return getProductInfoFromProductDB(productId);
            });

            CompletableFuture<List<OrderInfo>> ordersFuture = CompletableFuture.supplyAsync(() -> {
                return getOrdersByProductIdFromOrderDB(productId, startTime, endTime);
            });

            CompletableFuture<List<PaymentInfo>> paymentsFuture = CompletableFuture.supplyAsync(() -> {
                return getPaymentsByProductIdFromPaymentDB(productId, startTime, endTime);
            });

            // 等待完成
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                    productFuture, ordersFuture, paymentsFuture);
            allFutures.get();

            // 聚合数据
            ProductSalesStatistics statistics = new ProductSalesStatistics();
            statistics.setProductInfo(productFuture.get());
            statistics.setTotalOrders(ordersFuture.get().size());
            statistics.setTotalSales(ordersFuture.get().stream()
                    .mapToDouble(OrderInfo::getAmount)
                    .sum());
            statistics.setTotalPayments(paymentsFuture.get().size());
            statistics.setTotalPaymentAmount(paymentsFuture.get().stream()
                    .mapToDouble(PaymentInfo::getAmount)
                    .sum());

            // 计算趋势
            calculateSalesTrend(statistics, ordersFuture.get());

            // 缓存结果
            redisTemplate.opsForValue().set(cacheKey, statistics, 
                    Duration.ofSeconds(CROSS_DB_CACHE_EXPIRE));

            return statistics;

        } catch (Exception e) {
            log.error("跨库查询商品销售统计失败", e);
            throw new MongoCrossDatabaseQueryException("跨库查询商品销售统计失败", e);
        }
    }

    /**
     * 从商品数据库获取商品信息
     */
    private ProductInfo getProductInfoFromProductDB(Long productId) {
        try {
            Query query = new Query(Criteria.where("_id").is(productId));
            return productMongoTemplate.findOne(query, ProductInfo.class, "products");
        } catch (Exception e) {
            log.error("从商品数据库获取商品信息失败", e);
            throw new MongoCrossDatabaseQueryException("从商品数据库获取商品信息失败", e);
        }
    }

    /**
     * 根据商品ID从订单数据库获取订单信息
     */
    private List<OrderInfo> getOrdersByProductIdFromOrderDB(Long productId, Date startTime, Date endTime) {
        try {
            Query query = new Query(Criteria.where("productId").is(productId)
                    .and("createTime").gte(startTime).lte(endTime));
            return orderMongoTemplate.find(query, OrderInfo.class, "orders");
        } catch (Exception e) {
            log.error("根据商品ID从订单数据库获取订单信息失败", e);
            throw new MongoCrossDatabaseQueryException("根据商品ID从订单数据库获取订单信息失败", e);
        }
    }

    /**
     * 根据商品ID从支付数据库获取支付信息
     */
    private List<PaymentInfo> getPaymentsByProductIdFromPaymentDB(Long productId, Date startTime, Date endTime) {
        try {
            Query query = new Query(Criteria.where("productId").is(productId)
                    .and("createTime").gte(startTime).lte(endTime));
            return paymentMongoTemplate.find(query, PaymentInfo.class, "payments");
        } catch (Exception e) {
            log.error("根据商品ID从支付数据库获取支付信息失败", e);
            throw new MongoCrossDatabaseQueryException("根据商品ID从支付数据库获取支付信息失败", e);
        }
    }

    /**
     * 计算销售趋势
     */
    private void calculateSalesTrend(ProductSalesStatistics statistics, List<OrderInfo> orders) {
        try {
            // 按日期分组统计
            Map<String, Double> dailySales = orders.stream()
                    .collect(Collectors.groupingBy(order -> 
                            order.getCreateTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDate().toString(),
                            Collectors.summingDouble(OrderInfo::getAmount)));

            statistics.setDailySalesTrend(dailySales);

            // 按周分组统计
            Map<String, Double> weeklySales = orders.stream()
                    .collect(Collectors.groupingBy(order -> {
                        LocalDate date = order.getCreateTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDate();
                        return date.getYear() + "-W" + date.get(WeekFields.ISO.weekOfYear());
                    }, Collectors.summingDouble(OrderInfo::getAmount)));

            statistics.setWeeklySalesTrend(weeklySales);

            // 按月分组统计
            Map<String, Double> monthlySales = orders.stream()
                    .collect(Collectors.groupingBy(order -> {
                        LocalDate date = order.getCreateTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDate();
                        return date.getYear() + "-" + String.format("%02d", date.getMonthValue());
                    }, Collectors.summingDouble(OrderInfo::getAmount)));

            statistics.setMonthlySalesTrend(monthlySales);

        } catch (Exception e) {
            log.error("计算销售趋势失败", e);
        }
    }
}

/**
 * MongoDB事件驱动服务
 */
@Service
public class MongoEventDrivenService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private MongoAggregationService mongoAggregationService;

    @Autowired
    private MongoCrossDatabaseQueryService crossDatabaseQueryService;

    private final String MONGO_EVENT_QUEUE_PREFIX = "mongo_cross_db_query.event.";
    private final String MONGO_EVENT_CACHE_PREFIX = "mongo_event_cache:";

    /**
     * 发布MongoDB跨库查询事件
     */
    public void publishMongoCrossDatabaseQueryEvent(MongoCrossDatabaseQueryEvent event) {
        try {
            // 设置事件ID
            if (event.getEventId() == null) {
                event.setEventId(UUID.randomUUID().toString());
            }

            // 设置时间戳
            event.setTimestamp(new Date());

            // 发布到消息队列
            String queueName = MONGO_EVENT_QUEUE_PREFIX + event.getEventType();
            rabbitTemplate.convertAndSend(queueName, event);

            log.info("发布MongoDB跨库查询事件成功: {}", event.getEventId());

        } catch (Exception e) {
            log.error("发布MongoDB跨库查询事件失败", e);
            throw new MongoCrossDatabaseQueryException("发布MongoDB跨库查询事件失败", e);
        }
    }

    /**
     * 处理用户信息更新事件
     */
    @RabbitListener(queues = "mongo_cross_db_query.event.user.updated")
    public void handleUserUpdatedEvent(MongoUserUpdatedEvent event) {
        try {
            log.info("处理MongoDB用户信息更新事件: {}", event.getUserId());

            // 清除相关缓存
            clearUserRelatedCache(event.getUserId());

            // 更新聚合数据
            updateUserAggregatedData(event.getUserId());

            // 发布数据更新事件
            MongoDataUpdatedEvent dataUpdatedEvent = new MongoDataUpdatedEvent();
            dataUpdatedEvent.setEventId(UUID.randomUUID().toString());
            dataUpdatedEvent.setEventType("MONGO_USER_DATA_UPDATED");
            dataUpdatedEvent.setUserId(event.getUserId());
            dataUpdatedEvent.setTimestamp(new Date());

            publishMongoCrossDatabaseQueryEvent(dataUpdatedEvent);

        } catch (Exception e) {
            log.error("处理MongoDB用户信息更新事件失败", e);
        }
    }

    /**
     * 处理订单创建事件
     */
    @RabbitListener(queues = "mongo_cross_db_query.event.order.created")
    public void handleOrderCreatedEvent(MongoOrderCreatedEvent event) {
        try {
            log.info("处理MongoDB订单创建事件: {}", event.getOrderId());

            // 清除相关缓存
            clearOrderRelatedCache(event.getOrderId());

            // 更新用户聚合数据
            updateUserAggregatedData(event.getUserId());

            // 更新商品聚合数据
            updateProductAggregatedData(event.getProductId());

            // 发布数据更新事件
            MongoDataUpdatedEvent dataUpdatedEvent = new MongoDataUpdatedEvent();
            dataUpdatedEvent.setEventId(UUID.randomUUID().toString());
            dataUpdatedEvent.setEventType("MONGO_ORDER_DATA_UPDATED");
            dataUpdatedEvent.setOrderId(event.getOrderId());
            dataUpdatedEvent.setUserId(event.getUserId());
            dataUpdatedEvent.setTimestamp(new Date());

            publishMongoCrossDatabaseQueryEvent(dataUpdatedEvent);

        } catch (Exception e) {
            log.error("处理MongoDB订单创建事件失败", e);
        }
    }

    /**
     * 处理支付完成事件
     */
    @RabbitListener(queues = "mongo_cross_db_query.event.payment.completed")
    public void handlePaymentCompletedEvent(MongoPaymentCompletedEvent event) {
        try {
            log.info("处理MongoDB支付完成事件: {}", event.getPaymentId());

            // 清除相关缓存
            clearPaymentRelatedCache(event.getPaymentId());

            // 更新订单聚合数据
            updateOrderAggregatedData(event.getOrderId());

            // 更新用户聚合数据
            updateUserAggregatedData(event.getUserId());

            // 发布数据更新事件
            MongoDataUpdatedEvent dataUpdatedEvent = new MongoDataUpdatedEvent();
            dataUpdatedEvent.setEventId(UUID.randomUUID().toString());
            dataUpdatedEvent.setEventType("MONGO_PAYMENT_DATA_UPDATED");
            dataUpdatedEvent.setPaymentId(event.getPaymentId());
            dataUpdatedEvent.setOrderId(event.getOrderId());
            dataUpdatedEvent.setUserId(event.getUserId());
            dataUpdatedEvent.setTimestamp(new Date());

            publishMongoCrossDatabaseQueryEvent(dataUpdatedEvent);

        } catch (Exception e) {
            log.error("处理MongoDB支付完成事件失败", e);
        }
    }

    /**
     * 清除用户相关缓存
     */
    private void clearUserRelatedCache(Long userId) {
        try {
            String pattern = MONGO_EVENT_CACHE_PREFIX + "user_*:" + userId;
            Set<String> keys = redisTemplate.keys(pattern);

            if (keys != null && !keys.isEmpty()) {
                redisTemplate.delete(keys);
            }

            log.info("清除MongoDB用户相关缓存成功: {}", userId);

        } catch (Exception e) {
            log.error("清除MongoDB用户相关缓存失败", e);
        }
    }

    /**
     * 清除订单相关缓存
     */
    private void clearOrderRelatedCache(Long orderId) {
        try {
            String pattern = MONGO_EVENT_CACHE_PREFIX + "order_*:" + orderId;
            Set<String> keys = redisTemplate.keys(pattern);

            if (keys != null && !keys.isEmpty()) {
                redisTemplate.delete(keys);
            }

            log.info("清除MongoDB订单相关缓存成功: {}", orderId);

        } catch (Exception e) {
            log.error("清除MongoDB订单相关缓存失败", e);
        }
    }

    /**
     * 清除支付相关缓存
     */
    private void clearPaymentRelatedCache(Long paymentId) {
        try {
            String pattern = MONGO_EVENT_CACHE_PREFIX + "payment_*:" + paymentId;
            Set<String> keys = redisTemplate.keys(pattern);

            if (keys != null && !keys.isEmpty()) {
                redisTemplate.delete(keys);
            }

            log.info("清除MongoDB支付相关缓存成功: {}", paymentId);

        } catch (Exception e) {
            log.error("清除MongoDB支付相关缓存失败", e);
        }
    }

    /**
     * 更新用户聚合数据
     */
    private void updateUserAggregatedData(Long userId) {
        try {
            // 异步更新用户聚合数据
            CompletableFuture.runAsync(() -> {
                try {
                    mongoAggregationService.aggregateUserOrderInfo(userId);
                    log.info("更新MongoDB用户聚合数据成功: {}", userId);
                } catch (Exception e) {
                    log.error("更新MongoDB用户聚合数据失败: {}", userId, e);
                }
            });

        } catch (Exception e) {
            log.error("更新MongoDB用户聚合数据失败", e);
        }
    }

    /**
     * 更新订单聚合数据
     */
    private void updateOrderAggregatedData(Long orderId) {
        try {
            // 异步更新订单聚合数据
            CompletableFuture.runAsync(() -> {
                try {
                    crossDatabaseQueryService.crossDatabaseQueryOrderDetailInfo(orderId);
                    log.info("更新MongoDB订单聚合数据成功: {}", orderId);
                } catch (Exception e) {
                    log.error("更新MongoDB订单聚合数据失败: {}", orderId, e);
                }
            });

        } catch (Exception e) {
            log.error("更新MongoDB订单聚合数据失败", e);
        }
    }

    /**
     * 更新商品聚合数据
     */
    private void updateProductAggregatedData(Long productId) {
        try {
            // 异步更新商品聚合数据
            CompletableFuture.runAsync(() -> {
                try {
                    Date endTime = new Date();
                    Date startTime = new Date(endTime.getTime() - 30L * 24 * 60 * 60 * 1000); // 30天前
                    crossDatabaseQueryService.crossDatabaseQueryProductSalesStatistics(productId, startTime, endTime);
                    log.info("更新MongoDB商品聚合数据成功: {}", productId);
                } catch (Exception e) {
                    log.error("更新MongoDB商品聚合数据失败: {}", productId, e);
                }
            });

        } catch (Exception e) {
            log.error("更新MongoDB商品聚合数据失败", e);
        }
    }
}

/**
 * MongoDB微服务数据查询管理服务
 */
@Service
public class MongoMicroserviceDataQueryManagementService {

    @Autowired
    private MongoAggregationService mongoAggregationService;

    @Autowired
    private MongoCrossDatabaseQueryService crossDatabaseQueryService;

    @Autowired
    private MongoEventDrivenService mongoEventDrivenService;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private final String MONGO_QUERY_CACHE_PREFIX = "mongo_query_cache:";
    private final long MONGO_QUERY_CACHE_EXPIRE = 1800; // 30分钟

    /**
     * 执行MongoDB跨库查询
     */
    public MongoCrossDatabaseQueryResult executeMongoCrossDatabaseQuery(MongoCrossDatabaseQueryRequest request) {
        try {
            MongoCrossDatabaseQueryResult result = new MongoCrossDatabaseQueryResult();
            result.setRequestId(request.getRequestId());
            result.setStartTime(new Date());

            // 验证请求
            validateMongoQueryRequest(request);

            // 选择查询策略
            MongoQueryStrategy strategy = selectMongoQueryStrategy(request);
            result.setStrategy(strategy);

            // 执行查询
            switch (strategy) {
                case AGGREGATION_PIPELINE:
                    result = executeAggregationPipelineQuery(request);
                    break;
                case CROSS_DATABASE_QUERY:
                    result = executeCrossDatabaseQuery(request);
                    break;
                case EVENT_DRIVEN:
                    result = executeEventDrivenQuery(request);
                    break;
                case HYBRID:
                    result = executeHybridQuery(request);
                    break;
                default:
                    throw new IllegalArgumentException("不支持的MongoDB查询策略: " + strategy);
            }

            result.setStatus(MongoQueryStatus.SUCCESS);
            result.setEndTime(new Date());

            log.info("执行MongoDB跨库查询成功: 请求ID={}, 策略={}", request.getRequestId(), strategy);
            return result;

        } catch (Exception e) {
            log.error("执行MongoDB跨库查询失败", e);
            throw new MongoCrossDatabaseQueryException("执行MongoDB跨库查询失败", e);
        }
    }

    /**
     * 验证MongoDB查询请求
     */
    private void validateMongoQueryRequest(MongoCrossDatabaseQueryRequest request) {
        if (request.getRequestId() == null || request.getRequestId().trim().isEmpty()) {
            throw new IllegalArgumentException("请求ID不能为空");
        }

        if (request.getQueryType() == null) {
            throw new IllegalArgumentException("查询类型不能为空");
        }

        if (request.getParameters() == null) {
            request.setParameters(new HashMap<>());
        }
    }

    /**
     * 选择MongoDB查询策略
     */
    private MongoQueryStrategy selectMongoQueryStrategy(MongoCrossDatabaseQueryRequest request) {
        try {
            // 根据查询类型选择策略
            switch (request.getQueryType()) {
                case USER_ORDER_AGGREGATION:
                    return MongoQueryStrategy.AGGREGATION_PIPELINE;
                case USER_COMPLETE_INFO:
                    return MongoQueryStrategy.CROSS_DATABASE_QUERY;
                case ORDER_DETAIL_INFO:
                    return MongoQueryStrategy.CROSS_DATABASE_QUERY;
                case PRODUCT_SALES_STATISTICS:
                    return MongoQueryStrategy.CROSS_DATABASE_QUERY;
                case COMPLEX_AGGREGATION:
                    return MongoQueryStrategy.AGGREGATION_PIPELINE;
                case SEARCH_RESULTS:
                    return MongoQueryStrategy.HYBRID;
                default:
                    return MongoQueryStrategy.CROSS_DATABASE_QUERY;
            }

        } catch (Exception e) {
            log.error("选择MongoDB查询策略失败", e);
            return MongoQueryStrategy.CROSS_DATABASE_QUERY; // 默认策略
        }
    }

    /**
     * 执行聚合管道查询
     */
    private MongoCrossDatabaseQueryResult executeAggregationPipelineQuery(MongoCrossDatabaseQueryRequest request) {
        try {
            MongoCrossDatabaseQueryResult result = new MongoCrossDatabaseQueryResult();
            result.setRequestId(request.getRequestId());
            result.setStrategy(MongoQueryStrategy.AGGREGATION_PIPELINE);

            // 根据查询类型执行不同的聚合管道查询
            switch (request.getQueryType()) {
                case USER_ORDER_AGGREGATION:
                    Long userId = Long.valueOf(request.getParameters().get("userId").toString());
                    UserOrderAggregationResult userOrderAggregation = mongoAggregationService.aggregateUserOrderInfo(userId);
                    result.setData(userOrderAggregation);
                    break;

                case COMPLEX_AGGREGATION:
                    ComplexAggregationRequest complexRequest = (ComplexAggregationRequest) request.getParameters().get("complexRequest");
                    ComplexAggregationResult complexResult = mongoAggregationService.executeComplexAggregation(complexRequest);
                    result.setData(complexResult);
                    break;

                default:
                    throw new IllegalArgumentException("不支持的聚合管道查询类型: " + request.getQueryType());
            }

            return result;

        } catch (Exception e) {
            log.error("执行聚合管道查询失败", e);
            throw new MongoCrossDatabaseQueryException("执行聚合管道查询失败", e);
        }
    }

    /**
     * 执行跨库查询
     */
    private MongoCrossDatabaseQueryResult executeCrossDatabaseQuery(MongoCrossDatabaseQueryRequest request) {
        try {
            MongoCrossDatabaseQueryResult result = new MongoCrossDatabaseQueryResult();
            result.setRequestId(request.getRequestId());
            result.setStrategy(MongoQueryStrategy.CROSS_DATABASE_QUERY);

            // 根据查询类型执行不同的跨库查询
            switch (request.getQueryType()) {
                case USER_COMPLETE_INFO:
                    Long userId = Long.valueOf(request.getParameters().get("userId").toString());
                    UserCompleteInfo userCompleteInfo = crossDatabaseQueryService.crossDatabaseQueryUserCompleteInfo(userId);
                    result.setData(userCompleteInfo);
                    break;

                case ORDER_DETAIL_INFO:
                    Long orderId = Long.valueOf(request.getParameters().get("orderId").toString());
                    OrderDetailInfo orderDetailInfo = crossDatabaseQueryService.crossDatabaseQueryOrderDetailInfo(orderId);
                    result.setData(orderDetailInfo);
                    break;

                case PRODUCT_SALES_STATISTICS:
                    Long productId = Long.valueOf(request.getParameters().get("productId").toString());
                    Date startTime = (Date) request.getParameters().get("startTime");
                    Date endTime = (Date) request.getParameters().get("endTime");
                    ProductSalesStatistics statistics = crossDatabaseQueryService.crossDatabaseQueryProductSalesStatistics(
                            productId, startTime, endTime);
                    result.setData(statistics);
                    break;

                default:
                    throw new IllegalArgumentException("不支持的跨库查询类型: " + request.getQueryType());
            }

            return result;

        } catch (Exception e) {
            log.error("执行跨库查询失败", e);
            throw new MongoCrossDatabaseQueryException("执行跨库查询失败", e);
        }
    }

    /**
     * 执行事件驱动查询
     */
    private MongoCrossDatabaseQueryResult executeEventDrivenQuery(MongoCrossDatabaseQueryRequest request) {
        try {
            MongoCrossDatabaseQueryResult result = new MongoCrossDatabaseQueryResult();
            result.setRequestId(request.getRequestId());
            result.setStrategy(MongoQueryStrategy.EVENT_DRIVEN);

            // 发布查询事件
            MongoCrossDatabaseQueryEvent event = new MongoCrossDatabaseQueryEvent();
            event.setEventId(UUID.randomUUID().toString());
            event.setEventType("MONGO_CROSS_DATABASE_QUERY");
            event.setRequestId(request.getRequestId());
            event.setQueryType(request.getQueryType());
            event.setParameters(request.getParameters());
            event.setTimestamp(new Date());

            mongoEventDrivenService.publishMongoCrossDatabaseQueryEvent(event);

            // 等待事件处理完成（实际应用中可能需要更复杂的等待机制）
            Thread.sleep(1000);

            result.setData("MongoDB事件已发布，请稍后查询结果");

            return result;

        } catch (Exception e) {
            log.error("执行事件驱动查询失败", e);
            throw new MongoCrossDatabaseQueryException("执行事件驱动查询失败", e);
        }
    }

    /**
     * 执行混合查询
     */
    private MongoCrossDatabaseQueryResult executeHybridQuery(MongoCrossDatabaseQueryRequest request) {
        try {
            MongoCrossDatabaseQueryResult result = new MongoCrossDatabaseQueryResult();
            result.setRequestId(request.getRequestId());
            result.setStrategy(MongoQueryStrategy.HYBRID);

            // 根据查询类型执行混合查询
            switch (request.getQueryType()) {
                case SEARCH_RESULTS:
                    MongoSearchRequest searchRequest = (MongoSearchRequest) request.getParameters().get("searchRequest");
                    MongoSearchResult searchResult = executeMongoSearch(searchRequest);
                    result.setData(searchResult);
                    break;

                default:
                    throw new IllegalArgumentException("不支持的混合查询类型: " + request.getQueryType());
            }

            return result;

        } catch (Exception e) {
            log.error("执行混合查询失败", e);
            throw new MongoCrossDatabaseQueryException("执行混合查询失败", e);
        }
    }

    /**
     * 执行MongoDB搜索
     */
    private MongoSearchResult executeMongoSearch(MongoSearchRequest request) {
        try {
            MongoSearchResult result = new MongoSearchResult();

            // 构建搜索查询
            Query query = new Query();

            if (request.getKeyword() != null && !request.getKeyword().trim().isEmpty()) {
                query.addCriteria(Criteria.where("name").regex(request.getKeyword(), "i")
                        .orOperator(Criteria.where("description").regex(request.getKeyword(), "i")));
            }

            if (request.getCategory() != null) {
                query.addCriteria(Criteria.where("category").is(request.getCategory()));
            }

            if (request.getPriceMin() != null) {
                query.addCriteria(Criteria.where("price").gte(request.getPriceMin()));
            }

            if (request.getPriceMax() != null) {
                query.addCriteria(Criteria.where("price").lte(request.getPriceMax()));
            }

            // 设置分页
            query.with(Sort.by(Sort.Direction.DESC, "createTime"));
            query.skip(request.getFrom());
            query.limit(request.getSize());

            // 执行搜索
            List<ProductInfo> products = productMongoTemplate.find(query, ProductInfo.class, "products");

            // 构建结果
            result.setProducts(products);
            result.setTotalCount(products.size());
            result.setFrom(request.getFrom());
            result.setSize(request.getSize());

            return result;

        } catch (Exception e) {
            log.error("执行MongoDB搜索失败", e);
            throw new MongoCrossDatabaseQueryException("执行MongoDB搜索失败", e);
        }
    }

    /**
     * 异步执行MongoDB跨库查询
     */
    @Async
    public CompletableFuture<MongoCrossDatabaseQueryResult> executeMongoCrossDatabaseQueryAsync(
            MongoCrossDatabaseQueryRequest request) {
        try {
            MongoCrossDatabaseQueryResult result = executeMongoCrossDatabaseQuery(request);
            return CompletableFuture.completedFuture(result);
        } catch (Exception e) {
            return CompletableFuture.failedFuture(e);
        }
    }

    /**
     * 获取MongoDB查询结果
     */
    public MongoCrossDatabaseQueryResult getMongoQueryResult(String requestId) {
        try {
            // 从缓存获取
            String cacheKey = MONGO_QUERY_CACHE_PREFIX + requestId;
            MongoCrossDatabaseQueryResult cachedResult = (MongoCrossDatabaseQueryResult) redisTemplate.opsForValue().get(cacheKey);

            if (cachedResult != null) {
                return cachedResult;
            }

            // 从数据库获取（实际应用中需要存储查询结果）
            throw new MongoCrossDatabaseQueryException("MongoDB查询结果不存在: " + requestId);

        } catch (Exception e) {
            log.error("获取MongoDB查询结果失败", e);
            throw new MongoCrossDatabaseQueryException("获取MongoDB查询结果失败", e);
        }
    }

    /**
     * 缓存MongoDB查询结果
     */
    public void cacheMongoQueryResult(String requestId, MongoCrossDatabaseQueryResult result) {
        try {
            String cacheKey = MONGO_QUERY_CACHE_PREFIX + requestId;
            redisTemplate.opsForValue().set(cacheKey, result, Duration.ofSeconds(MONGO_QUERY_CACHE_EXPIRE));

        } catch (Exception e) {
            log.error("缓存MongoDB查询结果失败", e);
        }
    }

    /**
     * 获取MongoDB查询统计信息
     */
    public MongoQueryStatistics getMongoQueryStatistics(Date startTime, Date endTime) {
        try {
            MongoQueryStatistics statistics = new MongoQueryStatistics();
            statistics.setStartTime(startTime);
            statistics.setEndTime(endTime);

            // 统计查询次数
            statistics.setTotalQueries(1000); // 实际应用中需要从数据库统计

            // 统计查询策略使用情况
            Map<MongoQueryStrategy, Long> strategyCount = new HashMap<>();
            strategyCount.put(MongoQueryStrategy.AGGREGATION_PIPELINE, 400L);
            strategyCount.put(MongoQueryStrategy.CROSS_DATABASE_QUERY, 300L);
            strategyCount.put(MongoQueryStrategy.EVENT_DRIVEN, 100L);
            strategyCount.put(MongoQueryStrategy.HYBRID, 200L);
            statistics.setStrategyCount(strategyCount);

            // 统计查询类型使用情况
            Map<MongoQueryType, Long> typeCount = new HashMap<>();
            typeCount.put(MongoQueryType.USER_ORDER_AGGREGATION, 300L);
            typeCount.put(MongoQueryType.USER_COMPLETE_INFO, 250L);
            typeCount.put(MongoQueryType.ORDER_DETAIL_INFO, 200L);
            typeCount.put(MongoQueryType.PRODUCT_SALES_STATISTICS, 150L);
            typeCount.put(MongoQueryType.COMPLEX_AGGREGATION, 100L);
            statistics.setTypeCount(typeCount);

            // 统计平均响应时间
            statistics.setAverageResponseTime(300.0); // 300ms

            // 统计成功率
            statistics.setSuccessRate(0.98); // 98%

            return statistics;

        } catch (Exception e) {
            log.error("获取MongoDB查询统计信息失败", e);
            throw new MongoCrossDatabaseQueryException("获取MongoDB查询统计信息失败", e);
        }
    }
}