第232集并行查询接口架构实战：高并发数据聚合、异步查询、结果合并的企业级解决方案

前言

在当今大数据、高并发的企业级应用中，并行查询已成为提升系统性能和用户体验的关键技术。传统的串行查询方式往往响应缓慢、资源利用率低，无法满足复杂业务场景的查询需求。基于并行查询的接口设计，不仅能够充分利用多核CPU资源，还能实现数据的并行处理、异步查询和智能结果合并。随着微服务架构和分布式系统的普及，构建可扩展、高性能的并行查询框架，已成为企业级架构师必须掌握的核心技能。

本文将深入探讨并行查询接口的架构设计与实战应用，从高并发数据聚合到异步查询处理，从结果合并到性能优化，为企业构建高效、稳定的并行查询解决方案提供全面的技术指导。

一、并行查询接口架构概述与核心原理

1.1 并行查询接口架构设计

并行查询系统采用分层架构设计，通过任务分解、并行执行、结果聚合等技术，实现高效的并行查询处理能力。

graph TB
    A[查询请求] --> B[查询解析器]
    B --> C[任务分解器]
    C --> D[并行执行器]
    D --> E[结果收集器]
    E --> F[结果合并器]
    F --> G[响应返回]
    
    H[并行策略] --> I[数据分片]
    H --> J[任务并行]
    H --> K[流水线并行]
    H --> L[混合并行]
    
    M[执行引擎] --> N[线程池]
    M --> O[ForkJoinPool]
    M --> P[CompletableFuture]
    M --> Q[异步处理]
    
    R[结果处理] --> S[数据聚合]
    R --> T[排序合并]
    R --> U[去重处理]
    R --> V[分页处理]
    
    W[监控体系] --> X[性能监控]
    W --> Y[资源监控]
    W --> Z[异常监控]
    W --> AA[结果监控]

1.2 并行查询核心特性

1.2.1 任务分解策略

数据分片：将大数据集分解为多个小数据集
查询分解：将复杂查询分解为多个简单查询
依赖分析：分析查询间的依赖关系
负载均衡：智能分配查询任务到不同线程

1.2.2 并行执行机制

线程池管理：使用线程池管理并行任务
异步处理：基于CompletableFuture的异步处理
工作窃取：使用ForkJoinPool实现工作窃取
资源控制：精确控制并行度和资源使用

1.2.3 结果聚合处理

数据合并：智能合并多个查询结果
排序处理：高效处理结果排序
去重处理：处理重复数据
分页支持：支持结果分页处理

二、并行查询核心实现

2.1 并行查询引擎

// 并行查询引擎
@Service
@Slf4j
public class ParallelQueryEngine {
    
    @Autowired
    private QueryTaskExecutor taskExecutor;
    
    @Autowired
    private ResultAggregator resultAggregator;
    
    @Autowired
    private QueryMonitorService monitorService;
    
    private final ForkJoinPool forkJoinPool = new ForkJoinPool(
        Runtime.getRuntime().availableProcessors(),
        ForkJoinPool.defaultForkJoinWorkerThreadFactory,
        null,
        true
    );
    
    /**
     * 执行并行查询
     */
    public <T> CompletableFuture<QueryResult<T>> executeParallelQuery(
            ParallelQueryRequest<T> request) {
        
        long startTime = System.currentTimeMillis();
        String queryId = generateQueryId();
        
        try {
            // 1. 解析查询请求
            QueryContext context = parseQueryRequest(request);
            
            // 2. 分解查询任务
            List<QueryTask<T>> tasks = decomposeQueryTasks(context);
            
            // 3. 并行执行任务
            CompletableFuture<List<QueryResult<T>>> taskResults = executeTasksInParallel(tasks);
            
            // 4. 聚合结果
            CompletableFuture<QueryResult<T>> finalResult = taskResults.thenApply(results -> {
                try {
                    QueryResult<T> aggregatedResult = resultAggregator.aggregate(results, context);
                    
                    // 5. 记录监控信息
                    long executionTime = System.currentTimeMillis() - startTime;
                    monitorService.recordQueryExecution(queryId, executionTime, results.size());
                    
                    return aggregatedResult;
                    
                } catch (Exception e) {
                    log.error("结果聚合失败: queryId={}", queryId, e);
                    throw new RuntimeException("结果聚合失败", e);
                }
            });
            
            return finalResult;
            
        } catch (Exception e) {
            log.error("并行查询执行失败: queryId={}", queryId, e);
            return CompletableFuture.failedFuture(e);
        }
    }
    
