Java并行任务中的健壮性：独立异常处理与结果收集策略

private void disableXYZ(Long rId, List<Long> disableIds, String requestedBy) {
    for (Long disableId : disableIds) {
        try {
            disablePackXYZ(UnSubscribeRequest.unsubscriptionRequest()
                    .requestedBy(requestedBy)
                    .cancellationReason("system")
                    .id(disableId)
                    .build());
        } catch (Exception e) {
            // 捕获并记录单个任务的异常，不中断循环
            log.error("Failed to disable pack. id: {}, rId: {}. Error: {}", disableId, rId, e.getMessage());
        }
    }
}

import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

// 假设 UnSubscribeRequest 和 Log 类已定义
// ... (此处省略 UnSubscribeRequest 和 Log 的定义，参见原问题上下文)

/**
 * 封装并行任务的执行结果，可包含成功数据或异常信息。
 * @param <T> 成功时的数据类型
 */
class TaskResult<T> {
    private final T data;
    private final Throwable error;

    private TaskResult(T data, Throwable error) {
        this.data = data;
        this.error = error;
    }

    /** 创建一个成功的任务结果 */
    public static <T> TaskResult<T> success(T data) {
        return new TaskResult<>(data, null);
    }

    /** 创建一个失败的任务结果 */
    public static <T> TaskResult<T> failure(Throwable error) {
        return new TaskResult<>(null, error);
    }

    /** 判断任务是否成功 */
    public boolean isSuccess() {
        return error == null;
    }

    /** 获取成功时的数据 */
    public T getData() {
        return data;
    }

    /** 获取失败时的异常 */
    public Throwable getError() {
        return error;
    }

    @Override
    public String toString() {
        if (isSuccess()) {
            return "Success: " + data;
        } else {
            return "Failure: " + (data != null ? "ID " + data + ", " : "") + "Error: " + error.getMessage();
        }
    }
}

// 模拟日志类
class Log {
    public static void error(String format, Object... args) {
        System.err.printf(format + "%n", args);
    }
    public static 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 Builder unsubscriptionRequest() {
        return new Builder();
    }

    public static class Builder {
        private UnSubscribeRequest request = new UnSubscribeRequest();
        public Builder requestedBy(String requestedBy) { request.requestedBy = requestedBy; return this; }
        public Builder cancellationReason(String cancellationReason) { request.cancellationReason = cancellationReason; return this; }
        public Builder id(Long id) { request.id = id; return this; }
        public UnSubscribeRequest build() { return request; }
    }
}

public class ParallelTaskExecutor {

    // 使用共享的 ExecutorService 以更好地管理线程资源
    private final ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    private static final Log log = new Log();

    // 模拟要并行调用的方法，可能抛出异常
    private void disablePackXYZ(UnSubscribeRequest request) throws Exception {
        // 模拟网络请求或数据库操作
        if (request.id % 3 == 0) { // 模拟ID为3的倍数时发生失败
            throw new RuntimeException("Simulated failure for ID: " + request.id);
        }
        log.info("Successfully disabled pack for ID: %d", request.id);
        // 模拟处理时间
        Thread.sleep(100);
    }

    /**
     * 并行禁用操作，并独立处理每个任务的异常。
     * @param rId 请求ID
     * @param disableIds 要禁用的ID列表
     * @param requestedBy 请求者
     * @return 包含每个任务结果（成功或失败）的列表
     */
    public List<TaskResult<Long>> disableXYZParallel(Long rId, List<Long> disableIds, String requestedBy) {
        // 为每个ID创建一个CompletableFuture
        List<CompletableFuture<TaskResult<Long>>> futures = disableIds.stream()
            .map(disableId -> {
                UnSubscribeRequest request = UnSubscribeRequest.unsubscriptionRequest()
                    .requestedBy(requestedBy)
                    .cancellationReason("system")
                    .id(disableId)
                    .build();

                // 使用 supplyAsync 在自定义线程池中异步执行任务
                return CompletableFuture.supplyAsync(() -> {
                    try {
                        disablePackXYZ(request);
                        // 任务成功，返回成功结果
                        return TaskResult.success(disableId);
                    } catch (Exception e) {
                        // 任务失败，捕获异常并记录日志，返回失败结果
                        log.error("Failed to disable pack. id: %d, rId: %d. Error: %s", disableId, rId, e.getMessage());
                        return TaskResult.failure(e);
                    }
                }, executor); // 指定线程池
            })
            .collect(Collectors.toList());

        // 等待所有 CompletableFuture 完成
        CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

        // 在所有任务完成后，收集并返回它们的具体结果
        return allOf.thenApply(v -> futures.stream()
            .map(CompletableFuture::join) // join 不会抛出异常，因为异常已在 TaskResult 中封装
            .collect(Collectors.toList()))
            .join(); // 阻塞并获取最终结果列表
    }

    // 在应用程序关闭时，记得关闭线程池
    public void shutdown() {
        executor.shutdown();
        log.info("ExecutorService shut down.");
    }

    public static void main(String[] args) {
        ParallelTaskExecutor executor = new ParallelTaskExecutor();
        List<Long> idsToDisable = List.of(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L);
        Long rId = 12345L;
        String requestedBy = "system_user";

        System.out.println("Starting parallel disable operations...");
        List<TaskResult<Long>> results = executor.disableXYZParallel(rId, idsToDisable, requestedBy);
        System.out.println("\nAll parallel operations completed. Results:");

        // 遍历并打印每个任务的结果
        for (TaskResult<Long> result : results) {
            if (result.isSuccess()) {
                System.out.println("  " + result);
            } else {
                System.out.println("  " + result);
            }
        }

        executor.shutdown();
    }
}