Java 并行方法调用中的异常隔离与处理

final CompletableFuture<ParseException> thrownException = new CompletableFuture<>();
Stream.of(columns).parallel().forEach(column -> {
    try {
        result[column.index] = parseColumn(valueCache[column.index], column.type);
    } catch (ParseException e) {
        // 这种方式可能导致forEach提前终止
        thrownException.complete(e); 
    }
});

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

public class ParallelTaskExecutor {

    // 假设这是需要并行执行的方法
    private static void disablePackXYZ(Long id, String requestedBy) {
        if (id % 2 != 0) { // 模拟奇数ID导致异常
            throw new RuntimeException("Failed to disable pack for ID: " + id);
        }
        System.out.println("Successfully disabled pack for ID: " + id + " by " + requestedBy);
        // 模拟耗时操作
        try {
            Thread.sleep(100); 
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // 封装单个任务，返回一个CompletableFuture，并在内部处理异常
    private CompletableFuture<TaskResult> executeDisablePackXYZAsync(Long disableId, String requestedBy, ExecutorService executor) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                disablePackXYZ(disableId, requestedBy);
                return new TaskResult(disableId, true, null); // 成功
            } catch (Exception e) {
                System.err.println("Error processing ID " + disableId + ": " + e.getMessage());
                return new TaskResult(disableId, false, e); // 失败，捕获异常
            }
        }, executor);
    }

    // 任务结果封装类
    static class TaskResult {
        Long id;
        boolean success;
        Exception exception;

        public TaskResult(Long id, boolean success, Exception exception) {
            this.id = id;
            this.success = success;
            this.exception = exception;
        }

        @Override
        public String toString() {
            return "TaskResult{" +
                   "id=" + id +
                   ", success=" + success +
                   ", exception=" + (exception != null ? exception.getMessage() : "null") +
                   '}';
        }
    }

    public List<TaskResult> disableXYZInParallel(Long rId, List<Long> disableIds, String requestedBy) {
        // 推荐使用自定义的线程池，避免ForkJoinPool的阻塞问题
        ExecutorService executor = Executors.newFixedThreadPool(Math.min(disableIds.size(), 10)); // 线程池大小可配置

        List<CompletableFuture<TaskResult>> futures = disableIds.stream()
            .map(id -> executeDisablePackXYZAsync(id, requestedBy, executor))
            .collect(Collectors.toList());

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

        try {
            // get()会阻塞直到所有future完成，但单个future的异常已被封装，不会导致此处的ExecutionException
            allOf.get(); 
        } catch (InterruptedException | ExecutionException e) {
            // 理论上，如果所有单个future都正确处理了异常并返回了TaskResult，这里不应该捕获到业务异常
            // 除非是allOf.get()本身的异常，例如线程中断。
            System.err.println("An unexpected error occurred while waiting for all tasks: " + e.getMessage());
        } finally {
            executor.shutdown(); // 关闭线程池
        }

        // 收集所有任务的结果
        List<TaskResult> results = new ArrayList<>();
        for (CompletableFuture<TaskResult> future : futures) {
            try {
                results.add(future.get()); // 获取每个任务的最终结果（成功或失败）
            } catch (InterruptedException | ExecutionException e) {
                // 这通常不应该发生，因为TaskResult已经包含了内部异常
                // 但作为防御性编程，可以处理一下，例如记录一个未知错误
                System.err.println("Could not retrieve result from a future: " + e.getMessage());
            }
        }
        return results;
    }

    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);
        String requestedBy = "system_user";
        Long rId = 123L;

        System.out.println("Starting parallel disable operations...");
        List<TaskResult> finalResults = executor.disableXYZInParallel(rId, idsToDisable, requestedBy);

        System.out.println("\n--- All tasks completed. Summary: ---");
        for (TaskResult result : finalResults) {
            System.out.println(result);
        }

        long successfulCount = finalResults.stream().filter(r -> r.success).count();
        long failedCount = finalResults.stream().filter(r -> !r.success).count();
        System.out.println("Successful tasks: " + successfulCount);
        System.out.println("Failed tasks: " + failedCount);
    }
}