    /**
     * 执行数据分片查询
     */
    public <T> CompletableFuture<QueryResult<T>> executeShardedQuery(
            ShardedQueryRequest<T> request) {
        
        long startTime = System.currentTimeMillis();
        String queryId = generateQueryId();
        
        try {
            // 1. 获取数据分片信息
            List<DataShard> shards = getDataShards(request.getDataSource(), request.getShardKey());
            
            // 2. 创建分片查询任务
            List<CompletableFuture<QueryResult<T>>> shardFutures = shards.stream()
                .map(shard -> CompletableFuture.supplyAsync(() -> {
                    try {
                        return executeShardQuery(shard, request);
                    } catch (Exception e) {
                        log.error("分片查询失败: shard={}", shard.getId(), e);
                        return QueryResult.empty();
                    }
                }, forkJoinPool))
                .collect(Collectors.toList());
            
            // 3. 等待所有分片查询完成
            CompletableFuture<List<QueryResult<T>>> allResults = CompletableFuture.allOf(
                shardFutures.toArray(new CompletableFuture[0]))
                .thenApply(v -> shardFutures.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList()));
            
            // 4. 合并分片结果
            CompletableFuture<QueryResult<T>> finalResult = allResults.thenApply(results -> {
                try {
                    QueryResult<T> mergedResult = resultAggregator.mergeShardResults(results, request);
                    
                    // 5. 记录监控信息
                    long executionTime = System.currentTimeMillis() - startTime;
                    monitorService.recordShardedQueryExecution(queryId, executionTime, shards.size());
                    
                    return mergedResult;
                    
                } catch (Exception e) {
                    log.error("分片结果合并失败: queryId={}", queryId, e);
                    throw new RuntimeException("分片结果合并失败", e);
                }
            });
            
            return finalResult;
            
        } catch (Exception e) {
            log.error("分片查询执行失败: queryId={}", queryId, e);
            return CompletableFuture.failedFuture(e);
        }
    }
    
    /**
     * 执行流水线查询
     */
    public <T> CompletableFuture<QueryResult<T>> executePipelineQuery(
            PipelineQueryRequest<T> request) {
        
        long startTime = System.currentTimeMillis();
        String queryId = generateQueryId();
        
        try {
            // 1. 构建查询流水线
            List<PipelineStage<T>> stages = buildPipelineStages(request);
            
            // 2. 执行流水线
            CompletableFuture<QueryResult<T>> pipelineResult = executePipeline(stages);
            
            // 3. 记录监控信息
            long executionTime = System.currentTimeMillis() - startTime;
            monitorService.recordPipelineQueryExecution(queryId, executionTime, stages.size());
            
            return pipelineResult;
            
        } catch (Exception e) {
            log.error("流水线查询执行失败: queryId={}", queryId, e);
            return CompletableFuture.failedFuture(e);
        }
    }
    
    /**
     * 并行执行任务
     */
    private <T> CompletableFuture<List<QueryResult<T>>> executeTasksInParallel(
            List<QueryTask<T>> tasks) {
        
        List<CompletableFuture<QueryResult<T>>> futures = tasks.stream()
            .map(task -> CompletableFuture.supplyAsync(() -> {
                try {
                    return taskExecutor.executeTask(task);
                } catch (Exception e) {
                    log.error("任务执行失败: taskId={}", task.getId(), e);
                    return QueryResult.error(e.getMessage());
                }
            }, forkJoinPool))
            .collect(Collectors.toList());
        
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList()));
    }
    
    /**
     * 执行分片查询
     */
    private <T> QueryResult<T> executeShardQuery(DataShard shard, ShardedQueryRequest<T> request) {
        // 实现分片查询逻辑
        // 这里可以根据具体的数据源类型实现不同的查询逻辑
        return QueryResult.empty();
    }
    
    /**
     * 执行流水线
     */
    private <T> CompletableFuture<QueryResult<T>> executePipeline(List<PipelineStage<T>> stages) {
        CompletableFuture<QueryResult<T>> result = CompletableFuture.completedFuture(QueryResult.empty());
        
        for (PipelineStage<T> stage : stages) {
            result = result.thenCompose(stage::execute);
        }
        
        return result;
    }
    
    /**
     * 解析查询请求
     */
    private QueryContext parseQueryRequest(ParallelQueryRequest<?> request) {
        QueryContext context = new QueryContext();
        context.setQueryId(generateQueryId());
        context.setRequest(request);
        context.setStartTime(System.currentTimeMillis());
        return context;
    }
    
    /**
     * 分解查询任务
     */
    private <T> List<QueryTask<T>> decomposeQueryTasks(QueryContext context) {
        // 实现任务分解逻辑
        return new ArrayList<>();
    }
    
    /**
     * 获取数据分片
     */
    private List<DataShard> getDataShards(String dataSource, String shardKey) {
        // 实现分片获取逻辑
        return new ArrayList<>();
    }
    
    /**
     * 构建流水线阶段
     */
    private <T> List<PipelineStage<T>> buildPipelineStages(PipelineQueryRequest<T> request) {
        // 实现流水线构建逻辑
        return new ArrayList<>();
    }
    
    /**
     * 生成查询ID
     */
    private String generateQueryId() {
        return "PQ" + System.currentTimeMillis() + RandomUtils.nextInt(1000, 9999);
    }
    
    @PreDestroy
    public void destroy() {
        forkJoinPool.shutdown();
        try {
            if (!forkJoinPool.awaitTermination(30, TimeUnit.SECONDS)) {
                forkJoinPool.shutdownNow();
            }
        } catch (InterruptedException e) {
            forkJoinPool.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        log.info("并行查询引擎已关闭");
    }
}

// 并行查询请求
public class ParallelQueryRequest<T> {
    private String queryId;
    private String dataSource;
    private QueryCondition condition;
    private QueryOptions options;
    private Class<T> resultType;
    
    // 构造函数和getter/setter方法
}

// 分片查询请求
public class ShardedQueryRequest<T> extends ParallelQueryRequest<T> {
    private String shardKey;
    private ShardingStrategy shardingStrategy;
    
    // 构造函数和getter/setter方法
}

// 流水线查询请求
public class PipelineQueryRequest<T> extends ParallelQueryRequest<T> {
    private List<PipelineStageConfig> stageConfigs;
    
    // 构造函数和getter/setter方法
}

// 查询上下文
public class QueryContext {
    private String queryId;
    private ParallelQueryRequest<?> request;
    private long startTime;
    private Map<String, Object> attributes;
    
    // 构造函数和getter/setter方法
}

