#Scatter-Gather 散聚模式

public class ElasticsearchScatterGather {
    private final RestHighLevelClient esClient;

    public SearchResponse searchProducts(String query, int page, int size) {
        // 1. 获取集群状态，确定分片分布
        ClusterHealthResponse health = esClient.cluster().health();
        int totalShards = health.getNumberOfDataNodes();

        // 2. 构建分散查询请求
        SearchRequest searchRequest = new SearchRequest("products_*");
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.multiMatchQuery(query, "name", "description"));
        sourceBuilder.from(page * size);
        sourceBuilder.size(size);
        searchRequest.source(sourceBuilder);

        // 3. 分散查询：广播到所有相关分片
        // ES 自动将请求分发到各分片并行执行
        CompletableFuture<SearchResponse> future = new CompletableFuture<>();

        esClient.searchAsync(searchRequest, RequestOptions.DEFAULT,
            new ActionListener<SearchResponse>() {
                @Override
                public void onResponse(SearchResponse response) {
                    future.complete(response);
                }

                @Override
                public void onFailure(Exception e) {
                    future.completeExceptionally(e);
                }
            });

        try {
            // 4. 等待最慢的分片响应（默认超时 30s）
            SearchResponse response = future.get(30, TimeUnit.SECONDS);

            // 5. ES 自动在协调节点聚合各分片结果
            // 包括全局相关性评分、去重、分页等
            return response;
        } catch (Exception e) {
            // 处理超时或部分失败
            throw new SearchException("搜索失败", e);
        }
    }
}

public class ScatterGatherExecutor {
    private final ExecutorService executor;
    private final Duration timeout;

    public <T, R> ScatterGatherResult<R> execute(
            List<T> inputs,
            Function<T, R> callable,
            BiFunction<List<R>, List<Failure>, R> aggregator) {

        // 为每个输入创建异步任务
        List<CompletableFuture<R>> futures = inputs.stream()
            .map(input -> CompletableFuture
                .supplyAsync(() -> callable.apply(input), executor)
                .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS)
                .exceptionally(ex -> {
                    // 记录失败，但不影响其他任务
                    log.warn("Scatter 任务失败: input={}, error={}",
                        input, ex.getMessage());
                    return null;
                }))
            .collect(Collectors.toList());

        // 等待所有任务完成（或超时）
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

        // 收集结果和失败信息
        List<R> results = new ArrayList<>();
        List<Failure> failures = new ArrayList<>();

        for (int i = 0; i < inputs.size(); i++) {
            R result = futures.get(i).getNow(null);
            if (result != null) {
                results.add(result);
            } else {
                failures.add(new Failure(inputs.get(i), "超时或异常"));
            }
        }

        // 使用聚合器合并结果
        R aggregated = aggregator.apply(results, failures);

        return ScatterGatherResult.<R>builder()
            .result(aggregated)
            .totalCount(inputs.size())
            .successCount(results.size())
            .failureCount(failures.size())
            .failures(failures)
            .build();
    }
}

public class PriceAggregator {
    public List<PriceResult> aggregatePrices(List<PlatformPrice> prices, List<Failure> failures) {
        // 记录失败但继续处理
        if (!failures.isEmpty()) {
            log.warn("部分平台查询失败: {}", failures);
        }

        // 按价格排序
        List<PriceResult> results = prices.stream()
            .map(p -> PriceResult.builder()
                .platform(p.getPlatform())
                .price(p.getPrice())
                .url(p.getProductUrl())
                .stockStatus(p.getStockStatus())
                .build())
            .sorted(Comparator.comparing(PriceResult::getPrice))
            .collect(Collectors.toList());

        // 标记不可用的平台
        failures.forEach(f -> {
            results.add(PriceResult.builder()
                .platform("平台-" + f.getInput())
                .price(null)
                .status("查询失败: " + f.getReason())
                .build());
        });

        return results;
    }
}

public class PriceComparisonService {
    private final ScatterGatherExecutor executor;

    public List<PriceResult> comparePrices(String productId) {
        List<String> platforms = List.of("TMALL", "JD", "PDD", "DOUYIN");

        return executor.execute(
            platforms,
            platform -> queryPlatformPrice(platform, productId),
            this::aggregatePrices
        ).getResult();
    }

    private PlatformPrice queryPlatformPrice(String platform, String productId) {
        // 模拟平台 API 调用
        return switch (platform) {
            case "TMALL" -> tmallClient.getPrice(productId);
            case "JD" -> jdClient.getPrice(productId);
            case "PDD" -> pddClient.getPrice(productId);
            case "DOUYIN" -> douyinClient.getPrice(productId);
            default -> throw new UnsupportedOperationException();
        };
    }
}

public class LoadTestScatterGather {
    public LoadTestReport runLoadTest(LoadTestConfig config) {
        // 生成测试用例
        List<EndpointTestCase> testCases = generateTestCases(config);

        // 并行发送到所有压测节点
        List<CompletableFuture<NodeReport>> futures = testCases.stream()
            .map(this::executeOnNodes)
            .flatMap(List::stream)
            .collect(Collectors.toList());

        // 收集所有节点报告
        List<NodeReport> reports = futures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());

        // 聚合报告
        return aggregateReports(reports);
    }
}