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

/**
 * ES多索引查询服务
 */
@Service
public class EsMultiIndexQueryService {

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private final String ES_QUERY_CACHE_PREFIX = "es_query_cache:";
    private final long ES_QUERY_CACHE_EXPIRE = 1800; // 30分钟

    /**
     * 跨索引查询用户订单信息
     */
    public UserOrderSearchResult searchUserOrderInfo(Long userId) {
        try {
            // 从缓存获取
            String cacheKey = ES_QUERY_CACHE_PREFIX + "user_order:" + userId;
            UserOrderSearchResult cachedResult = (UserOrderSearchResult) redisTemplate.opsForValue().get(cacheKey);

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

            // 1. 构建多索引查询
            SearchRequest searchRequest = buildMultiIndexSearchRequest(userId);

            // 2. 执行搜索
            SearchResponse response = elasticsearchTemplate.search(searchRequest, RequestOptions.DEFAULT);

            // 3. 处理结果
            UserOrderSearchResult result = processSearchResponse(response);

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

            return result;

        } catch (Exception e) {
            log.error("跨索引查询用户订单信息失败", e);
            throw new EsCrossDatabaseQueryException("跨索引查询用户订单信息失败", e);
        }
    }

    /**
     * 构建多索引搜索请求
     */
    private SearchRequest buildMultiIndexSearchRequest(Long userId) {
        try {
            SearchRequest searchRequest = new SearchRequest("users", "orders", "products", "payments");

            // 构建查询
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            boolQuery.must(QueryBuilders.termQuery("userId", userId));

            // 构建搜索源
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(boolQuery);
            searchSourceBuilder.size(100);

            // 添加聚合
            searchSourceBuilder.aggregation(AggregationBuilders.terms("order_status_agg")
                    .field("status")
                    .size(10));

            searchSourceBuilder.aggregation(AggregationBuilders.stats("amount_stats")
                    .field("amount"));

            searchRequest.source(searchSourceBuilder);

            return searchRequest;

        } catch (Exception e) {
            log.error("构建多索引搜索请求失败", e);
            throw new EsCrossDatabaseQueryException("构建多索引搜索请求失败", e);
        }
    }

    /**
     * 处理搜索结果
     */
    private UserOrderSearchResult processSearchResponse(SearchResponse response) {
        try {
            UserOrderSearchResult result = new UserOrderSearchResult();
            result.setTotalHits(response.getHits().getTotalHits().value);
            result.setTook(response.getTook().millis());

            // 处理搜索结果
            List<SearchHit> hits = Arrays.asList(response.getHits().getHits());
            List<Object> documents = hits.stream()
                    .map(hit -> hit.getSourceAsMap())
                    .collect(Collectors.toList());

            result.setDocuments(documents);

            // 处理聚合结果
            Map<String, Object> aggregations = new HashMap<>();
            response.getAggregations().asMap().forEach((name, agg) -> {
                aggregations.put(name, agg);
            });
            result.setAggregations(aggregations);

            return result;

        } catch (Exception e) {
            log.error("处理搜索结果失败", e);
            throw new EsCrossDatabaseQueryException("处理搜索结果失败", e);
        }
    }

