Java parallelStream（）带有自定义池，调用方工作被窃取？

Question

通常，当使用Java8的parallelStream（）时，结果是通过默认的公共fork-join池（即forkjoinpool.commonpool（））执行。

然而，如果一个人的工作远离CPU的限制，例如，可能在很长时间内等待IO的工作，那么这显然是不可取的。在这种情况下，您将希望使用一个单独的池，根据其他标准（例如，任务可能实际使用CPU的时间的多少）进行大小调整。

没有明显的方法让parallelStream（）使用不同的池，但这里详细介绍了一种方法。

不幸的是，这种方法需要对来自fork-join池线程的并行流调用终端操作。这样做的缺点是，如果target-fork联接池完全忙于现有工作，则整个执行将在完全不做任何事情的情况下等待它。因此，池可能成为比单线程执行更糟糕的瓶颈。相反，当以“正常”方式使用parallelStream（）时，forkJoinPool.common.ExternalHelpComplete（）或forkJoinPool.common.TryExternalUnpush（）用于让池外部的调用线程帮助处理。

有没有人知道一种方法，既让parallelStream（）使用非默认的fork-join池，又让fork-join池外部的调用线程帮助处理这项工作（但不是fork-join池的其余工作）？

Answer 1

您可以在池上使用awaitquiescess来提供帮助。但是，您不能选择要帮助的任务，它只会从池中获取下一个挂起的任务，因此，如果有更多挂起的任务，您可能会在得到自己的任务之前执行这些任务。

ForkJoinPool forkJoinPool = new ForkJoinPool(1);
// make all threads busy:
forkJoinPool.submit(() -> LockSupport.parkNanos(Long.MAX_VALUE));
// submit our task (may contain your stream operation)
ForkJoinTask<Thread> task = forkJoinPool.submit(() -> Thread.currentThread());
// help out
while(!task.isDone()) // use zero timeout to execute one task only
    forkJoinPool.awaitQuiescence(0, TimeUnit.NANOSECONDS);
System.out.println(Thread.currentThread()==task.get());

将打印true。

鉴于

ForkJoinPool forkJoinPool = new ForkJoinPool(1);
// make all threads busy:
forkJoinPool.submit(() -> LockSupport.parkNanos(Long.MAX_VALUE));
// overload:
forkJoinPool.submit(() -> LockSupport.parkNanos(Long.MAX_VALUE));
// submit our task (may contain your stream operation)
ForkJoinTask<Thread> task = forkJoinPool.submit(() -> Thread.currentThread());
// help out
while(!task.isDone())
    forkJoinPool.awaitQuiescence(0, TimeUnit.NANOSECONDS);
System.out.println(Thread.currentThread()==task.get());

但请注意，fork/join框架和streamAPI之间的整个关系无论如何都是一个实现细节。