Java并行调用异常处理技巧

在Java并行编程中，当需要同时执行多个独立任务时，确保其中一个或多个任务的失败不会导致整个批处理过程中止至关重要。本文将探讨如何在利用CompletableFuture进行并行方法调用的同时，优雅地捕获并收集异常，从而实现即使部分任务失败也能保证所有任务尝试执行完毕，并在事后统一处理或报告所有错误。

1. 问题背景与挑战

在传统的迭代式处理中，如果一个任务抛出异常，通常可以通过try-catch块来捕获并继续下一个任务的执行。然而，当我们将处理方式转换为并行模式时，例如使用Java Stream API的parallel()或CompletableFuture，异常处理的策略需要重新考量。

一个常见的误区是，在并行流的forEach操作中，如果某个任务内部捕获到异常并尝试通过共享的CompletableFuture来完成异常状态（例如thrownException.complete(e)），这可能会导致流的提前终止。因为一旦CompletableFuture被标记为异常完成，后续对该CompletableFuture的等待或组合操作可能会立即抛出异常，从而中断整个并行批处理，阻止其他尚未完成的任务继续执行。

我们的目标是，即使在并行执行的某个disablePackXYZ调用中发生异常，也不应中断其他disablePackXYZ调用的执行，而是让所有任务尽可能地完成，并在最后汇总所有成功和失败的结果（包括捕获到的异常）。

2. 解决方案：基于CompletableFuture的异常收集机制

为了实现“失败不中断整体流程”的目标，核心思想是在每个并行任务内部独立捕获并处理异常，而不是将其传播出去导致外部流程中断。具体来说，我们不让CompletableFuture本身因内部异常而以“异常”状态完成，而是让它以“正常”状态完成，但同时将内部捕获的异常存储到一个共享的、线程安全的集合中。

2.1 核心思路

1. 为每个并行任务创建独立的CompletableFuture。 使用CompletableFuture.runAsync()（对于无返回值任务）或CompletableFuture.supplyAsync()（对于有返回值任务）来包装每个待执行的方法调用。
2. 在每个任务内部使用try-catch块。 这是关键步骤。在CompletableFuture的执行体（lambda表达式）内部，对可能抛出异常的代码块进行try-catch。
3. 收集异常而不是传播。 当捕获到异常时，不要将其重新抛出或用于使外部的CompletableFuture以异常状态完成。相反，将这个异常实例添加到一个预先定义的、线程安全的异常集合中。
4. 等待所有任务完成。 使用CompletableFuture.allOf()来等待所有独立的CompletableFuture实例完成。由于内部异常已被捕获并收集，这些CompletableFuture将以正常状态完成，因此allOf().join()不会抛出异常（除非有未捕获的运行时异常或取消）。
5. 事后处理收集到的异常。 在allOf().join()返回后，检查异常集合。如果集合不为空，则表示有任务失败，可以统一记录日志、生成报告或抛出一个包含所有子异常的复合异常。

2.2 示例代码

以下是将原有的迭代式disableXYZ方法改造为并行且容错的实现：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.Collectors;

// 模拟日志工具
class Logger {
    public void error(String format, Object... args) {
        System.err.printf(format + "%n", args);
    }
    public void info(String format, Object... args) {
        System.out.printf(format + "%n", args);
    }
}

// 模拟UnSubscribeRequest类
class UnSubscribeRequest {
    private String requestedBy;
    private String cancellationReason;
    private Long id;

    public static UnSubscribeRequest unsubscriptionRequest() {
        return new UnSubscribeRequest();
    }

    public UnSubscribeRequest requestedBy(String requestedBy) {
        this.requestedBy = requestedBy;
        return this;
    }

    public UnSubscribeRequest cancellationReason(String cancellationReason) {
        this.cancellationReason = cancellationReason;
        return this;
    }

    public UnSubscribeRequest id(Long id) {
        this.id = id;
        return this;
    }

    @Override
    public String toString() {
        return "UnSubscribeRequest [id=" + id + ", requestedBy=" + requestedBy + "]";
    }
}

