Java 8并行流中的自定义线程池

Question

是否可以为Java 8 并行流指定自定义线程池？我找不到任何地方.

想象一下,我有一个服务器应用程序,我想使用并行流.但是应用程序很大且是多线程的,因此我想将它划分为区分.我不想在另一个模块的应用程序块任务的一个模块中执行缓慢的任务.

如果我不能为不同的模块使用不同的线程池,这意味着我无法在大多数现实情况下安全地使用并行流.

请尝试以下示例.在单独的线程中执行一些CPU密集型任务.这些任务利用并行流.第一个任务被破坏,因此每个步骤需要1秒(通过线程休眠模拟).问题是其他线程卡住并等待损坏的任务完成.这是一个人为的例子,但想象一下servlet应用程序和有人向共享fork连接池提交长时间运行的任务.

public class ParallelTest {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService es = Executors.newCachedThreadPool();

        es.execute(() -> runTask(1000)); //incorrect task
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));


        es.shutdown();
        es.awaitTermination(60, TimeUnit.SECONDS);
    }

    private static void runTask(int delay) {
        range(1, 1_000_000).parallel().filter(ParallelTest::isPrime).peek(i -> Utils.sleep(delay)).max()
                .ifPresent(max -> System.out.println(Thread.currentThread() + " " + max));
    }

    public static boolean isPrime(long n) {
        return n > 1 && rangeClosed(2, (long) sqrt(n)).noneMatch(divisor -> n % divisor == 0);
    }
}

Answer 1

实际上有一个技巧如何在特定的fork-join池中执行并行操作.如果将其作为fork-join池中的任务执行,它将保留在那里并且不使用公共任务.

final int parallelism = 4;
ForkJoinPool forkJoinPool = null;
try {
    forkJoinPool = new ForkJoinPool(parallelism);
    final List<Integer> primes = forkJoinPool.submit(() ->
        // Parallel task here, for example
        IntStream.range(1, 1_000_000).parallel()
                .filter(PrimesPrint::isPrime)
                .boxed().collect(Collectors.toList())
    ).get();
    System.out.println(primes);
} catch (InterruptedException | ExecutionException e) {
    throw new RuntimeException(e);
} finally {
    if (forkJoinPool != null) {
        forkJoinPool.shutdown();
    }
}

诀窍是基于ForkJoinTask.fork,它指定:"安排在当前任务运行的池中异步执行此任务(如果适用),或者如果不是inForkJoinPool()则使用ForkJoinPool.commonPool()"

Answer 2

并行流使用默认值ForkJoinPool.commonPool,默认情况下,当您拥有处理器时,默认情况下只有一个线程(由此返回)Runtime.getRuntime().availableProcessors()(这意味着并行流使用所有处理器,因为它们也使用主线程):

对于需要单独或自定义池的应用程序,可以使用给定的目标并行度级别构造ForkJoinPool; 默认情况下,等于可用处理器的数量.

这也意味着如果您同时启动嵌套并行流或多个并行流,它们将共享同一个池.优点:您永远不会使用超过默认值(可用处理器数量).缺点:您可能无法为您启动的每个并行流分配"所有处理器"(如果您碰巧有多个).(显然你可以使用ManagedBlocker来规避它.)

要更改并行流的执行方式,您也可以

• 将并行流执行提交给您自己的ForkJoinPool:yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get();或
• 您可以使用系统属性更改公共池的大小:System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20")对于20个线程的目标并行度.

我的机器上有后者的例子有8个处理器.如果我运行以下程序:

long start = System.currentTimeMillis();
IntStream s = IntStream.range(0, 20);
//System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");
s.parallel().forEach(i -> {
    try { Thread.sleep(100); } catch (Exception ignore) {}
    System.out.print((System.currentTimeMillis() - start) + " ");
});

输出是:

215 216 216 216 216 216 216 216 315 316 316 316 316 316 316 316 415 416 416 416

