如何为并行流中的任务抛出TimeoutException超时异常

针对你给出的并行流代码，要实现「调用外部API并设置执行超时」的需求，有两种常用的Java实现方式，以下是具体代码和说明：

方法一：使用CompletableFuture实现超时 #

利用CompletableFuture的orTimeout方法可以直接为异步任务设置超时时间，超时后会抛出TimeoutException，正好匹配你代码中的捕获逻辑：

StreamSupport.stream(jsonArray.spliterator(), true).forEach(s -> {
    try {
        // 替换为你的实际API调用逻辑
        CompletableFuture.runAsync(() -> {
            // 这里写调用外部API的代码，例如：
            // externalApiClient.call(s.getAsJsonObject());
        }).orTimeout(5, TimeUnit.SECONDS) // 设置5秒超时，可调整时间和时间单位
          .get(); // 阻塞等待任务完成，超时则抛出TimeoutException
    } catch (TimeoutException e) {
        s.getAsJsonObject().addProperty("processStatus", "Failure");
    } catch (InterruptedException | ExecutionException e) {
        // 处理线程中断或API调用本身抛出的异常
        s.getAsJsonObject().addProperty("processStatus", "Error");
    }
});

说明 #

• orTimeout方法会在指定时间后自动触发超时，抛出TimeoutException；
• get()方法会阻塞当前线程等待任务结果，超时异常会被外层catch捕获；
• 额外捕获InterruptedException和ExecutionException是为了处理线程被中断或API调用本身出错的场景。

方法二：使用ExecutorService+Future实现超时 #

如果需要更灵活的线程池管理，可以自定义线程池，结合Future的get(long timeout, TimeUnit unit)方法实现超时：

// 提前初始化自定义线程池（建议在类级别声明，避免重复创建）
private static final ExecutorService apiExecutor = Executors.newFixedThreadPool(10);

StreamSupport.stream(jsonArray.spliterator(), true).forEach(s -> {
    try {
        Future<Void> future = apiExecutor.submit(() -> {
            // 这里写调用外部API的代码
            // externalApiClient.call(s.getAsJsonObject());
        });
        // 设置5秒超时
        future.get(5, TimeUnit.SECONDS);
    } catch (TimeoutException e) {
        s.getAsJsonObject().addProperty("processStatus", "Failure");
        // 超时后尝试取消任务
        future.cancel(true);
    } catch (InterruptedException e) {
        // 恢复线程中断状态
        Thread.currentThread().interrupt();
        s.getAsJsonObject().addProperty("processStatus", "Interrupted");
    } catch (ExecutionException e) {
        // 处理API调用本身抛出的异常
        s.getAsJsonObject().addProperty("processStatus", "Error");
    }
});

// 程序结束时记得关闭线程池
// apiExecutor.shutdown();

说明 #

• 自定义线程池可以更好地控制并发数，避免并行流默认的ForkJoinPool被占满；
• future.get()方法超时会抛出TimeoutException，此时可以调用cancel(true)尝试中断正在执行的API调用；
• 同样需要处理线程中断和API执行异常的情况。

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