// 查询任务
public class QueryTask<T> {
    private String id;
    private String dataSource;
    private QueryCondition condition;
    private Class<T> resultType;
    private int priority;
    
    // 构造函数和getter/setter方法
}

// 查询结果
public class QueryResult<T> {
    private boolean success;
    private List<T> data;
    private long totalCount;
    private String errorMessage;
    private Map<String, Object> metadata;
    
    public static <T> QueryResult<T> success(List<T> data) {
        QueryResult<T> result = new QueryResult<>();
        result.success = true;
        result.data = data;
        result.totalCount = data.size();
        return result;
    }
    
    public static <T> QueryResult<T> empty() {
        QueryResult<T> result = new QueryResult<>();
        result.success = true;
        result.data = new ArrayList<>();
        result.totalCount = 0;
        return result;
    }
    
    public static <T> QueryResult<T> error(String errorMessage) {
        QueryResult<T> result = new QueryResult<>();
        result.success = false;
        result.errorMessage = errorMessage;
        result.data = new ArrayList<>();
        result.totalCount = 0;
        return result;
    }
    
    // 构造函数和getter/setter方法
}

2.2 查询任务执行器

// 查询任务执行器
@Service
@Slf4j
public class QueryTaskExecutor {
    
    @Autowired
    private DataSourceManager dataSourceManager;
    
    @Autowired
    private QueryCacheService cacheService;
    
    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(
        10, 50, 60L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(1000),
        new ThreadFactory() {
            private final AtomicInteger threadNumber = new AtomicInteger(1);
            
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "query-task-" + threadNumber.getAndIncrement());
                thread.setDaemon(false);
                return thread;
            }
        },
        new ThreadPoolExecutor.CallerRunsPolicy()
    );
    
    /**
     * 执行查询任务
     */
    public <T> QueryResult<T> executeTask(QueryTask<T> task) {
        long startTime = System.currentTimeMillis();
        
        try {
            // 1. 检查缓存
            QueryResult<T> cachedResult = cacheService.getCachedResult(task);
            if (cachedResult != null) {
                log.debug("使用缓存结果: taskId={}", task.getId());
                return cachedResult;
            }
            
            // 2. 获取数据源
            DataSource dataSource = dataSourceManager.getDataSource(task.getDataSource());
            if (dataSource == null) {
                return QueryResult.error("数据源不存在: " + task.getDataSource());
            }
            
            // 3. 执行查询
            QueryResult<T> result = executeQuery(dataSource, task);
            
            // 4. 缓存结果
            if (result.isSuccess()) {
                cacheService.cacheResult(task, result);
            }
            
            // 5. 记录执行时间
            long executionTime = System.currentTimeMillis() - startTime;
            log.debug("任务执行完成: taskId={}, executionTime={}ms", task.getId(), executionTime);
            
            return result;
            
        } catch (Exception e) {
            log.error("任务执行异常: taskId={}", task.getId(), e);
            return QueryResult.error("任务执行异常: " + e.getMessage());
        }
    }
    
    /**
     * 异步执行查询任务
     */
    public <T> CompletableFuture<QueryResult<T>> executeTaskAsync(QueryTask<T> task) {
        return CompletableFuture.supplyAsync(() -> executeTask(task), executor);
    }
    
    /**
     * 批量执行查询任务
     */
    public <T> List<QueryResult<T>> executeTasksBatch(List<QueryTask<T>> tasks) {
        List<CompletableFuture<QueryResult<T>>> futures = tasks.stream()
            .map(this::executeTaskAsync)
            .collect(Collectors.toList());
        
        return futures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());
    }
    
    /**
     * 执行查询
     */
    private <T> QueryResult<T> executeQuery(DataSource dataSource, QueryTask<T> task) {
        try {
            // 根据数据源类型执行不同的查询逻辑
            if (dataSource instanceof DatabaseDataSource) {
                return executeDatabaseQuery((DatabaseDataSource) dataSource, task);
            } else if (dataSource instanceof CacheDataSource) {
                return executeCacheQuery((CacheDataSource) dataSource, task);
            } else if (dataSource instanceof ApiDataSource) {
                return executeApiQuery((ApiDataSource) dataSource, task);
            } else {
                return QueryResult.error("不支持的数据源类型: " + dataSource.getClass().getSimpleName());
            }
            
        } catch (Exception e) {
            log.error("查询执行失败: dataSource={}, taskId={}", dataSource.getName(), task.getId(), e);
            return QueryResult.error("查询执行失败: " + e.getMessage());
        }
    }
    
    /**
     * 执行数据库查询
     */
    private <T> QueryResult<T> executeDatabaseQuery(DatabaseDataSource dataSource, QueryTask<T> task) {
        try (Connection connection = dataSource.getConnection()) {
            // 构建SQL查询
            String sql = buildSqlQuery(task);
            
            try (PreparedStatement statement = connection.prepareStatement(sql)) {
                // 设置查询参数
                setQueryParameters(statement, task.getCondition());
                
                // 执行查询
                try (ResultSet resultSet = statement.executeQuery()) {
                    List<T> results = parseResultSet(resultSet, task.getResultType());
                    return QueryResult.success(results);
                }
            }
            
        } catch (Exception e) {
            log.error("数据库查询失败: dataSource={}, taskId={}", dataSource.getName(), task.getId(), e);
            return QueryResult.error("数据库查询失败: " + e.getMessage());
        }
    }
    
    /**
     * 执行缓存查询
     */
    private <T> QueryResult<T> executeCacheQuery(CacheDataSource dataSource, QueryTask<T> task) {
        try {
            String cacheKey = buildCacheKey(task);
            Object cachedData = dataSource.get(cacheKey);
            
            if (cachedData != null) {
                List<T> results = convertCachedData(cachedData, task.getResultType());
                return QueryResult.success(results);
            } else {
                return QueryResult.empty();
            }
            
        } catch (Exception e) {
            log.error("缓存查询失败: dataSource={}, taskId={}", dataSource.getName(), task.getId(), e);
            return QueryResult.error("缓存查询失败: " + e.getMessage());
        }
    }
    
    /**
     * 执行API查询
     */
    private <T> QueryResult<T> executeApiQuery(ApiDataSource dataSource, QueryTask<T> task) {
        try {
            // 构建API请求
            String url = buildApiUrl(dataSource, task);
            Map<String, String> headers = buildApiHeaders(dataSource);
            Map<String, Object> parameters = buildApiParameters(task.getCondition());
            
            // 发送HTTP请求
            String response = sendHttpRequest(url, headers, parameters);
            
            // 解析响应
            List<T> results = parseApiResponse(response, task.getResultType());
            return QueryResult.success(results);
            
        } catch (Exception e) {
            log.error("API查询失败: dataSource={}, taskId={}", dataSource.getName(), task.getId(), e);
            return QueryResult.error("API查询失败: " + e.getMessage());
        }
    }
    
    /**
     * 构建SQL查询
     */
    private String buildSqlQuery(QueryTask<?> task) {
        // 实现SQL构建逻辑
        return "SELECT * FROM " + task.getDataSource();
    }
    
    /**
     * 设置查询参数
     */
    private void setQueryParameters(PreparedStatement statement, QueryCondition condition) throws SQLException {
        // 实现参数设置逻辑
    }
    
    /**
     * 解析结果集
     */
    private <T> List<T> parseResultSet(ResultSet resultSet, Class<T> resultType) throws SQLException {
        List<T> results = new ArrayList<>();
        // 实现结果集解析逻辑
        return results;
    }
    
    /**
     * 构建缓存键
     */
    private String buildCacheKey(QueryTask<?> task) {
        return task.getDataSource() + ":" + task.getId();
    }
    
    /**
     * 转换缓存数据
     */
    private <T> List<T> convertCachedData(Object cachedData, Class<T> resultType) {
        // 实现缓存数据转换逻辑
        return new ArrayList<>();
    }
    
    /**
     * 构建API URL
     */
    private String buildApiUrl(ApiDataSource dataSource, QueryTask<?> task) {
        return dataSource.getBaseUrl() + "/query";
    }
    
    /**
     * 构建API请求头
     */
    private Map<String, String> buildApiHeaders(ApiDataSource dataSource) {
        Map<String, String> headers = new HashMap<>();
        headers.put("Content-Type", "application/json");
        headers.put("Authorization", "Bearer " + dataSource.getApiKey());
        return headers;
    }
    
    /**
     * 构建API参数
     */
    private Map<String, Object> buildApiParameters(QueryCondition condition) {
        Map<String, Object> parameters = new HashMap<>();
        // 实现参数构建逻辑
        return parameters;
    }
    
    /**
     * 发送HTTP请求
     */
    private String sendHttpRequest(String url, Map<String, String> headers, Map<String, Object> parameters) {
        // 实现HTTP请求发送逻辑
        return "{}";
    }
    
    /**
     * 解析API响应
     */
    private <T> List<T> parseApiResponse(String response, Class<T> resultType) {
        // 实现API响应解析逻辑
        return new ArrayList<>();
    }
    
    @PreDestroy
    public void destroy() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        log.info("查询任务执行器已关闭");
    }
}

