Java并行方法调用中的容错处理：确保独立执行与错误记录

import java.util.concurrent.*;
import java.util.List;
import java.util.ArrayList;
import java.util.stream.Collectors;

// 模拟的日志工具
class LoggerFactory {
    public static Logger getLogger(Class<?> clazz) {
        return new Logger();
    }
}

class Logger {
    public void info(String format, Object... args) {
        System.out.println("INFO: " + String.format(format, args));
    }
    public void error(String format, Object... args) {
        System.err.println("ERROR: " + String.format(format, args));
    }
}

// 假设的UnSubscribeRequest类
class UnSubscribeRequest {
    private String requestedBy;
    private String cancellationReason;
    private Long id;

    private UnSubscribeRequest() {}

    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;
    }

    public Long getId() { return id; }
}

/**
 * 封装单个并行任务的执行结果
 */
class TaskExecutionResult {
    private final Long taskId;
    private final boolean success;
    private final Throwable error; // 存储任务执行过程中捕获的异常

    public TaskExecutionResult(Long taskId, boolean success, Throwable error) {
        this.taskId = taskId;
        this.success = success;
        this.error = error;
    }

    public Long getTaskId() { return taskId; }
    public boolean isSuccess() { return success; }
    public Throwable getError() { return error; }

    @Override
    public String toString() {
        return "TaskExecutionResult{" +
               "taskId=" + taskId +
               ", success=" + success +
               ", error=" + (error != null ? error.getMessage() : "null") +
               '}';
    }
}

class PackService {
    private static final Logger log = LoggerFactory.getLogger(PackService.class);

    // 模拟的原始禁用方法，可能会抛出异常
    public void disablePackXYZ(UnSubscribeRequest request) throws Exception {
        // 模拟偶数ID禁用失败的情况
        if (request.getId() % 2 == 0) {
            throw new RuntimeException("Simulated failure to disable pack for ID: " + request.getId());
        }
        log.info("Successfully disabled pack for ID: {}", request.getId());
        // 实际的禁用逻辑
    }

    /**
     * 并行禁用XYZ包，并容错处理
     *
     * @param rId         请求ID
     * @param disableIds  需要禁用的ID列表
     * @param requestedBy 请求者
     */
    public void disableXYZParallel(Long rId, List<Long> disableIds, String requestedBy) {
        // 推荐使用自定义的ExecutorService，以便更好地控制线程资源
        // 线程池大小可根据实际CPU核心数和任务类型调整
        ExecutorService executor = Executors.newFixedThreadPool(
            Math.min(disableIds.size(), Runtime.getRuntime().availableProcessors() * 2)
        );

        List<CompletableFuture<TaskExecutionResult>> futures = disableIds.stream()
            .map(disableId -> CompletableFuture.supplyAsync(() -> {
                // 在每个CompletableFuture内部捕获异常
                try {
                    disablePackXYZ(UnSubscribeRequest.unsubscriptionRequest()
                            .requestedBy(requestedBy)
                            .cancellationReason("system")
                            .id(disableId)
                            .build());
                    // 任务成功，返回成功结果
                    return new TaskExecutionResult(disableId, true, null);
                } catch (Exception e) {
                    // 任务失败，记录错误并返回失败结果，不向上抛出
                    log.error("Failed to disable pack. id: {}, rId: {}. Error: {}", disableId, rId, e.getMessage());
                    return new TaskExecutionResult(disableId, false, e);
                }
            }, executor)) // 指定使用自定义的Executor
            .collect(Collectors.toList());

        // 使用CompletableFuture.allOf等待所有并行任务完成
        // allOf.join() 只有在某个future本身没有处理异常并向上抛出时才会抛出CompletionException
        // 由于我们在supplyAsync内部已经捕获并处理了异常，这里通常不会抛出异常
        CompletableFuture<Void> allTasks = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

        try {
            allTasks.join(); // 阻塞等待所有任务完成
        } catch (CompletionException e) {
            // 理论上这里不会捕获到业务异常，因为业务异常已在任务内部处理
            // 除非是CompletableFuture自身运行环境的非预期错误
            log.error("An unexpected error occurred while waiting for all tasks to complete: {}", e.getMessage());
        }

        // 收集并处理所有任务的最终结果
        List<TaskExecutionResult> finalResults = futures.stream()
            .map(CompletableFuture::join) // 获取每个CompletableFuture的最终结果
            .collect(Collectors.toList());

        log.info("Parallel disable operations completed. Summary for rId {}:", rId);
        for (TaskExecutionResult result : finalResults) {
            if (result.isSuccess()) {
                log.info("ID {} processed successfully.", result.getTaskId());
            } else {
                log.error("ID {} failed with error: {}", result.getTaskId(), result.getError().getMessage());
            }
        }

        // 关闭ExecutorService，释放资源
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow(); // 强制关闭未完成的任务
            }
        } catch (InterruptedException ex) {
            executor.shutdownNow();
            Thread.currentThread().interrupt(); // 恢复中断状态
        }
    }

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

        service.disableXYZParallel(requestId, idsToDisable, requestedBy);
    }
}