Java 8基于CompletableFuture实现带超时异步API调用咨询

你原有代码无法满足需求的核心原因有三点：

• parallelStream本身不提供全局超时能力，一旦任务提交到流中，会一直阻塞到所有任务执行完成，无法中途截断获取部分结果
• CompletableFuture版本中，你在forEach循环里创建的future是局部变量，外层无法拿到所有任务的引用，自然没法统一管控等待时长和收集结果
• 对单个future调用超时get方法没有意义，需要对批量任务做统一的超时等待逻辑

以下实现完全兼容Java 8版本，同时满足可配置超时、全异步处理、支持任意数量待处理条目的核心要求：

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class BatchAsyncApiCaller {
    // 自定义独立线程池，避免使用ForkJoin公共池影响其他业务，核心线程数可根据实际并发需求调整
    private static final ExecutorService API_EXECUTOR = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors() * 2,
            r -> {
                Thread t = new Thread(r);
                t.setName("api-call-worker-" + t.getId());
                t.setDaemon(true); // 设为守护线程，不阻塞JVM正常退出
                return t;
            }
    );

    public static void main(String[] args) {
        List<Integer> ids = Arrays.asList(1,2,3,4,5);
        // 全局超时配置，单位秒
        long timeoutConfig = 60;
        List<Integer> successResults = batchCallWithPartialResult(ids, timeoutConfig);
        // 后续直接处理超时时间内拿到的成功结果即可
        System.out.println("超时窗口内成功返回的结果：" + successResults);
    }

    /**
     * 批量异步调用API，超时后返回已成功的结果
     * @param ids 待处理的ID列表，支持任意长度
     * @param timeoutSeconds 全局超时时间
     * @return 超时时间内正常返回的结果列表
     */
    private static List<Integer> batchCallWithPartialResult(List<Integer> ids, long timeoutSeconds) {
        // 1. 为每个ID创建独立异步任务，持有所有任务的引用
        List<CompletableFuture<Integer>> allTaskFutures = ids.stream()
                .map(id -> CompletableFuture.supplyAsync(() -> serviceCall(id), API_EXECUTOR)
                        // 单个任务异常隔离，避免单接口失败打断整体流程
                        .exceptionally(ex -> {
                            System.err.printf("ID为%d的调用失败：%s%n", id, ex.getMessage());
                            return null;
                        })
                )
                .collect(Collectors.toList());

        // 2. 组合所有任务的等待句柄
        CompletableFuture<Void> allTaskWaiter = CompletableFuture.allOf(
                allTaskFutures.toArray(new CompletableFuture[0])
        );

        try {
            // 统一等待，要么所有任务完成，要么触发超时
            allTaskWaiter.get(timeoutSeconds, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            // 超时是预期内场景，无需抛出，直接进入结果收集流程
            System.out.println("等待超时，开始收集已完成的结果");
        } catch (InterruptedException | ExecutionException e) {
            System.err.println("等待过程中出现异常：" + e.getMessage());
        } finally {
            List<Integer> validResults = new ArrayList<>();
            for (CompletableFuture<Integer> future : allTaskFutures) {
                // 只收集正常完成、没有抛出异常的任务结果
                if (future.isDone() && !future.isCompletedExceptionally()) {
                    try {
                        Integer res = future.get();
                        if (res != null) { // 过滤调用失败返回的空值
                            validResults.add(res);
                        }
                    } catch (InterruptedException | ExecutionException ignored) {}
                } else {
                    // 取消未完成的任务，释放线程资源，中断正在执行的无效调用
                    future.cancel(true);
                }
            }
            return validResults;
        }
    }

    // 模拟业务API调用
    private static Integer serviceCall(Integer id) {
        try {
            // 替换为实际的API调用逻辑即可，这里模拟不同接口的耗时差异
            Thread.sleep(id * 20 * 1000L);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("调用被中断", e);
        }
        return id + 1;
    }
}

• 超时控制：通过CompletableFuture.allOf()组合所有批量任务后，统一调用带超时参数的get()方法，超时触发后立即终止等待，不会阻塞主线程
• 异步处理：所有API调用都提交到独立的自定义线程池执行，和业务主线程隔离，线程池参数可以根据实际的接口并发能力、QPS限制灵活调整
• 任意数量条目支持：待处理ID列表没有固定长度限制，会自动为每个ID创建独立异步任务，不需要提前预知待处理条目总数
• 异常隔离：单个API调用失败、超时不会中断整个批量流程，只会跳过异常任务，正常收集其余成功结果
• 资源回收：等待结束后会主动取消所有未完成的任务，避免无效的API调用占用线程、网络资源

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