// 数据源接口
public interface DataSource {
    String getName();
    String getType();
}

// 数据库数据源
public class DatabaseDataSource implements DataSource {
    private String name;
    private String url;
    private String username;
    private String password;
    private String driverClassName;
    
    public Connection getConnection() throws SQLException {
        // 实现数据库连接获取逻辑
        return null;
    }
    
    // 构造函数和getter/setter方法
}

// 缓存数据源
public class CacheDataSource implements DataSource {
    private String name;
    private RedisTemplate<String, Object> redisTemplate;
    
    public Object get(String key) {
        return redisTemplate.opsForValue().get(key);
    }
    
    // 构造函数和getter/setter方法
}

// API数据源
public class ApiDataSource implements DataSource {
    private String name;
    private String baseUrl;
    private String apiKey;
    
    // 构造函数和getter/setter方法
}

2.3 结果聚合器

// 结果聚合器
@Service
@Slf4j
public class ResultAggregator {
    
    @Autowired
    private QuerySortService sortService;
    
    @Autowired
    private QueryDeduplicationService deduplicationService;
    
    /**
     * 聚合查询结果
     */
    public <T> QueryResult<T> aggregate(List<QueryResult<T>> results, QueryContext context) {
        try {
            // 1. 检查是否有失败的结果
            List<QueryResult<T>> failedResults = results.stream()
                .filter(result -> !result.isSuccess())
                .collect(Collectors.toList());
            
            if (!failedResults.isEmpty()) {
                log.warn("存在失败的查询结果: count={}", failedResults.size());
                // 根据配置决定是否继续处理
                if (context.getRequest().getOptions().isFailFast()) {
                    return QueryResult.error("存在失败的查询结果");
                }
            }
            
            // 2. 合并成功的结果
            List<T> allData = results.stream()
                .filter(QueryResult::isSuccess)
                .flatMap(result -> result.getData().stream())
                .collect(Collectors.toList());
            
            // 3. 去重处理
            if (context.getRequest().getOptions().isDeduplication()) {
                allData = deduplicationService.deduplicate(allData, context.getRequest().getResultType());
            }
            
            // 4. 排序处理
            if (context.getRequest().getOptions().getSortFields() != null) {
                allData = sortService.sort(allData, context.getRequest().getOptions().getSortFields());
            }
            
            // 5. 分页处理
            QueryResult<T> finalResult = applyPagination(allData, context.getRequest().getOptions());
            
            // 6. 设置元数据
            setResultMetadata(finalResult, results, context);
            
            return finalResult;
            
        } catch (Exception e) {
            log.error("结果聚合失败: queryId={}", context.getQueryId(), e);
            return QueryResult.error("结果聚合失败: " + e.getMessage());
        }
    }
    
