ParallelStream池大小

Question

我有一个并行流，里面有一些数据库查询，如下所示：

private void processParallel() {
List\<Result\> = objects.parallelStream().peek(object -\> {
doSomething(object)
}
}

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

我需要指定池大小，但是设置属性"java.util.concurrent.ForkJoinPool.common.parallelism"，"20“不起作用，可能是因为锁定了流。有没有办法限制线程的最大数量？

Answer 1

由于并行流在幕后使用Fork/Join框架，为了限制流所使用的踏板数量，您可以使用Callable包装流，并定义具有所述所需并行级别的ForkJoinPool。

并行流占用的线程将从新创建的ForkJoinPool中提取，可调用的任务是按照here的描述提交的(而不是从普通轮询中提交的)。

这种方法的缺点是，您依赖于Stream的实现细节。

而且，正如@Louis在comment 中指出的那样，您可能需要另一种方法来限制流中使用的线程数量。因为您正在执行数据库查询，因此每个处理步骤都需要一个数据库连接来完成其工作，因此线程的数量不应该大于数据源目前可以提供的可用连接数。如果您有多个进程可以同时触发这些异步任务(例如，在web应用程序中)，那么尝试开发自己的解决方案似乎不是一个好主意。如果是这样的话，您可以考虑使用像Spring这样的框架。