Java ParallelStream 自定义线程池与I/O密集型任务优化

import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class ParallelStreamCustomPoolExample {

    // 模拟一个执行数据库查询的服务
    static class ObjectService {
        public String getParam(String field) {
            System.out.println(Thread.currentThread().getName() + " - Querying DB for: " + field);
            try {
                // 模拟数据库查询耗时，线程会在此处阻塞
                Thread.sleep(200);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("DB query interrupted", e);
            }
            return "Data_for_" + field;
        }
    }

    // 模拟待处理的对象
    static class MyObject {
        String field;
        public MyObject(String field) { this.field = field; }
        public String getField() { return field; }
    }

    /**
     * 使用自定义ForkJoinPool执行并行流任务
     * @param objects 待处理对象列表
     * @param poolSize 自定义线程池大小，即并行度
     * @return 处理结果列表
     */
    public List<String> processWithCustomParallelStream(List<MyObject> objects, int poolSize) {
        ObjectService objectService = new ObjectService(); // 实例化服务
        ForkJoinPool customPool = null;
        try {
            // 创建一个指定并行度的自定义ForkJoinPool
            customPool = new ForkJoinPool(poolSize);
            System.out.println("\n--- 使用自定义线程池 (大小: " + poolSize + ") ---");

            // 将并行流任务提交到自定义线程池中执行
            // .submit() 返回一个 Future，.get() 会阻塞直到所有任务完成
            return customPool.submit(() ->
                    objects.parallelStream()
                            .map(obj -> objectService.getParam(obj.getField())) // 这里的任务是I/O密集型的
                            .collect(Collectors.toList())
            ).get();
        } catch (InterruptedException | ExecutionException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("并行流执行失败", e);
        } finally {
            // 确保在任务完成后关闭自定义线程池，释放资源
            if (customPool != null) {
                customPool.shutdown();
            }
        }
    }

    public static void main(String[] args) {
        ParallelStreamCustomPoolExample app = new ParallelStreamCustomPoolExample();
        List<MyObject> data = List.of(
                new MyObject("Item1"), new MyObject("Item2"), new MyObject("Item3"),
                new MyObject("Item4"), new MyObject("Item5"), new MyObject("Item6"),
                new MyObject("Item7"), new MyObject("Item8"), new MyObject("Item9"),
                new MyObject("Item10")
        );

        // 示例：使用自定义线程池大小为4
        List<String> results4 = app.processWithCustomParallelStream(data, 4);
        System.out.println("\n处理完成 (池大小4): " + results4);

        // 示例：使用自定义线程池大小为2
        List<String> results2 = app.processWithCustomParallelStream(data, 2);
        System.out.println("\n处理完成 (池大小2): " + results2);
    }
}