    /**
     * 合并分片结果
     */
    public <T> QueryResult<T> mergeShardResults(List<QueryResult<T>> results, ShardedQueryRequest<T> request) {
        try {
            // 1. 合并所有分片数据
            List<T> allData = results.stream()
                .filter(QueryResult::isSuccess)
                .flatMap(result -> result.getData().stream())
                .collect(Collectors.toList());
            
            // 2. 全局排序
            if (request.getOptions().getSortFields() != null) {
                allData = sortService.sort(allData, request.getOptions().getSortFields());
            }
            
            // 3. 全局去重
            if (request.getOptions().isDeduplication()) {
                allData = deduplicationService.deduplicate(allData, request.getResultType());
            }
            
            // 4. 应用分页
            QueryResult<T> finalResult = applyPagination(allData, request.getOptions());
            
            // 5. 设置分片元数据
            setShardMetadata(finalResult, results);
            
            return finalResult;
            
        } catch (Exception e) {
            log.error("分片结果合并失败", e);
            return QueryResult.error("分片结果合并失败: " + e.getMessage());
        }
    }
    
    /**
     * 应用分页
     */
    private <T> QueryResult<T> applyPagination(List<T> data, QueryOptions options) {
        if (options.getPageSize() == null || options.getPageSize() <= 0) {
            return QueryResult.success(data);
        }
        
        int pageSize = options.getPageSize();
        int pageNumber = options.getPageNumber() != null ? options.getPageNumber() : 1;
        int offset = (pageNumber - 1) * pageSize;
        
        if (offset >= data.size()) {
            return QueryResult.success(new ArrayList<>());
        }
        
        int endIndex = Math.min(offset + pageSize, data.size());
        List<T> pagedData = data.subList(offset, endIndex);
        
        QueryResult<T> result = QueryResult.success(pagedData);
        result.setTotalCount(data.size());
        result.getMetadata().put("pageNumber", pageNumber);
        result.getMetadata().put("pageSize", pageSize);
        result.getMetadata().put("totalPages", (int) Math.ceil((double) data.size() / pageSize));
        
        return result;
    }
    
    /**
     * 设置结果元数据
     */
    private <T> void setResultMetadata(QueryResult<T> result, List<QueryResult<T>> sourceResults, QueryContext context) {
        Map<String, Object> metadata = result.getMetadata();
        
        // 设置执行信息
        metadata.put("queryId", context.getQueryId());
        metadata.put("executionTime", System.currentTimeMillis() - context.getStartTime());
        metadata.put("sourceResultCount", sourceResults.size());
        metadata.put("successfulResultCount", sourceResults.stream().mapToLong(r -> r.isSuccess() ? 1 : 0).sum());
        metadata.put("failedResultCount", sourceResults.stream().mapToLong(r -> r.isSuccess() ? 0 : 1).sum());
        
        // 设置聚合信息
        metadata.put("aggregationType", "parallel");
        metadata.put("deduplicationApplied", context.getRequest().getOptions().isDeduplication());
        metadata.put("sortingApplied", context.getRequest().getOptions().getSortFields() != null);
    }
    
    /**
     * 设置分片元数据
     */
    private <T> void setShardMetadata(QueryResult<T> result, List<QueryResult<T>> shardResults) {
        Map<String, Object> metadata = result.getMetadata();
        
        metadata.put("shardCount", shardResults.size());
        metadata.put("successfulShardCount", shardResults.stream().mapToLong(r -> r.isSuccess() ? 1 : 0).sum());
        metadata.put("failedShardCount", shardResults.stream().mapToLong(r -> r.isSuccess() ? 0 : 1).sum());
        metadata.put("aggregationType", "sharded");
    }
}

// 查询选项
public class QueryOptions {
    private Integer pageNumber;
    private Integer pageSize;
    private List<SortField> sortFields;
    private boolean deduplication = false;
    private boolean failFast = true;
    private long timeout = 30000; // 30秒超时
    
    // 构造函数和getter/setter方法
}

// 排序字段
public class SortField {
    private String fieldName;
    private SortDirection direction;
    
    public enum SortDirection {
        ASC, DESC
    }
    
    // 构造函数和getter/setter方法
}

// 查询条件
public class QueryCondition {
    private Map<String, Object> filters;
    private String whereClause;
    private Map<String, Object> parameters;
    
    // 构造函数和getter/setter方法
}

2.4 并行查询控制器

// 并行查询控制器
@RestController
@RequestMapping("/api/parallel-query")
@Slf4j
public class ParallelQueryController {
    
    @Autowired
    private ParallelQueryEngine queryEngine;
    
    @Autowired
    private QueryValidationService validationService;
    
