如何为 Java 8 并行流指定 ForkJoinPool？答案

【问题标题】：How to specify ForkJoinPool for Java 8 parallel stream?如何为 Java 8 并行流指定 ForkJoinPool？
【发布时间】：2018-09-12 04:55:52
【问题描述】：

据我所知，并行流使用默认的ForkJoinPool.commonPool，默认情况下它的线程数比您的处理器少一个。我想使用自己的自定义线程池。

像这样：

@Test
public void stream() throws Exception {
    //System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");
    ForkJoinPool pool = new ForkJoinPool(10);
    List<Integer> testList = Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20);
    long start = System.currentTimeMillis();
    List<Integer> result = pool.submit(() -> testList.parallelStream().map(item -> {
        try {
            // read from database
            Thread.sleep(1000);
            System.out.println("task" + item + ":" + Thread.currentThread());
        } catch (Exception e) {
        }
        return item * 10;
    })).get().collect(Collectors.toList());
    System.out.println(result);
    System.out.println(System.currentTimeMillis() - start);
}

结果：

我的自定义 ForkJoinPool 从未使用过。我像这样更改默认并行度：

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");

效果很好 - 任务只需大约 1 秒。

在我的应用程序中，任务包含繁重的 IO 操作（从 db 读取数据）。所以我需要更高的并行度，但我不想更改 JVM 属性。

那么指定我自己的ForkJoinPool 的正确方法是什么？

或者如何在 IO 密集型的情况下使用并行流？

【问题讨论】：

标签： java-8 java-stream

【解决方案1】：

流是惰性的；当您开始终端操作时，所有工作都已完成。在您的情况下，终端操作是.collect(Collectors.toList())，您在main 线程中调用get() 的结果。因此，实际工作的完成方式与您在 main 线程中构建整个流的方式相同。

为了让你的池子生效，你必须将终端操作移到提交的任务中：

ForkJoinPool pool = new ForkJoinPool(10);
List<Integer> testList = Arrays.asList(
    1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20);
long start = System.currentTimeMillis();
List<Integer> result = pool.submit(() -> testList.parallelStream().map(item -> {
    try {
        // read from database
        Thread.sleep(1000);
        System.out.println("task" + item + ":" + Thread.currentThread());
    } catch (InterruptedException e) {}
    return item * 10;
}).collect(Collectors.toList())).join();
System.out.println(result);
System.out.println(System.currentTimeMillis() - start);

我们也可以通过在main线程中构造流，只将终端操作提交到池中来演示终端操作的相关性：

Stream<Integer> stream = testList.parallelStream().map(item -> {
    try {
        // read from database
        Thread.sleep(1000);
        System.out.println("task" + item + ":" + Thread.currentThread());
    } catch (InterruptedException e) {}
    return item * 10;
});
List<Integer> result = pool.submit(() -> stream.collect(Collectors.toList())).join();

但您应该记住，这是未记录的行为，无法保证。实际的答案肯定是当前形式的 Stream API 没有线程控制（也没有帮助处理检查的异常），不适合并行 I/O 操作。

【讨论】：

【解决方案2】：

我假设你发现了这里描述的技巧：

Custom thread pool in Java 8 parallel stream

哪些状态

技巧基于ForkJoinTask.fork，它指定：“安排到在当前任务所在的池中异步执行此任务运行，如果适用，或使用ForkJoinPool.commonPool() if 不是inForkJoinPool()"

在您的代码中，parallelStream() 和 map(...) 在自定义 ForkJoinPool 中被调用，但传递给 map 的 Function 不是。

记住Stream#map是一个中间操作。它的Function 只会在终端操作被链接后对其元素执行。在您的情况下，该终端操作是collect(...)。由于collect(Collectors.toList() 在main 线程中被调用，map 的Function 在commonPool 中的每个元素上并行调用。

您可以简单地将collect(...) 调用移动到您的submit(...) 中。

List<Integer> result = pool.submit(() -> testList.parallelStream().map(item -> {
    try {
        // read from database
        Thread.sleep(1000);
        System.out.println("task" + item + ":" + Thread.currentThread());
    } catch (Exception e) {
    }
    return item * 10;
}).collect(Collectors.toList())).get();

【讨论】：