如何限制含数据库查询的ParallelStream线程池大小？

我编写了一段包含数据库查询的ParallelStream代码，具体如下：

private void processParallel() {
    List<Result> = objects.parallelStream().peek(object -> {
        doSomething(object);
    }).collect(Collectors.toList());
}

private void doSomething(Object object) {
    CompletableFuture<String> param =
            CompletableFuture.supplyAsync(() -> objectService.getParam(object.getField()),
                    executor)
                    .thenApply(obj -> Optional.ofNullable(obj)
                            .map(Object::getParam)
                            .orElse(null));
}

我需要指定线程池大小，但设置参数java.util.concurrent.ForkJoinPool.common.parallelism为20并未生效，推测是流锁定导致。请问有什么方法可以限制最大线程数？

1. 用自定义ForkJoinPool托管ParallelStream执行 #

ParallelStream默认依赖公共ForkJoinPool，而公共池的并行度设置可能因JVM启动时机、其他代码提前初始化池等原因失效。直接创建自定义池包裹流执行，完全规避公共池的影响：

// 初始化指定并行度的自定义ForkJoinPool
ForkJoinPool customForkJoinPool = new ForkJoinPool(20);
try {
    List<Result> results = customForkJoinPool.submit(() -> 
        objects.parallelStream()
               .peek(this::doSomething)
               .collect(Collectors.toList())
    ).get();
} catch (InterruptedException | ExecutionException e) {
    Thread.currentThread().interrupt();
    throw new RuntimeException("并行处理失败", e);
} finally {
    customForkJoinPool.shutdown();
}

2. 确认CompletableFuture的线程池配置 #

你的代码中doSomething已使用自定义executor执行异步任务，需确保该线程池的参数符合预期：

// 示例：创建固定大小20的线程池
ExecutorService executor = Executors.newFixedThreadPool(20);
// 或者自定义ThreadPoolExecutor实现更精细控制
ExecutorService executor = new ThreadPoolExecutor(
        10, // 核心线程数
        20, // 最大线程数
        60L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(100),
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.AbortPolicy()
);

注意：ParallelStream的线程池与CompletableFuture的线程池相互独立，两者的线程数需配合调整，避免过度占用系统资源。

3. 优化异步嵌套逻辑（可选） #

当前代码存在ParallelStream并行 + CompletableFuture异步的双重线程调度，会导致实际线程数翻倍。若无需双重异步，可选择：

• 直接在ParallelStream任务中同步执行数据库查询，线程数由自定义ForkJoinPool控制；
• 弃用ParallelStream，直接用自定义线程池批量提交任务：

List<CompletableFuture<Void>> futures = objects.stream()
        .map(object -> CompletableFuture.runAsync(() -> doSomething(object), executor))
        .collect(Collectors.toList());
// 等待所有任务完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

这种方式线程数完全由executor配置决定，可控性更强。

内容的提问来源于stack exchange，提问作者cookerfresh