    /**
     * 执行并行查询
     */
    @PostMapping("/execute")
    public ResponseEntity<CompletableFuture<QueryResult<Object>>> executeParallelQuery(
            @RequestBody ParallelQueryRequest<Object> request) {
        
        try {
            // 1. 验证请求
            ValidationResult validation = validationService.validateRequest(request);
            if (!validation.isValid()) {
                return ResponseEntity.badRequest().build();
            }
            
            // 2. 执行并行查询
            CompletableFuture<QueryResult<Object>> result = queryEngine.executeParallelQuery(request);
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            log.error("并行查询执行失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 执行分片查询
     */
    @PostMapping("/sharded")
    public ResponseEntity<CompletableFuture<QueryResult<Object>>> executeShardedQuery(
            @RequestBody ShardedQueryRequest<Object> request) {
        
        try {
            // 1. 验证请求
            ValidationResult validation = validationService.validateRequest(request);
            if (!validation.isValid()) {
                return ResponseEntity.badRequest().build();
            }
            
            // 2. 执行分片查询
            CompletableFuture<QueryResult<Object>> result = queryEngine.executeShardedQuery(request);
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            log.error("分片查询执行失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 执行流水线查询
     */
    @PostMapping("/pipeline")
    public ResponseEntity<CompletableFuture<QueryResult<Object>>> executePipelineQuery(
            @RequestBody PipelineQueryRequest<Object> request) {
        
        try {
            // 1. 验证请求
            ValidationResult validation = validationService.validateRequest(request);
            if (!validation.isValid()) {
                return ResponseEntity.badRequest().build();
            }
            
            // 2. 执行流水线查询
            CompletableFuture<QueryResult<Object>> result = queryEngine.executePipelineQuery(request);
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            log.error("流水线查询执行失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取查询状态
     */
    @GetMapping("/status/{queryId}")
    public ResponseEntity<QueryStatus> getQueryStatus(@PathVariable String queryId) {
        try {
            QueryStatus status = queryEngine.getQueryStatus(queryId);
            return ResponseEntity.ok(status);
        } catch (Exception e) {
            log.error("获取查询状态失败: queryId={}", queryId, e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 取消查询
     */
    @PostMapping("/cancel/{queryId}")
    public ResponseEntity<Void> cancelQuery(@PathVariable String queryId) {
        try {
            boolean cancelled = queryEngine.cancelQuery(queryId);
            if (cancelled) {
                return ResponseEntity.ok().build();
            } else {
                return ResponseEntity.notFound().build();
            }
        } catch (Exception e) {
            log.error("取消查询失败: queryId={}", queryId, e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
}

// 查询验证服务
@Service
@Slf4j
public class QueryValidationService {
    
    /**
     * 验证查询请求
     */
    public ValidationResult validateRequest(ParallelQueryRequest<?> request) {
        ValidationResult result = new ValidationResult();
        
        try {
            // 1. 验证数据源
            if (request.getDataSource() == null || request.getDataSource().trim().isEmpty()) {
                result.addError("数据源不能为空");
            }
            
            // 2. 验证查询条件
            if (request.getCondition() == null) {
                result.addError("查询条件不能为空");
            }
            
            // 3. 验证结果类型
            if (request.getResultType() == null) {
                result.addError("结果类型不能为空");
            }
            
            // 4. 验证分页参数
            if (request.getOptions() != null) {
                validatePaginationOptions(request.getOptions(), result);
            }
            
            // 5. 验证分片查询
            if (request instanceof ShardedQueryRequest) {
                validateShardedRequest((ShardedQueryRequest<?>) request, result);
            }
            
            // 6. 验证流水线查询
            if (request instanceof PipelineQueryRequest) {
                validatePipelineRequest((PipelineQueryRequest<?>) request, result);
            }
            
        } catch (Exception e) {
            log.error("查询请求验证异常", e);
            result.addError("请求验证异常: " + e.getMessage());
        }
        
        return result;
    }
    
    /**
     * 验证分页选项
     */
    private void validatePaginationOptions(QueryOptions options, ValidationResult result) {
        if (options.getPageSize() != null && options.getPageSize() <= 0) {
            result.addError("页面大小必须大于0");
        }
        
        if (options.getPageNumber() != null && options.getPageNumber() <= 0) {
            result.addError("页码必须大于0");
        }
        
        if (options.getPageSize() != null && options.getPageSize() > 10000) {
            result.addError("页面大小不能超过10000");
        }
    }
    
    /**
     * 验证分片请求
     */
    private void validateShardedRequest(ShardedQueryRequest<?> request, ValidationResult result) {
        if (request.getShardKey() == null || request.getShardKey().trim().isEmpty()) {
            result.addError("分片键不能为空");
        }
        
        if (request.getShardingStrategy() == null) {
            result.addError("分片策略不能为空");
        }
    }
    
    /**
     * 验证流水线请求
     */
    private void validatePipelineRequest(PipelineQueryRequest<?> request, ValidationResult result) {
        if (request.getStageConfigs() == null || request.getStageConfigs().isEmpty()) {
            result.addError("流水线阶段配置不能为空");
        }
    }
}

// 验证结果
public class ValidationResult {
    private boolean valid = true;
    private List<String> errors = new ArrayList<>();
    
    public void addError(String error) {
        this.valid = false;
        this.errors.add(error);
    }
    
    // getter/setter方法
}

// 查询状态
public class QueryStatus {
    private String queryId;
    private QueryState state;
    private long startTime;
    private long endTime;
    private String errorMessage;
    private Map<String, Object> progress;
    
    public enum QueryState {
        PENDING,    // 等待中
        RUNNING,    // 执行中
        COMPLETED,  // 已完成
        FAILED,     // 失败
        CANCELLED   // 已取消
    }
    
    // 构造函数和getter/setter方法
}

三、性能优化与监控

3.1 查询缓存服务

// 查询缓存服务
@Service
@Slf4j
public class QueryCacheService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    private static final String CACHE_KEY_PREFIX = "query_cache:";
    private static final long DEFAULT_TTL = 300; // 5分钟
    
    /**
     * 获取缓存结果
     */
    public <T> QueryResult<T> getCachedResult(QueryTask<T> task) {
        try {
            String cacheKey = buildCacheKey(task);
            Object cachedData = redisTemplate.opsForValue().get(cacheKey);
            
            if (cachedData != null) {
                log.debug("命中查询缓存: taskId={}", task.getId());
                return convertCachedData(cachedData, task.getResultType());
            }
            
            return null;
            
        } catch (Exception e) {
            log.error("获取缓存结果失败: taskId={}", task.getId(), e);
            return null;
        }
    }
    
    /**
     * 缓存查询结果
     */
    public <T> void cacheResult(QueryTask<T> task, QueryResult<T> result) {
        try {
            if (!result.isSuccess()) {
                return; // 不缓存失败的结果
            }
            
            String cacheKey = buildCacheKey(task);
            long ttl = calculateTTL(task);
            
            redisTemplate.opsForValue().set(cacheKey, result, ttl, TimeUnit.SECONDS);
            
            log.debug("缓存查询结果: taskId={}, ttl={}s", task.getId(), ttl);
            
        } catch (Exception e) {
            log.error("缓存查询结果失败: taskId={}", task.getId(), e);
        }
    }
    
    /**
     * 删除缓存
     */
    public void evictCache(QueryTask<?> task) {
        try {
            String cacheKey = buildCacheKey(task);
            redisTemplate.delete(cacheKey);
            
            log.debug("删除查询缓存: taskId={}", task.getId());
            
        } catch (Exception e) {
            log.error("删除查询缓存失败: taskId={}", task.getId(), e);
        }
    }
    
    /**
     * 清空所有缓存
     */
    public void clearAllCache() {
        try {
            Set<String> keys = redisTemplate.keys(CACHE_KEY_PREFIX + "*");
            if (keys != null && !keys.isEmpty()) {
                redisTemplate.delete(keys);
                log.info("清空所有查询缓存: count={}", keys.size());
            }
        } catch (Exception e) {
            log.error("清空查询缓存失败", e);
        }
    }
    
    /**
     * 构建缓存键
     */
    private String buildCacheKey(QueryTask<?> task) {
        StringBuilder keyBuilder = new StringBuilder(CACHE_KEY_PREFIX);
        keyBuilder.append(task.getDataSource()).append(":");
        keyBuilder.append(task.getId()).append(":");
        keyBuilder.append(task.getCondition().hashCode());
        return keyBuilder.toString();
    }
    
    /**
     * 计算TTL
     */
    private long calculateTTL(QueryTask<?> task) {
        // 根据任务类型和数据源计算TTL
        // 这里可以实现更复杂的TTL计算逻辑
        return DEFAULT_TTL;
    }
    
    /**
     * 转换缓存数据
     */
    @SuppressWarnings("unchecked")
    private <T> QueryResult<T> convertCachedData(Object cachedData, Class<T> resultType) {
        if (cachedData instanceof QueryResult) {
            return (QueryResult<T>) cachedData;
        }
        return null;
    }
}

3.2 查询监控服务

// 查询监控服务
@Service
@Slf4j
public class QueryMonitorService {
    
    private final Map<String, QueryExecutionInfo> executionInfos = new ConcurrentHashMap<>();
    private final AtomicLong totalQueries = new AtomicLong(0);
    private final AtomicLong successfulQueries = new AtomicLong(0);
    private final AtomicLong failedQueries = new AtomicLong(0);
    
    /**
     * 记录查询执行
     */
    public void recordQueryExecution(String queryId, long executionTime, int taskCount) {
        QueryExecutionInfo info = new QueryExecutionInfo();
        info.setQueryId(queryId);
        info.setExecutionTime(executionTime);
        info.setTaskCount(taskCount);
        info.setEndTime(System.currentTimeMillis());
        
        executionInfos.put(queryId, info);
        
        totalQueries.incrementAndGet();
        successfulQueries.incrementAndGet();
        
        log.info("查询执行完成: queryId={}, executionTime={}ms, taskCount={}", 
                queryId, executionTime, taskCount);
    }
    
    /**
     * 记录分片查询执行
     */
    public void recordShardedQueryExecution(String queryId, long executionTime, int shardCount) {
        QueryExecutionInfo info = new QueryExecutionInfo();
        info.setQueryId(queryId);
        info.setExecutionTime(executionTime);
        info.setShardCount(shardCount);
        info.setEndTime(System.currentTimeMillis());
        
        executionInfos.put(queryId, info);
        
        totalQueries.incrementAndGet();
        successfulQueries.incrementAndGet();
        
        log.info("分片查询执行完成: queryId={}, executionTime={}ms, shardCount={}", 
                queryId, executionTime, shardCount);
    }
    
    /**
     * 记录流水线查询执行
     */
    public void recordPipelineQueryExecution(String queryId, long executionTime, int stageCount) {
        QueryExecutionInfo info = new QueryExecutionInfo();
        info.setQueryId(queryId);
        info.setExecutionTime(executionTime);
        info.setStageCount(stageCount);
        info.setEndTime(System.currentTimeMillis());
        
        executionInfos.put(queryId, info);
        
        totalQueries.incrementAndGet();
        successfulQueries.incrementAndGet();
        
        log.info("流水线查询执行完成: queryId={}, executionTime={}ms, stageCount={}", 
                queryId, executionTime, stageCount);
    }
    
    /**
     * 记录查询失败
     */
    public void recordQueryFailure(String queryId, String errorMessage) {
        QueryExecutionInfo info = executionInfos.get(queryId);
        if (info != null) {
            info.setErrorMessage(errorMessage);
            info.setEndTime(System.currentTimeMillis());
        }
        
        totalQueries.incrementAndGet();
        failedQueries.incrementAndGet();
        
        log.error("查询执行失败: queryId={}, error={}", queryId, errorMessage);
    }
    
    /**
     * 获取查询统计
     */
    public QueryStatistics getQueryStatistics() {
        QueryStatistics stats = new QueryStatistics();
        stats.setTotalQueries(totalQueries.get());
        stats.setSuccessfulQueries(successfulQueries.get());
        stats.setFailedQueries(failedQueries.get());
        stats.setSuccessRate(calculateSuccessRate());
        stats.setAverageExecutionTime(calculateAverageExecutionTime());
        stats.setMaxExecutionTime(calculateMaxExecutionTime());
        stats.setMinExecutionTime(calculateMinExecutionTime());
        
        return stats;
    }
    
    /**
     * 获取查询执行信息
     */
    public QueryExecutionInfo getQueryExecutionInfo(String queryId) {
        return executionInfos.get(queryId);
    }
    
    /**
     * 获取所有查询执行信息
     */
    public Map<String, QueryExecutionInfo> getAllQueryExecutionInfos() {
        return new HashMap<>(executionInfos);
    }
    
    /**
     * 计算成功率
     */
    private double calculateSuccessRate() {
        long total = totalQueries.get();
        return total > 0 ? (double) successfulQueries.get() / total : 0.0;
    }
    
    /**
     * 计算平均执行时间
     */
    private double calculateAverageExecutionTime() {
        return executionInfos.values().stream()
            .mapToLong(QueryExecutionInfo::getExecutionTime)
            .average()
            .orElse(0.0);
    }
    
    /**
     * 计算最大执行时间
     */
    private long calculateMaxExecutionTime() {
        return executionInfos.values().stream()
            .mapToLong(QueryExecutionInfo::getExecutionTime)
            .max()
            .orElse(0L);
    }
    
    /**
     * 计算最小执行时间
     */
    private long calculateMinExecutionTime() {
        return executionInfos.values().stream()
            .mapToLong(QueryExecutionInfo::getExecutionTime)
            .min()
            .orElse(0L);
    }
}

// 查询执行信息
public class QueryExecutionInfo {
    private String queryId;
    private long executionTime;
    private int taskCount;
    private int shardCount;
    private int stageCount;
    private long startTime;
    private long endTime;
    private String errorMessage;
    
    // 构造函数和getter/setter方法
}

// 查询统计
public class QueryStatistics {
    private long totalQueries;
    private long successfulQueries;
    private long failedQueries;
    private double successRate;
    private double averageExecutionTime;
    private long maxExecutionTime;
    private long minExecutionTime;
    
    // 构造函数和getter/setter方法
}

四、最佳实践与总结

4.1 并行查询最佳实践

4.1.1 任务分解策略

数据分片：合理划分数据分片大小
查询分解：将复杂查询分解为简单查询
依赖分析：分析查询间的依赖关系
负载均衡：智能分配任务到不同线程

4.1.2 并行执行优化

线程池配置：根据CPU核心数配置线程池
异步处理：合理使用异步处理提高响应速度
资源控制：精确控制并行度和资源使用
超时处理：设置合理的超时时间

4.1.3 结果聚合策略

数据合并：高效合并多个查询结果
排序优化：使用外部排序处理大数据集
去重处理：智能去重避免重复数据
分页支持：支持高效的分页处理

4.1.4 性能优化策略

缓存策略：合理使用缓存减少重复查询
索引优化：优化数据库索引提高查询效率
连接池：使用连接池管理数据库连接
监控告警：建立完善的监控和告警机制

4.2 架构演进建议

4.2.1 微服务架构演进

服务拆分：将查询服务拆分为多个微服务
服务治理：实现服务的注册发现、负载均衡
数据一致性：使用分布式事务保证数据一致性
容错处理：实现熔断、降级、重试机制

4.2.2 云原生架构演进

容器化部署：使用Docker等容器技术部署
弹性伸缩：实现基于负载的自动扩缩容
服务网格：使用Istio等服务网格技术
云原生存储：使用云原生的存储服务

4.2.3 智能化查询

AI驱动优化：使用机器学习优化查询策略
自适应查询：根据数据特征自动调整查询策略
预测性查询：预测查询需求并提前准备
智能缓存：基于访问模式的智能缓存

4.3 总结

并行查询接口是企业级应用性能优化的重要手段，通过合理的架构设计，完善的并行处理，智能的结果聚合，可以实现高效、稳定的并行查询解决方案。随着大数据和云原生技术的普及，并行查询将更加智能化和自动化。

在未来的发展中，企业需要持续关注技术发展趋势，不断优化和完善并行查询策略，以适应不断变化的业务需求和技术环境。通过本文的深入分析和实践指导，希望能够为企业构建高质量的并行查询解决方案提供有价值的参考和帮助，推动企业级应用在数据处理场景下的稳定运行和持续发展。