Java如何实现支持优先级执行的CompletableFuture？

CompletableFuture原生API完全可以实现这个需求，不需要依赖额外框架，核心是实现严格优先级判定+高优结果命中后的低优任务取消逻辑，不要直接套用anyOf()——anyOf()会返回最先完成的任务结果，完全不感知优先级，只要低优任务响应更快就会返回错误结果，不符合业务要求。
核心逻辑遵循3个规则：

• 所有优先级的任务并行启动，不存在串行等待，避免高优任务慢导致整体响应延迟
• 严格按照配置的优先级顺序判定结果：只要更高优先级的任务还未执行完成，哪怕低优先级任务已经返回有效结果，也暂时不返回，等待高优任务判定
• 高优先级任务返回有效非null结果时，立即终止所有更低优先级的运行中任务，直接返回高优结果；如果高优任务返回null/抛出异常，自动降级到次优先级任务的结果判定

注意：CompletableFuture.cancel(true)会触发线程中断，主流DynamoDB SDK、JDBC驱动都默认支持中断响应，调用cancel后会立即终止IO等待，不会出现资源泄漏。如果使用的是不支持中断的老旧客户端，建议在任务逻辑内部增加中断状态检查，避免无效资源开销。

import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

public class PriorityParallelFetcher<T> {

    /**
     * 按配置的优先级并行拉取数据
     * @param fetchTasks 按优先级从高到低传入的数据获取任务
     * @return 匹配最高优先级的有效结果
     * @param <T> 数据类型
     */
    @SafeVarargs
    public static <T> CompletableFuture<T> fetch(Supplier<T>... fetchTasks) {
        CompletableFuture<T> finalResult = new CompletableFuture<>();
        // 存储所有异步任务的引用
        CompletableFuture<T>[] taskFutures = new CompletableFuture[fetchTasks.length];

        // 并行启动所有任务
        for (int i = 0; i < fetchTasks.length; i++) {
            int taskIndex = i;
            taskFutures[i] = CompletableFuture.supplyAsync(fetchTasks[i]);
            // 每个任务完成时触发优先级判定
            taskFutures[i].whenComplete((res, ex) -> {
                if (finalResult.isDone()) {
                    return;
                }
                // 从最高优先级开始遍历，找第一个有效结果
                for (int priorityIdx = 0; priorityIdx < taskFutures.length; priorityIdx++) {
                    CompletableFuture<T> currentTask = taskFutures[priorityIdx];
                    // 遇到第一个未完成的高优任务，停止判定等待其完成
                    if (!currentTask.isDone()) {
                        break;
                    }
                    try {
                        T currentRes = currentTask.getNow(null);
                        // 找到有效非null结果，填充最终返回值
                        if (currentRes != null) {
                            finalResult.complete(currentRes);
                            // 取消所有优先级更低的未完成任务
                            for (int j = priorityIdx + 1; j < taskFutures.length; j++) {
                                taskFutures[j].cancel(true);
                            }
                            return;
                        }
                    } catch (Exception e) {
                        // 高优任务抛出异常，继续判定下一个优先级
                        continue;
                    }

                    // 已经遍历到最后一个任务，无有效结果，直接返回当前状态
                    if (priorityIdx == taskFutures.length - 1) {
                        if (ex != null) {
                            finalResult.completeExceptionally(ex);
                        } else {
                            finalResult.complete(res);
                        }
                    }
                }
            });
        }
        return finalResult;
    }
}

调用示例 #

针对优先级配置为["A","B"]的场景，只需要按优先级顺序传入对应方法引用即可：

// A（DynamoDB）优先级高于B（SQL），两个任务并行执行
CompletableFuture<UserData> result = PriorityParallelFetcher.fetch(
        () -> dynamoDbMapper.load(UserData.class, userId), // 方法A
        () -> sqlUserMapper.selectById(userId)             // 方法B
);

// 需要调整优先级为B高于A时，调换两个Supplier的传入顺序即可

• 线程池选择：上述示例默认使用ForkJoinPool.commonPool()，由于DynamoDB、SQL调用都是IO密集型任务，建议自定义IO专用线程池，作为第二个参数传入CompletableFuture.supplyAsync(fetchTasks[i], yourIoExecutor)，避免公共池被IO阻塞影响其他业务逻辑。
• 异常逻辑调整：上述实现默认高优任务抛出异常时会自动降级到次优先级任务，如果需要高优任务异常直接返回、不做降级，只需要在捕获currentTask异常时直接将异常填充到finalResult即可。
• 取消逻辑的兼容性：如果业务任务逻辑里包含不响应中断的阻塞操作，需要在任务内部定期检查Thread.currentThread().isInterrupted()状态，主动退出逻辑，避免任务取消后仍然占用资源。

内容的提问来源于stack exchange，提问作者Konstantin Rezchikov