    /**
     * 并行查询多个索引
     */
    public CompletableFuture<MultiIndexSearchResult> searchMultipleIndicesAsync(
            MultiIndexSearchRequest request) {
        try {
            CompletableFuture<SearchResponse> usersFuture = CompletableFuture.supplyAsync(() -> {
                return searchUsers(request.getUserId());
            });

            CompletableFuture<SearchResponse> ordersFuture = CompletableFuture.supplyAsync(() -> {
                return searchOrders(request.getUserId());
            });

            CompletableFuture<SearchResponse> productsFuture = CompletableFuture.supplyAsync(() -> {
                return searchProducts(request.getUserId());
            });

            CompletableFuture<SearchResponse> paymentsFuture = CompletableFuture.supplyAsync(() -> {
                return searchPayments(request.getUserId());
            });

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

            return allFutures.thenApply(v -> {
                MultiIndexSearchResult result = new MultiIndexSearchResult();
                try {
                    result.setUsersResponse(usersFuture.get());
                    result.setOrdersResponse(ordersFuture.get());
                    result.setProductsResponse(productsFuture.get());
                    result.setPaymentsResponse(paymentsFuture.get());
                    result.setStatus(SearchStatus.SUCCESS);
                } catch (Exception e) {
                    result.setStatus(SearchStatus.FAILED);
                    result.setErrorMessage(e.getMessage());
                }
                return result;
            });

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

    /**
     * 搜索用户
     */
    private SearchResponse searchUsers(Long userId) {
        try {
            SearchRequest searchRequest = new SearchRequest("users");
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(QueryBuilders.termQuery("_id", userId));
            searchRequest.source(searchSourceBuilder);

            return elasticsearchTemplate.search(searchRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error("搜索用户失败", e);
            throw new EsCrossDatabaseQueryException("搜索用户失败", e);
        }
    }

    /**
     * 搜索订单
     */
    private SearchResponse searchOrders(Long userId) {
        try {
            SearchRequest searchRequest = new SearchRequest("orders");
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(QueryBuilders.termQuery("userId", userId));
            searchRequest.source(searchSourceBuilder);

            return elasticsearchTemplate.search(searchRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error("搜索订单失败", e);
            throw new EsCrossDatabaseQueryException("搜索订单失败", e);
        }
    }

    /**
     * 搜索商品
     */
    private SearchResponse searchProducts(Long userId) {
        try {
            SearchRequest searchRequest = new SearchRequest("products");
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(QueryBuilders.termQuery("userId", userId));
            searchRequest.source(searchSourceBuilder);

            return elasticsearchTemplate.search(searchRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error("搜索商品失败", e);
            throw new EsCrossDatabaseQueryException("搜索商品失败", e);
        }
    }

    /**
     * 搜索支付
     */
    private SearchResponse searchPayments(Long userId) {
        try {
            SearchRequest searchRequest = new SearchRequest("payments");
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(QueryBuilders.termQuery("userId", userId));
            searchRequest.source(searchSourceBuilder);

            return elasticsearchTemplate.search(searchRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error("搜索支付失败", e);
            throw new EsCrossDatabaseQueryException("搜索支付失败", e);
        }
    }
}

/**
 * ES聚合查询服务
 */
@Service
public class EsAggregationQueryService {

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private final String ES_AGGREGATION_CACHE_PREFIX = "es_aggregation_cache:";
    private final long ES_AGGREGATION_CACHE_EXPIRE = 1800; // 30分钟

    /**
     * 聚合查询用户订单统计
     */
    public UserOrderAggregationResult aggregateUserOrderStatistics(Long userId) {
        try {
            // 从缓存获取
            String cacheKey = ES_AGGREGATION_CACHE_PREFIX + "user_order_stats:" + userId;
            UserOrderAggregationResult cachedResult = (UserOrderAggregationResult) redisTemplate.opsForValue().get(cacheKey);

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

            // 构建聚合查询
            SearchRequest searchRequest = buildAggregationSearchRequest(userId);

            // 执行搜索
            SearchResponse response = elasticsearchTemplate.search(searchRequest, RequestOptions.DEFAULT);

            // 处理聚合结果
            UserOrderAggregationResult result = processAggregationResponse(response, userId);

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

            return result;

        } catch (Exception e) {
            log.error("聚合查询用户订单统计失败", e);
            throw new EsCrossDatabaseQueryException("聚合查询用户订单统计失败", e);
        }
    }

    /**
     * 构建聚合搜索请求
     */
    private SearchRequest buildAggregationSearchRequest(Long userId) {
        try {
            SearchRequest searchRequest = new SearchRequest("orders");

            // 构建查询
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            boolQuery.must(QueryBuilders.termQuery("userId", userId));

            // 构建搜索源
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(boolQuery);
            searchSourceBuilder.size(0); // 只返回聚合结果

            // 添加聚合
            searchSourceBuilder.aggregation(AggregationBuilders.terms("status_agg")
                    .field("status")
                    .size(10));

            searchSourceBuilder.aggregation(AggregationBuilders.stats("amount_stats")
                    .field("amount"));

            searchSourceBuilder.aggregation(AggregationBuilders.dateHistogram("time_agg")
                    .field("createTime")
                    .calendarInterval(DateHistogramInterval.DAY));

            searchRequest.source(searchSourceBuilder);

            return searchRequest;

        } catch (Exception e) {
            log.error("构建聚合搜索请求失败", e);
            throw new EsCrossDatabaseQueryException("构建聚合搜索请求失败", e);
        }
    }

    /**
     * 处理聚合响应
     */
    private UserOrderAggregationResult processAggregationResponse(SearchResponse response, Long userId) {
        try {
            UserOrderAggregationResult result = new UserOrderAggregationResult();
            result.setUserId(userId);
            result.setTotalHits(response.getHits().getTotalHits().value);
            result.setTook(response.getTook().millis());

            // 处理状态聚合
            Terms statusAgg = response.getAggregations().get("status_agg");
            Map<String, Long> statusCount = new HashMap<>();
            for (Terms.Bucket bucket : statusAgg.getBuckets()) {
                statusCount.put(bucket.getKeyAsString(), bucket.getDocCount());
            }
            result.setStatusCount(statusCount);

            // 处理金额统计
            Stats amountStats = response.getAggregations().get("amount_stats");
            result.setTotalAmount(amountStats.getSum());
            result.setAvgAmount(amountStats.getAvg());
            result.setMinAmount(amountStats.getMin());
            result.setMaxAmount(amountStats.getMax());

            // 处理时间聚合
            Histogram timeAgg = response.getAggregations().get("time_agg");
            Map<String, Long> timeCount = new HashMap<>();
            for (Histogram.Bucket bucket : timeAgg.getBuckets()) {
                timeCount.put(bucket.getKeyAsString(), bucket.getDocCount());
            }
            result.setTimeCount(timeCount);

            return result;

        } catch (Exception e) {
            log.error("处理聚合响应失败", e);
            throw new EsCrossDatabaseQueryException("处理聚合响应失败", e);
        }
    }

    /**
     * 复杂聚合查询
     */
    public ComplexAggregationResult executeComplexAggregation(ComplexAggregationRequest request) {
        try {
            SearchRequest searchRequest = new SearchRequest(request.getIndices());

            // 构建查询
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(request.getQuery());
            searchSourceBuilder.size(0);

            // 添加聚合
            for (AggregationBuilder aggregation : request.getAggregations()) {
                searchSourceBuilder.aggregation(aggregation);
            }

            searchRequest.source(searchSourceBuilder);

            // 执行搜索
            SearchResponse response = elasticsearchTemplate.search(searchRequest, RequestOptions.DEFAULT);

            // 处理结果
            ComplexAggregationResult result = new ComplexAggregationResult();
            result.setTotalHits(response.getHits().getTotalHits().value);
            result.setTook(response.getTook().millis());

            // 处理聚合结果
            Map<String, Object> aggregations = new HashMap<>();
            response.getAggregations().asMap().forEach((name, agg) -> {
                aggregations.put(name, agg);
            });
            result.setAggregations(aggregations);

            return result;

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

/**
 * ES事件驱动服务
 */
@Service
public class EsEventDrivenService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private EsAggregationQueryService esAggregationQueryService;

    private final String ES_EVENT_QUEUE_PREFIX = "es_cross_db_query.event.";
    private final String ES_EVENT_CACHE_PREFIX = "es_event_cache:";

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

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

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

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

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

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

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

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

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

            publishEsCrossDatabaseQueryEvent(dataUpdatedEvent);

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

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

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

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

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

            publishEsCrossDatabaseQueryEvent(dataUpdatedEvent);

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

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

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

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

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

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

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

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

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

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

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

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

    @Autowired
    private EsMultiIndexQueryService multiIndexQueryService;

    @Autowired
    private EsAggregationQueryService aggregationQueryService;

    @Autowired
    private EsEventDrivenService eventDrivenService;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private final String ES_QUERY_CACHE_PREFIX = "es_query_cache:";
    private final long ES_QUERY_CACHE_EXPIRE = 1800; // 30分钟

    /**
     * 执行ES跨库查询
     */
    public EsCrossDatabaseQueryResult executeEsCrossDatabaseQuery(EsCrossDatabaseQueryRequest request) {
        try {
            EsCrossDatabaseQueryResult result = new EsCrossDatabaseQueryResult();
            result.setRequestId(request.getRequestId());
            result.setStartTime(new Date());

            // 验证请求
            validateEsQueryRequest(request);

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

            // 执行查询
            switch (strategy) {
                case MULTI_INDEX_QUERY:
                    result = executeMultiIndexQuery(request);
                    break;
                case AGGREGATION_QUERY:
                    result = executeAggregationQuery(request);
                    break;
                case EVENT_DRIVEN:
                    result = executeEventDrivenQuery(request);
                    break;
                case HYBRID:
                    result = executeHybridQuery(request);
                    break;
                default:
                    throw new IllegalArgumentException("不支持的ES查询策略: " + strategy);
            }

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

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

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

    /**
     * 验证ES查询请求
     */
    private void validateEsQueryRequest(EsCrossDatabaseQueryRequest 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<>());
        }
    }

    /**
     * 选择ES查询策略
     */
    private EsQueryStrategy selectEsQueryStrategy(EsCrossDatabaseQueryRequest request) {
        try {
            // 根据查询类型选择策略
            switch (request.getQueryType()) {
                case USER_ORDER_SEARCH:
                    return EsQueryStrategy.MULTI_INDEX_QUERY;
                case USER_ORDER_AGGREGATION:
                    return EsQueryStrategy.AGGREGATION_QUERY;
                case COMPLEX_AGGREGATION:
                    return EsQueryStrategy.AGGREGATION_QUERY;
                case SEARCH_RESULTS:
                    return EsQueryStrategy.HYBRID;
                default:
                    return EsQueryStrategy.MULTI_INDEX_QUERY;
            }

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

    /**
     * 执行多索引查询
     */
    private EsCrossDatabaseQueryResult executeMultiIndexQuery(EsCrossDatabaseQueryRequest request) {
        try {
            EsCrossDatabaseQueryResult result = new EsCrossDatabaseQueryResult();
            result.setRequestId(request.getRequestId());
            result.setStrategy(EsQueryStrategy.MULTI_INDEX_QUERY);

            // 根据查询类型执行不同的多索引查询
            switch (request.getQueryType()) {
                case USER_ORDER_SEARCH:
                    Long userId = Long.valueOf(request.getParameters().get("userId").toString());
                    UserOrderSearchResult userOrderSearch = multiIndexQueryService.searchUserOrderInfo(userId);
                    result.setData(userOrderSearch);
                    break;

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

            return result;

        } catch (Exception e) {
            log.error("执行多索引查询失败", e);
            throw new EsCrossDatabaseQueryException("执行多索引查询失败", e);
        }
    }

    /**
     * 执行聚合查询
     */
    private EsCrossDatabaseQueryResult executeAggregationQuery(EsCrossDatabaseQueryRequest request) {
        try {
            EsCrossDatabaseQueryResult result = new EsCrossDatabaseQueryResult();
            result.setRequestId(request.getRequestId());
            result.setStrategy(EsQueryStrategy.AGGREGATION_QUERY);

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

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

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

            return result;

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

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

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

            eventDrivenService.publishEsCrossDatabaseQueryEvent(event);

            // 等待事件处理完成
            Thread.sleep(1000);

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

            return result;

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

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

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

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

            return result;

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

    /**
     * 执行ES搜索
     */
    private EsSearchResult executeEsSearch(EsSearchRequest request) {
        try {
            EsSearchResult result = new EsSearchResult();

            // 构建搜索查询
            SearchRequest searchRequest = new SearchRequest(request.getIndices());
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            if (request.getKeyword() != null && !request.getKeyword().trim().isEmpty()) {
                searchSourceBuilder.query(QueryBuilders.multiMatchQuery(request.getKeyword(), 
                        "name", "description", "tags"));
            } else {
                searchSourceBuilder.query(QueryBuilders.matchAllQuery());
            }

            // 设置分页
            searchSourceBuilder.from(request.getFrom());
            searchSourceBuilder.size(request.getSize());

            searchRequest.source(searchSourceBuilder);

            // 执行搜索
            SearchResponse response = elasticsearchTemplate.search(searchRequest, RequestOptions.DEFAULT);

            // 构建结果
            result.setTotalHits(response.getHits().getTotalHits().value);
            result.setTook(response.getTook().millis());

            // 处理搜索结果
            List<SearchHit> hits = Arrays.asList(response.getHits().getHits());
            List<Object> documents = hits.stream()
                    .map(hit -> hit.getSourceAsMap())
                    .collect(Collectors.toList());

            result.setDocuments(documents);

            return result;

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

    /**
     * 异步执行ES跨库查询
     */
    @Async
    public CompletableFuture<EsCrossDatabaseQueryResult> executeEsCrossDatabaseQueryAsync(
            EsCrossDatabaseQueryRequest request) {
        try {
            EsCrossDatabaseQueryResult result = executeEsCrossDatabaseQuery(request);
            return CompletableFuture.completedFuture(result);
        } catch (Exception e) {
            return CompletableFuture.failedFuture(e);
        }
    }

    /**
     * 获取ES查询结果
     */
    public EsCrossDatabaseQueryResult getEsQueryResult(String requestId) {
        try {
            // 从缓存获取
            String cacheKey = ES_QUERY_CACHE_PREFIX + requestId;
            EsCrossDatabaseQueryResult cachedResult = (EsCrossDatabaseQueryResult) redisTemplate.opsForValue().get(cacheKey);

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

            // 从数据库获取
            throw new EsCrossDatabaseQueryException("ES查询结果不存在: " + requestId);

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

    /**
     * 缓存ES查询结果
     */
    public void cacheEsQueryResult(String requestId, EsCrossDatabaseQueryResult result) {
        try {
            String cacheKey = ES_QUERY_CACHE_PREFIX + requestId;
            redisTemplate.opsForValue().set(cacheKey, result, Duration.ofSeconds(ES_QUERY_CACHE_EXPIRE));

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

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

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

            // 统计查询策略使用情况
            Map<EsQueryStrategy, Long> strategyCount = new HashMap<>();
            strategyCount.put(EsQueryStrategy.MULTI_INDEX_QUERY, 300L);
            strategyCount.put(EsQueryStrategy.AGGREGATION_QUERY, 250L);
            strategyCount.put(EsQueryStrategy.EVENT_DRIVEN, 100L);
            strategyCount.put(EsQueryStrategy.HYBRID, 150L);
            statistics.setStrategyCount(strategyCount);

            // 统计查询类型使用情况
            Map<EsQueryType, Long> typeCount = new HashMap<>();
            typeCount.put(EsQueryType.USER_ORDER_SEARCH, 200L);
            typeCount.put(EsQueryType.USER_ORDER_AGGREGATION, 150L);
            typeCount.put(EsQueryType.COMPLEX_AGGREGATION, 100L);
            typeCount.put(EsQueryType.SEARCH_RESULTS, 350L);
            statistics.setTypeCount(typeCount);

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

            // 统计成功率
            statistics.setSuccessRate(0.99); // 99%

            return statistics;

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

/**
 * ES跨库查询性能优化服务
 */
@Service
public class EsCrossDatabaseQueryPerformanceOptimizationService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private CaffeineCache localCache;

    private final String ES_PERFORMANCE_CACHE_PREFIX = "es_performance_cache:";

    /**
     * 缓存ES查询性能数据
     */
    public void cacheEsQueryPerformanceData(String requestId, Object data) {
        String cacheKey = ES_PERFORMANCE_CACHE_PREFIX + requestId;

        try {
            // 写入本地缓存
            localCache.put(cacheKey, data);

            // 写入Redis缓存
            String redisCacheKey = "redis_cache:" + cacheKey;
            redisTemplate.opsForValue().set(redisCacheKey, data, Duration.ofHours(1));

        } catch (Exception e) {
            log.error("缓存ES查询性能数据失败", e);
        }
    }

    /**
     * 获取缓存的ES查询性能数据
     */
    public Object getCachedEsQueryPerformanceData(String requestId) {
        String cacheKey = ES_PERFORMANCE_CACHE_PREFIX + requestId;

        try {
            // 从本地缓存获取
            Object cachedData = localCache.getIfPresent(cacheKey);
            if (cachedData != null) {
                return cachedData;
            }

            // 从Redis获取
            String redisCacheKey = "redis_cache:" + cacheKey;
            Object redisData = redisTemplate.opsForValue().get(redisCacheKey);
            if (redisData != null) {
                localCache.put(cacheKey, redisData);
                return redisData;
            }

        } catch (Exception e) {
            log.error("获取缓存的ES查询性能数据失败", e);
        }

        return null;
    }

    /**
     * 批量处理ES查询请求
     */
    public List<EsCrossDatabaseQueryResult> batchProcessEsQueryRequests(
            List<EsCrossDatabaseQueryRequest> requests) {
        List<EsCrossDatabaseQueryResult> results = new ArrayList<>();

        try {
            // 按查询类型分组
            Map<EsQueryType, List<EsCrossDatabaseQueryRequest>> typeGroups = requests.stream()
                    .collect(Collectors.groupingBy(EsCrossDatabaseQueryRequest::getQueryType));

            // 并行处理各类型
            typeGroups.entrySet().parallelStream().forEach(entry -> {
                EsQueryType queryType = entry.getKey();
                List<EsCrossDatabaseQueryRequest> typeRequests = entry.getValue();

                try {
                    List<EsCrossDatabaseQueryResult> typeResults = processEsTypeRequests(queryType, typeRequests);

                    synchronized (results) {
                        results.addAll(typeResults);
                    }

                } catch (Exception e) {
                    log.error("处理ES查询类型失败: {}", queryType, e);
                }
            });

        } catch (Exception e) {
            log.error("批量处理ES查询请求失败", e);
        }

        return results;
    }

    /**
     * 处理ES类型请求
     */
    private List<EsCrossDatabaseQueryResult> processEsTypeRequests(EsQueryType queryType, 
            List<EsCrossDatabaseQueryRequest> requests) {
        List<EsCrossDatabaseQueryResult> results = new ArrayList<>();

        for (EsCrossDatabaseQueryRequest request : requests) {
            try {
                EsCrossDatabaseQueryResult result = processEsQueryRequest(request);
                results.add(result);
            } catch (Exception e) {
                log.error("处理ES查询请求失败: {}", request.getRequestId(), e);
                EsCrossDatabaseQueryResult errorResult = new EsCrossDatabaseQueryResult();
                errorResult.setRequestId(request.getRequestId());
                errorResult.setStatus(EsQueryStatus.FAILED);
                errorResult.setErrorMessage(e.getMessage());
                results.add(errorResult);
            }
        }

        return results;
    }

    /**
     * 处理ES查询请求
     */
    private EsCrossDatabaseQueryResult processEsQueryRequest(EsCrossDatabaseQueryRequest request) {
        // 实现ES查询处理逻辑
        EsCrossDatabaseQueryResult result = new EsCrossDatabaseQueryResult();
        result.setRequestId(request.getRequestId());
        result.setStatus(EsQueryStatus.SUCCESS);
        result.setData("ES查询成功");
        return result;
    }

    /**
     * 预热ES查询性能缓存
     */
    @PostConstruct
    public void warmupEsQueryPerformanceCache() {
        try {
            // 预热常用ES查询性能数据
            List<String> commonRequestIds = Arrays.asList("es_req_1", "es_req_2", "es_req_3");

            for (String requestId : commonRequestIds) {
                try {
                    String cacheKey = ES_PERFORMANCE_CACHE_PREFIX + requestId;
                    Object performanceData = new Object();
                    cacheEsQueryPerformanceData(requestId, performanceData);
                } catch (Exception e) {
                    log.error("预热ES查询性能缓存失败: {}", requestId, e);
                }
            }

        } catch (Exception e) {
            log.error("预热ES查询性能缓存失败", e);
        }
    }

    /**
     * 清理过期缓存
     */
    @Scheduled(fixedRate = 300000) // 5分钟
    public void cleanupExpiredCache() {
        try {
            // 清理本地缓存
            localCache.cleanUp();

            // 清理Redis过期缓存
            cleanupRedisExpiredCache();

        } catch (Exception e) {
            log.error("清理过期缓存失败", e);
        }
    }

    /**
     * 清理Redis过期缓存
     */
    private void cleanupRedisExpiredCache() {
        try {
            Set<String> cacheKeys = redisTemplate.keys("redis_cache:" + ES_PERFORMANCE_CACHE_PREFIX + "*");

            for (String key : cacheKeys) {
                Long ttl = redisTemplate.getExpire(key);
                if (ttl != null && ttl <= 0) {
                    redisTemplate.delete(key);
                }
            }

        } catch (Exception e) {
            log.error("清理Redis过期缓存失败", e);
        }
    }
}

/**
 * ES跨库查询监控指标
 */
@Component
public class EsCrossDatabaseQueryMetrics {

    private final MeterRegistry meterRegistry;

    public EsCrossDatabaseQueryMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    /**
     * 记录ES跨库查询次数
     */
    public void recordEsCrossDatabaseQueryCount(String queryType, String strategy) {
        Counter.builder("es_cross_db_query.count")
                .description("ES跨库查询次数")
                .tag("query_type", queryType)
                .tag("strategy", strategy)
                .register(meterRegistry)
                .increment();
    }

    /**
     * 记录ES跨库查询时间
     */
    public void recordEsCrossDatabaseQueryTime(String queryType, String strategy, long duration) {
        Timer.builder("es_cross_db_query.time")
                .description("ES跨库查询时间")
                .tag("query_type", queryType)
                .tag("strategy", strategy)
                .register(meterRegistry)
                .record(duration, TimeUnit.MILLISECONDS);
    }

    /**
     * 记录ES跨库查询数据量
     */
    public void recordEsCrossDatabaseQueryDataCount(String queryType, long count) {
        Counter.builder("es_cross_db_query.data.count")
                .description("ES跨库查询数据量")
                .tag("query_type", queryType)
                .register(meterRegistry)
                .increment(count);
    }

    /**
     * 记录ES跨库查询成功率
     */
    public void recordEsCrossDatabaseQuerySuccessRate(String queryType, double successRate) {
        Gauge.builder("es_cross_db_query.success.rate")
                .description("ES跨库查询成功率")
                .tag("query_type", queryType)
                .register(meterRegistry, successRate);
    }

    /**
     * 记录ES跨库查询失败率
     */
    public void recordEsCrossDatabaseQueryFailureRate(String queryType, double failureRate) {
        Gauge.builder("es_cross_db_query.failure.rate")
                .description("ES跨库查询失败率")
                .tag("query_type", queryType)
                .register(meterRegistry, failureRate);
    }

    /**
     * 记录ES跨库查询吞吐量
     */
    public void recordEsCrossDatabaseQueryThroughput(String queryType, double throughput) {
        Gauge.builder("es_cross_db_query.throughput")
                .description("ES跨库查询吞吐量")
                .tag("query_type", queryType)
                .register(meterRegistry, throughput);
    }

    /**
     * 记录ES跨库查询异常次数
     */
    public void recordEsCrossDatabaseQueryExceptionCount(String queryType, String exceptionType) {
        Counter.builder("es_cross_db_query.exception.count")
                .description("ES跨库查询异常次数")
                .tag("query_type", queryType)
                .tag("exception_type", exceptionType)
                .register(meterRegistry)
                .increment();
    }
}