如何从Futures列表中等待符合预期的结果？

可以利用CompletionService按任务完成顺序获取结果，依次检查每个完成任务的返回值，直到找到首个非null结果，同时取消剩余未完成任务；若当前任务返回null则继续等待下一个完成的任务。

针对Callable任务的实现代码 #

import java.util.List;
import java.util.concurrent.*;

public class FirstNonNullFutureHandler {
    public static <T> T getFirstNonNullResult(ExecutorService executor, List<Callable<T>> tasks) throws InterruptedException, ExecutionException {
        CompletionService<T> completionService = new ExecutorCompletionService<>(executor);
        
        // 提交所有任务到CompletionService
        for (Callable<T> task : tasks) {
            completionService.submit(task);
        }

        int remainingTasks = tasks.size();
        while (remainingTasks > 0) {
            // 阻塞获取下一个完成的任务
            Future<T> completedFuture = completionService.take();
            remainingTasks--;

            try {
                T result = completedFuture.get();
                if (result != null) {
                    // 找到有效结果，取消所有未完成任务
                    executor.shutdownNow();
                    return result;
                }
                // 结果为null，继续循环等待下一个完成任务
            } catch (ExecutionException e) {
                // 任务执行异常，此处视为无效结果，可根据需求调整逻辑（比如直接抛出异常）
                System.err.println("任务执行失败：" + e.getCause().getMessage());
            }
        }
        // 所有任务均返回null
        return null;
    }

    // 测试示例
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        List<Callable<String>> testTasks = List.of(
            () -> { Thread.sleep(100); return null; },
            () -> { Thread.sleep(50); return "第一个有效结果"; },
            () -> { Thread.sleep(200); return "后续结果"; }
        );

        String result = getFirstNonNullResult(executor, testTasks);
        System.out.println("最终获取结果：" + result);
        
        executor.shutdown();
    }
}

针对已有Future列表的适配 #

如果你手里已经是List<Future<T>>而不是Callable，可以将每个Future包装为Callable提交到CompletionService，完成后手动取消剩余未完成任务：

public static <T> T getFirstNonNullFromFutures(List<Future<T>> futures) throws InterruptedException, ExecutionException {
    ExecutorService executor = Executors.newSingleThreadExecutor();
    CompletionService<T> completionService = new ExecutorCompletionService<>(executor);

    // 包装已有Future为Callable提交
    for (Future<T> future : futures) {
        completionService.submit(() -> future.get());
    }

    int remainingTasks = futures.size();
    while (remainingTasks > 0) {
        Future<T> completedFuture = completionService.take();
        remainingTasks--;

        try {
            T result = completedFuture.get();
            if (result != null) {
                // 取消所有未完成的原Future
                for (Future<T> f : futures) {
                    if (!f.isDone()) {
                        f.cancel(true);
                    }
                }
                executor.shutdownNow();
                return result;
            }
        } catch (ExecutionException e) {
            System.err.println("任务执行异常：" + e.getCause());
        }
    }
    executor.shutdown();
    return null;
}

关键细节说明 #

• CompletionService的核心作用是将完成的Future按顺序存入队列，take()方法会阻塞直到有任务完成，保证我们按任务实际完成顺序处理结果。
• 调用shutdownNow()或future.cancel(true)时，任务需要响应中断才能被有效取消（比如任务中包含Thread.sleep()这类可中断方法）。
• 异常处理逻辑可根据业务调整：比如任务抛出异常时直接终止流程，还是跳过继续等待下一个任务。

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