引言：并行任务中的异常挑战
在现代软件开发中，为了提高系统吞吐量和响应速度，并行处理已成为常用技术。然而，当多个任务并发执行时，如何优雅地处理其中某个任务抛出的异常，同时不中断其他任务的正常执行，是一个常见且关键的挑战。传统上，如果一个线程或任务抛出未捕获的异常，它可能会导致整个进程或批处理操作中止，这在需要处理大量独立单元（如批量数据处理、多服务调用）的场景中是不可接受的。例如，在一个需要并行禁用多个系统组件的场景中，我们希望即使某个组件禁用失败，也不影响其他组件的禁用尝试。
传统并行流与异常传播的局限性
Java 8引入的流API和CompletableFuture极大地简化了并行编程。然而，在使用这些工具时，如果不当处理异常，仍可能遇到中断整体流程的问题。例如，当使用Stream.parallel().forEach()并尝试在内部通过CompletableFuture.complete(e)立即传播异常时，forEach的语义可能导致一旦有异常发生，整个流操作就会提前终止，而不等待所有并行任务的完成。这意味着其他尚未执行或正在执行的任务可能被中断，无法达到我们期望的“即使部分失败，整体也继续”的目标。
解决方案：独立异常捕获与收集
核心思想是，将每个并行任务的执行结果（包括成功的数据或发生的异常）封装起来，而不是让异常立即向上抛出并中断流程。这意味着每个任务在其内部捕获并处理自己的异常，然后将异常作为其“结果”的一部分返回。最终，所有任务完成后，我们可以统一收集并检查每个任务的执行结果，包括哪些任务成功，哪些任务失败以及失败的原因。
1. 基于 CompletableFuture 和 ExecutorService 的方案
CompletableFuture 结合 ExecutorService 提供了对并行任务执行和结果收集的强大控制。这种方法允许我们自定义线程池，并灵活地处理每个异步任务的成功或失败状态。
实现步骤：
- 定义任务结果封装类： 创建一个简单的类来封装每个任务的执行结果，该类可以包含任务ID、成功状态、实际结果数据或捕获到的异常。
- 使用 ExecutorService 提交任务： 创建一个 ExecutorService 来管理线程池，然后使用 CompletableFuture.supplyAsync() 将每个任务提交到该线程池。
- 在任务内部捕获异常： 在 supplyAsync 提供的 Supplier lambda 表达式内部，使用 try-catch 块捕获所有可能发生的异常。捕获到异常后，将其封装到结果封装类中并返回，而不是重新抛出。
- 等待所有任务完成并收集结果： 使用 CompletableFuture.allOf() 等待所有提交的 CompletableFuture 完成，然后通过遍历这些 CompletableFuture 并调用 join() 方法来获取每个任务的最终结果（即封装好的任务结果对象）。
- 统一处理结果和异常： 在所有任务完成后，遍历收集到的结果列表，区分成功和失败的任务，并进行相应的处理（如日志记录、错误报告等）。
- 关闭 ExecutorService： 确保在所有任务完成后，优雅地关闭 ExecutorService 以释放资源。
示例代码：
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
// 模拟 UnSubscribeRequest 类
class UnSubscribeRequest {
private Long id;
private String requestedBy;
private String cancellationReason;
// 构造器和Builder模式简化创建
private UnSubscribeRequest(Builder builder) {
this.id = builder.id;
this.requestedBy = builder.requestedBy;
this.cancellationReason = builder.cancellationReason;
}
public Long getId() { return id; }
public String getRequestedBy() { return requestedBy; }
public String getCancellationReason() { return cancellationReason; }
public static Builder unsubscriptionRequest() {
return new Builder();
}
public static class Builder {
private Long id;
private String requestedBy;
private String cancellationReason;
public Builder id(Long id) { this.id = id; return this; }
public Builder requestedBy(String requestedBy) { this.requestedBy = requestedBy; return this; }
public Builder cancellationReason(String cancellationReason) { this.cancellationReason = cancellationReason; return this; }
public UnSubscribeRequest build() { return new UnSubscribeRequest(this); }
}
}
// 封装任务执行结果的类
class TaskOutcome {
private final Long id;
private final boolean success;
private final Exception exception;
private final String message;
public static TaskOutcome success(Long id, String message) {
return new TaskOutcome(id, true, null, message);
}
public static TaskOutcome failure(Long id, Exception exception) {
return new TaskOutcome(id, false, exception, exception.getMessage());
}
private TaskOutcome(Long id, boolean success, Exception exception, String message) {
this.id = id;
this.success = success;
this.exception = exception;
this.message = message;
}
public Long getId() { return id; }
public boolean isSuccess() { return success; }
public Exception getException() { return exception; }
public String getMessage() { return message; }
@Override
public String toString() {
return "TaskOutcome{" +
"id=" + id +
", success=" + success +
", message='" + message + '\'' +
(exception != null ? ", exceptionType=" + exception.getClass().getSimpleName() : "") +
'}';
}
}
// 模拟日志工具
class Logger {
public void error(String format, Object... args) { System.err.printf("[ERROR] " + format + "%n", args); }
public void info(String format, Object... args) { System.out.printf("[INFO] " + format + "%n", args); }
}
public class ParallelMethodExecutor {
private static final Logger log = new Logger();
// 模拟原始的 disablePackXYZ 方法，可能抛出异常
private static void disablePackXYZ(UnSubscribeRequest request) throws Exception {
Long id = request.getId();
if (id % 2 != 0) { // 模拟奇数ID禁用失败
throw new RuntimeException("Simulated failure for ID: " + id);
}
// 模拟耗时操作
Thread.sleep(50);
log.info("Successfully disabled pack for ID: %d", id);
}
/**
* 并行执行禁用操作，确保单个任务失败不中断整体流程。
* @param rId 请求ID
* @param disableIds 待禁用的ID列表
* @param requestedBy 请求者
*/
public static void executeParallelDisable(Long rId, List disableIds, String requestedBy) {
// 创建固定大小的线程池，通常建议根据CPU核心数和任务类型（IO密集/CPU密集）来设定
// 这里使用10个线程，但实际应用中应根据负载调整
ExecutorService executor = Executors.newFixedThreadPool(Math.min(disableIds.size(), 10));
List> futures = disableIds.stream()
.map(disableId -> CompletableFuture.supplyAsync(() -> {
try {
// 调用实际的业务方法
disablePackXYZ(UnSubscribeRequest.unsubscriptionRequest()
.requestedBy(requestedBy)
.cancellationReason("system")
.id(disableId)
.build());
return TaskOutcome.success(disableId, "Pack disabled successfully.");
} catch (Exception e) {
// 捕获异常，并将其封装到 TaskOutcome 中返回
log.error("Failed to disable pack. id: %d, rId: %d. Error: %s", disableId, rId, e.getMessage());
return TaskOutcome.failure(disableId, e);
}
}, executor)) // 指定使用自定义的线程池
.collect(Collectors.toList());
// 等待所有 CompletableFuture 完成。
// allOf() 不会抛出单个任务的异常，它只会在所有任务都完成（无论成功或失败）后返回一个 CompletableFuture。
CompletableFuture allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
try {
allOf.join(); // 阻塞等待所有任务完成。如果单个任务内部已捕获异常，这里不会抛出。
} catch (Exception e) {
// 只有当 allOf 本身出现问题（如编程错误）时，才会进入此catch块
log.error("An unexpected error occurred while waiting for all tasks: %s", e.getMessage());
} finally {
// 确保线程池被关闭
executor.shutdown();
try {
if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { // 等待60秒
executor.shutdownNow(); // 强制关闭
}
} catch (InterruptedException ex) {
executor.shutdownNow();
Thread.currentThread().interrupt(); // 恢复中断状态
}
}
// 收集
