如何（全局）替换Java并行流的公共线程池后端？

Question

我希望全局替换Java并行流默认使用的公共线程池，例如

IntStream.range(0,100).parallel().forEach(i -> {
    doWork();
});

我知道，通过将这样的指令提交到专用线程池（参见Java8并行流中的自定义线程池），可以使用专用的ForkJoinPool。这里的问题是

• 是否可以用其他实现（例如executors.NewFixedThreadPool(10)？
来替换常见的ForkJoinPool
• 是否可以通过某些全局设置（例如，某些JVM属性）来实现？

备注：我喜欢替换f/j池的原因是，它似乎有一个bug，使它不能用于嵌套的并行循环。

嵌套并行循环性能较差，可能导致死锁，请参见http://christian-fries.de/blog/files/2014-nested-java-8-Paralle-foreach.html

例如：以下代码导致死锁：

// Outer loop
IntStream.range(0,24).parallel().forEach(i -> {

    // (omitted:) do some heavy work here (consuming majority of time)

    // Need to synchronize for a small "subtask" (e.g. updating a result)
    synchronized(this) {
        // Inner loop (does s.th. completely free of side-effects, i.e. expected to work)
        IntStream.range(0,100).parallel().forEach(j -> {
            // do work here
        });
    }
});

我的问题是如何取代FJP。如果您想讨论嵌套的并行循环，您可以检查嵌套的Java8并行forEach循环Perforeach Poor。这种行为在意料之中吗？.

Answer 1

我认为这不是stream API的使用方式。您似乎（错误地）使用它来简单地执行并行任务（专注于任务，而不是数据），而不是进行并行流处理（专注于流中的数据）。您的代码在某种程度上违反了流的一些主要原则。（我写的是“莫名其妙”，因为它不是真的被禁止，而是被劝阻）：避免状态和副作用。

除此之外（或者可能是因为副作用），您在外部循环中使用了重度同步，这是无害的！

虽然文档中没有提到，但并行流在内部使用了通用的forkjoinpool。不管这是不是缺少文件，我们必须接受这一事实。forkjointask的JavaDoc声明：

定义和使用可能阻塞的ForkJoinTasks是可能的，但要做到这一点需要三个进一步的考虑：（1）如果任何其他任务都依赖于一个阻塞外部同步或I/O的任务，则完成很少的任务。从不联接的事件样式异步任务（例如，那些子类化CountedCompleter的任务）经常属于这一类。（2）最小化资源影响，任务要小；理想情况下只执行（可能的）阻塞操作。（3）除非使用ForkJoinPool.ManagedBlocker API，或者已知可能被阻塞的任务数少于池的ForkJoinPool.GetParallelism级别，否则池无法保证将有足够的线程来确保进度或良好的性能。

同样，您似乎在使用流来替换简单的for-loop和executor服务。

• 如果只想并行执行n任务，请使用executionservice
• 如果有一个任务创建子任务的更复杂示例，请考虑改用forkjoinpool（与forkjoinasks)。（它确保线程数量恒定，不会因为等待其他线程完成的任务太多而出现死锁的危险，因为等待的任务不会阻塞它们正在执行的线程）。
• 如果要（并行）处理数据，请考虑使用流API。
• 不能“安装”自定义公用池。它是在内部私有静态代码中创建的。
• 但是您可以使用某些系统属性（请参见ForkJoinPool的JavaDoc)
来影响公共池的并行性、线程工厂和异常处理程序

不要混淆executionservice和forkjoinpool。他们（通常）不能互相替代！