public class ParallelSafeExecutor {

    private static final Logger log = new Logger();

    // 模拟待并行执行的方法，可能抛出异常
    private void disablePackXYZ(UnSubscribeRequest request) throws Exception {
        // 模拟业务逻辑，例如：ID为奇数时模拟失败
        if (request.id % 2 != 0) {
            throw new RuntimeException("Simulated failure for ID: " + request.id);
        }
        log.info("Successfully disabled pack for ID: " + request.id);
    }

    /**
     * 并行执行多个 disablePackXYZ 调用，并收集所有异常，不中断整体流程。
     *
     * @param rId 相关ID
     * @param disableIds 待禁用ID列表
     * @param requestedBy 请求者
     */
    public void disableXYZParallelSafe(Long rId, List disableIds, String requestedBy) {
        // 使用线程安全的集合来存储捕获到的异常
        ConcurrentLinkedQueue caughtExceptions = new ConcurrentLinkedQueue<>();

        // 创建一系列CompletableFuture任务
        List> futures = disableIds.stream()
                .map(disableId -> CompletableFuture.runAsync(() -> {
                    try {
                        // 执行核心业务逻辑
                        disablePackXYZ(UnSubscribeRequest.unsubscriptionRequest()
                                .requestedBy(requestedBy)
                                .cancellationReason("system")
                                .id(disableId)
                                .build());
                    } catch (Exception e) {
                        // 捕获异常，并将其添加到线程安全的异常集合中
                        log.error("Failed to disable pack (async). id: {}, rId: {}. Error: {}", disableId, rId, e.getMessage());
                        caughtExceptions.add(e); // 关键：收集异常，而不是重新抛出
                    }
                }))
                .collect(Collectors.toList());

        // 等待所有CompletableFuture任务完成
        // CompletableFuture.allOf() 会创建一个新的 CompletableFuture，
        // 当所有给定的 CompletableFuture 都完成时，它也会完成。
        // 如果内部的 CompletableFuture 已经捕获并处理了异常，
        // 那么它们将以正常状态完成，allOf().join() 不会抛出异常。
        CompletableFuture allOfTasks = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

        try {
            allOfTasks.join(); // 阻塞直到所有任务完成
            log.info("All parallel tasks have completed their execution attempts.");
        } catch (Exception e) {
            // 这个catch块只会在 CompletableFuture.allOf() 本身因某种原因（如任务被取消或未捕获的运行时错误）
            // 导致异常完成时触发，而不是由 disablePackXYZ 内部捕获的异常触发。
            log.error("An unexpected error occurred while waiting for all tasks to complete: {}", e.getMessage());
        }

        // 检查并处理所有收集到的异常
        if (!caughtExceptions.isEmpty()) {
            log.error("Parallel processing finished with {} failures. Details:", caughtExceptions.size());
            caughtExceptions.forEach(e -> log.error("  - {}", e.getMessage()));
            // 根据业务需求，可以在这里抛出一个包含所有子异常的复合异常
            // 例如：throw new BatchProcessingException("Some tasks failed", new ArrayList<>(caughtExceptions));
        } else {
            log.info("All parallel tasks completed successfully without any reported failures.");
        }
    }

    public static void main(String[] args) {
        ParallelSafeExecutor executor = new ParallelSafeExecutor();
        List idsToDisable = List.of(1L, 2L, 3L, 4L, 5L, 6L, 7L); // 包含奇数和偶数ID
        Long requestId = 1001L;
        String requestedBy = "systemAdmin";

        log.info("--- Starting parallel safe execution ---");
        executor.disableXYZParallelSafe(requestId, idsToDisable, requestedBy);
        log.info("--- Parallel safe execution finished ---");
    }
}

2.3 代码解释