因此,您可以看到并行流一次处理8个项目,即它使用8个线程.但是,如果我取消注释注释行,则输出为:

215 215 215 215 215 216 216 216 216 216 216 216 216 216 216 216 216 216 216 216 216

这次,并行流使用了20个线程,并且流中的所有20个元素已同时处理.

Answer 3

除了在您自己的forkJoinPool中触发并行计算的技巧之外,您还可以将该池传递给CompletableFuture.supplyAsync方法,如:

ForkJoinPool forkJoinPool = new ForkJoinPool(2);
CompletableFuture<List<Integer>> primes = CompletableFuture.supplyAsync(() ->
    //parallel task here, for example
    range(1, 1_000_000).parallel().filter(PrimesPrint::isPrime).collect(toList()), 
    forkJoinPool
);

Answer 4

使用ForkJoinPool并提交并行流不能可靠地使用所有线程.如果你看一下(来自HashSet的并行流不并行运行)和这个(为什么并行流不使用ForkJoinPool的所有线程？),你会看到推理.

简短版本:如果ForkJoinPool/submit不适合您,请使用

ForkJoinPool fjpool = new ForkJoinPool(10);
System.out.println("stream.parallel");
IntStream range = IntStream.range(0, 20);
fjpool.submit(() -> range.parallel()
        .forEach((int theInt) ->
        {
            try { Thread.sleep(100); } catch (Exception ignore) {}
            System.out.println(Thread.currentThread().getName() + " -- " + theInt);
        })).get();
System.out.println("list.parallelStream");
int [] array = IntStream.range(0, 20).toArray();
List<Integer> list = new ArrayList<>();
for (int theInt: array)
{
    list.add(theInt);
}
fjpool.submit(() -> list.parallelStream()
        .forEach((theInt) ->
        {
            try { Thread.sleep(100); } catch (Exception ignore) {}
            System.out.println(Thread.currentThread().getName() + " -- " + theInt);
        })).get();

Answer 5

我们可以使用以下属性更改默认并行度：

-Djava.util.concurrent.ForkJoinPool.common.parallelism=16

可以设置使用更多并行性。

Answer 6

要测量实际使用的线程数,您可以检查Thread.activeCount():

    Runnable r = () -> IntStream
            .range(-42, +42)
            .parallel()
            .map(i -> Thread.activeCount())
            .max()
            .ifPresent(System.out::println);

    ForkJoinPool.commonPool().submit(r).join();
    new ForkJoinPool(42).submit(r).join();

这可以在4核CPU上产生如下输出:

5 // common pool
23 // custom pool

没有.parallel()它给出:

3 // common pool
4 // custom pool

Answer 7

到目前为止,我使用了这个问题的答案中描述的解决方案.现在,我想出了一个名为Parallel Stream Support的小库:

ForkJoinPool pool = new ForkJoinPool(NR_OF_THREADS);
ParallelIntStreamSupport.range(1, 1_000_000, pool)
    .filter(PrimesPrint::isPrime)
    .collect(toList())

但正如@PabloMatiasGomez在评论中指出的那样,并行流的分裂机制存在缺陷,这在很大程度上取决于公共池的大小.请参阅来自HashSet的并行流不并行运行.

我使用此解决方案只是为不同类型的工作提供单独的池,但即使我不使用它,我也无法将公共池的大小设置为1.

Answer 8

注意： JDK 10 中似乎实现了一个修复，可确保自定义线程池使用预期的线程数。

自定义 ForkJoinPool 中的并行流执行应遵循并行性 https://bugs.openjdk.java.net/browse/JDK-8190974

Answer 9

如果您不想依赖实现技巧，总有一种方法可以通过实现结合语义的自定义收集器来实现相同的目的map...collect并且您不会局限于 ForkJoinPool：

list.stream()
  .collect(parallel(i -> process(i), executor, 4))
  .join()

幸运的是，它已经在这里完成并可在 Maven Central 上使用： http: //github.com/pivovarit/parallel-collectors

免责声明：我写了它并承担责任。