Java并行方法调用中的异常处理：确保单个故障不中断整体流程

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<Long> disableIds, String requestedBy) {
        // 使用线程安全的集合来存储捕获到的异常
        ConcurrentLinkedQueue<Exception> caughtExceptions = new ConcurrentLinkedQueue<>();

        // 创建一系列CompletableFuture任务
        List<CompletableFuture<Void>> 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<Void> 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<Long> 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 ---");
    }
}