控制Java ParallelStream线程池大小与并发优化：策略与最佳实践

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

public class CustomParallelStreamPool {

    // 模拟一个执行数据库查询的服务
    static class ObjectService {
        public String getParam(String field) {
            // 模拟数据库查询耗时
            try {
                Thread.sleep(100); // 模拟I/O等待
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.println(Thread.currentThread().getName() + " - Fetched param for " + field);
            return "Param for " + field;
        }
    }

    static class MyObject {
        String field;
        public MyObject(String field) { this.field = field; }
        public String getField() { return field; }
    }

    private static ObjectService objectService = new ObjectService();

    /**
     * 使用自定义ForkJoinPool处理ParallelStream
     * @param objects 待处理对象列表
     * @param poolSize 自定义线程池大小
     * @return 处理结果列表
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public static List<String> processWithCustomPool(List<MyObject> objects, int poolSize)
            throws InterruptedException, ExecutionException {
        ForkJoinPool customThreadPool = null;
        try {
            // 创建一个指定并行度的ForkJoinPool
            customThreadPool = new ForkJoinPool(poolSize);

            // 将ParallelStream操作封装为Callable任务
            Callable<List<String>> task = () -> objects.parallelStream()
                    .map(object -> objectService.getParam(object.getField()))
                    .collect(Collectors.toList());

            // 提交任务并获取结果
            return customThreadPool.submit(task).get();
        } finally {
            // 关闭自定义线程池
            if (customThreadPool != null) {
                customThreadPool.shutdown();
            }
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<MyObject> data = List.of(
                new MyObject("A"), new MyObject("B"), new MyObject("C"), new MyObject("D"),
                new MyObject("E"), new MyObject("F"), new MyObject("G"), new MyObject("H"),
                new MyObject("I"), new MyObject("J")
        );

        System.out.println("--- Processing with custom pool size 4 ---");
        long startTime = System.currentTimeMillis();
        List<String> results = processWithCustomPool(data, 4);
        long endTime = System.currentTimeMillis();
        System.out.println("Results: " + results);
        System.out.println("Total time: " + (endTime - startTime) + "ms");
    }
}

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

public class ParallelStreamWithCompletableFuture {

    static class ObjectService {
        public String getParam(String field) {
            try {
                Thread.sleep(100); // 模拟I/O等待
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.println(Thread.currentThread().getName() + " - Fetched param for " + field);
            return "Param for " + field;
        }
    }

    static class MyObject {
        String field;
        public MyObject(String field) { this.field = field; }
        public String getField() { return field; }
    }

    private static ObjectService objectService = new ObjectService();
    // 建议使用有限的线程池处理I/O，其大小应与数据库连接池大小匹配
    private static ExecutorService ioExecutor = Executors.newFixedThreadPool(5); // 示例：假设数据库连接池最大为5

    /**
     * 使用ParallelStream结合CompletableFuture和专用I/O执行器处理异步I/O任务
     * @param objects 待处理对象列表
     * @return 处理结果列表
     */
    public static List<String> processParallelWithAsyncIO(List<MyObject> objects) {
        // ParallelStream用于快速提交CompletableFuture任务
        List<CompletableFuture<String>> futures = objects.parallelStream()
                .map(object -> CompletableFuture.supplyAsync(() -> objectService.getParam(object.getField()), ioExecutor)
                        .thenApply(param -> Optional.ofNullable(param).orElse("N/A")))
                .collect(Collectors.toList());

        // 阻塞等待所有CompletableFuture完成，并收集结果
        return futures.stream()
                .map(CompletableFuture::join) // join()会阻塞直到CompletableFuture完成
                .collect(Collectors.toList());
    }

    public static void main(String[] args) {
        List<MyObject> data = List.of(
                new MyObject("A"), new MyObject("B"), new MyObject("C"), new MyObject("D"),
                new MyObject("E"), new MyObject("F"), new MyObject("G"), new MyObject("H"),
                new MyObject("I"), new MyObject("J")
        );

        System.out.println("--- Processing with ParallelStream and async I/O ---");
        long startTime = System.currentTimeMillis();
        List<String> results = processParallelWithAsyncIO(data);
        long endTime = System.currentTimeMillis();
        System.out.println("Results: " + results);
        System.out.println("Total time: " + (endTime - startTime) + "ms");

        // 关闭I/O执行器
        ioExecutor.shutdown();
    }
}