• ConcurrentLinkedQueue caughtExceptions: 选择ConcurrentLinkedQueue是因为它是一个线程安全的队列，适合在多个线程并发写入时使用，而不需要额外的同步机制（如synchronized或ReentrantLock）。如果需要按索引访问或排序，也可以考虑CopyOnWriteArrayList，但对于简单的错误收集，ConcurrentLinkedQueue通常更高效。
• CompletableFuture.runAsync(() -> { ... }): 为每个disableId创建一个独立的异步任务。默认情况下，runAsync会使用ForkJoinPool.commonPool()来执行任务。如果需要更精细的线程池控制，可以传入自定义的Executor。
• try { disablePackXYZ(...) } catch (Exception e) { caughtExceptions.add(e); }: 这是实现容错的关键。在每个任务内部，任何由disablePackXYZ抛出的异常都会被捕获。捕获后，异常被添加到caughtExceptions队列中，而不是向上层传播。这意味着即使disablePackXYZ失败，其对应的CompletableFuture也会以正常状态完成（因为它内部的lambda执行体没有抛出未捕获的异常）。
• CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])): 这个方法接收一组CompletableFuture，并返回一个新的CompletableFuture，当且仅当所有输入的CompletableFuture都完成时，它才完成。
• allOfTasks.join(): 阻塞当前线程，直到allOfTasks完成。由于我们已经处理了内部异常，所以join()方法不会抛出任何由disablePackXYZ引起的异常。它只会等待所有并行任务的执行完成。

3. 注意事项与最佳实践

• 线程安全集合的选择: 根据实际需求选择合适的线程安全集合。ConcurrentLinkedQueue适用于简单的添加操作，CopyOnWriteArrayList适用于读多写少且需要迭代的场景，而Collections.synchronizedList()或Collections.synchronizedSet()则提供了同步包装器。

• 自定义线程池: CompletableFuture.runAsync()和supplyAsync()默认使用ForkJoinPool.commonPool()。对于I/O密集型任务或需要特定线程管理策略的场景，建议使用自定义的ExecutorService来避免阻塞公共线程池或资源耗尽。

  // 例如，使用固定大小的线程池
  ExecutorService customExecutor = Executors.newFixedThreadPool(10);
  // ...
  CompletableFuture.runAsync(() -> { /* task */ }, customExecutor);
  // ...
  // 记得在应用关闭时关闭线程池
  // customExecutor.shutdown();

• 结果与异常的关联: 在上述示例中，我们只收集了异常。如果每个并行任务除了可能抛出异常外，还有返回值，并且需要将返回值与对应的任务ID或异常关联起来，可以考虑使用CompletableFuture.supplyAsync()并返回一个包含结果或异常的自定义包装类，或者使用CompletableFuture.handle()来处理结果和异常。

  // 示例：返回结果或异常
  class TaskResult {
      Long id;
      Object result; // 实际业务结果
      Exception error; // 如果有错误
  
      public static TaskResult success(Long id, Object result) { /* ... */ }
      public static TaskResult failure(Long id, Exception error) { /* ... */ }
  }
  
  // ...
  List> futures = disableIds.stream()
      .map(disableId -> CompletableFuture.supplyAsync(() -> {
          try {
              // ... disablePackXYZ logic ...
              return TaskResult.success(disableId, "Success message");
          } catch (Exception e) {
              return TaskResult.failure(disableId, e);
          }
      }))
      .collect(Collectors.toList());
  
  CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
  
  List allResults = futures.stream()
      .map(CompletableFuture::join) // 获取每个任务的TaskResult
      .collect(Collectors.toList());
  
  // 遍历allResults，区分成功和失败

• 日志记录: 及时、详细地记录每个并行任务的成功与失败状态，对于调试和问题排查至关重要。

• 复合异常: 如果需要在所有并行任务完成后，将所有捕获到的异常作为一个整体向上层抛出，可以创建一个自定义的复合异常类，并在其中包含所有子异常。

4. 总结

通过在每个并行任务内部进行异常捕获和收集，并利用CompletableFuture.allOf()等待所有任务完成，我们能够构建出健壮且容错的并行处理流程。这种模式确保了即使在分布式或高并发环境中，单个组件的故障也不会导致整个批处理过程的中断，从而提高了系统的可用性和稳定性。这种“失败不中断，事后统一处理”的策略在处理大量独立且可能失败的任务时尤为有效。