java - 为什么具有短路操作的并行 Java Stream 会评估 Stream 的所有元素，而顺序 Stream 则不会？

Question

考虑两种测试方法parallel()和sequential():

  @Test
  public void parallel() throws Exception
  {
    System.out.println( "parallel start." );
    IntStream.of( 0, 1 ).parallel().map( this::work ).findAny();
    System.out.println( "parallel done." );
  }

  @Test
  public void sequential() throws Exception
  {
    System.out.println( "sequential start." );
    IntStream.of( 0, 1 ).map( this::work ).findAny();
    System.out.println( "sequential done." );
  }

  private int work(int i)
  {
    System.out.println( "working... " + i );
    Threads.sleepSafe( i * 1000 );
    System.out.println( "worked. " + i );
    return i;
  }

Threads.sleepSafe() 是 Thread.sleep() 的一个简单包装，它会吞掉异常，如果传递 0，则不执行任何操作。

运行测试方法时，结果如下:

sequential start.
working... 0
worked. 0
sequential done.

parallel start.
working... 1
working... 0
worked. 0
sleeping for 1000 ms ...
slept for 1000 ms.
worked. 1
parallel done.

sequential() 按我的预期运行，但 parallel() 却没有:我希望 parallel() 中的 findAny() 在 work() 第一次返回时立即返回(即对于值 0，因为它不会 hibernate )，而是仅在 work() 也完成值 1 后才返回。

为什么？

有没有办法让 findAny() 在 work() 第一次返回时立即返回？

Answer 1

并行模式下的Stream API基于ForkJoinPool范例，默认情况下使用最大 X 个线程(其中 X 等于可用处理器的数量)。如果您要增加迭代次数，可以检查此规则。

通常，并行流的默认线程池计数可以通过两种方式自定义:

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

Is there a way to make findAny() return as soon as work() returns for the first time?

根据 ForkJoin 算法的思想，基本上答案是否。当所有线程完成其工作时，它会“等待”。但正如前面所描述的，您可以将工作人员的数量限制为单个工作人员。显然它没有造成任何场景，因为这种方法类似于顺序执行，但由于冗余操作而产生额外开销。