Java并行方法调用中的异常处理：确保独立任务不中断整体流程

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<Long> disableIds, String requestedBy) {
        // 创建固定大小的线程池，通常建议根据CPU核心数和任务类型（IO密集/CPU密集）来设定
        // 这里使用10个线程，但实际应用中应根据负载调整
        ExecutorService executor = Executors.newFixedThreadPool(Math.min(disableIds.size(), 10));

        List<CompletableFuture<TaskOutcome>> 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<Void>。
        CompletableFuture<Void> 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(); // 恢复中断状态
            }
        }

        // 收集