在考虑并行化外部流之前,是否会完全并行处理内部并行流？

Question

从这个链接,我只是部分理解,至少在某些时候,java嵌套并行流存在问题.但是,我无法推断出以下问题的答案:

假设我有一个外部srtream和一个内部流,两者都使用并行流.事实证明,根据我的计算,如果内部流首先完全并行完成,那么它将更高效(由于数据位置,即L1/L2/L3 CPU缓存中的缓存)(如果且仅cpu核心可用)做外部流.我认为这对大多数人的情况都是如此.所以我的问题是:

Java会首先并行执行内部流,然后在outerstream上工作吗？如果是这样,它是在编译时还是在运行时做出决定？如果在运行时,JIT甚至足够聪明地意识到如果内部流确实具有比核心数量(32)更多的元素(例如数百个),那么它肯定应该使用所有32个内核来处理在从外部流移动下一个元素之前的内部流; 但是,如果元素的数量很小(例如<32),则可以"并行处理"来自"下一个"外部流的元素的元素.

Answer 1

也许下面的示例程序可以解释这个问题:

IntStream.range(0, 10).parallel().mapToObj(i -> "outer "+i)
         .map(outer -> outer+"\t"+IntStream.range(0, 10).parallel()
            .mapToObj(inner -> Thread.currentThread())
            .distinct() // using the identity of the threads
            .map(Thread::getName) // just to be paranoid, as names might not be unique
            .sorted()
            .collect(Collectors.toList()) )
         .collect(Collectors.toList())
         .forEach(System.out::println);

当然,结果会有所不同,但我机器上的输出看起来与此类似:

outer 0 [ForkJoinPool.commonPool-worker-6]
outer 1 [ForkJoinPool.commonPool-worker-3]
outer 2 [ForkJoinPool.commonPool-worker-1]
outer 3 [ForkJoinPool.commonPool-worker-1, ForkJoinPool.commonPool-worker-4, ForkJoinPool.commonPool-worker-5]
outer 4 [ForkJoinPool.commonPool-worker-5]
outer 5 [ForkJoinPool.commonPool-worker-2, ForkJoinPool.commonPool-worker-4, ForkJoinPool.commonPool-worker-7, main]
outer 6 [main]
outer 7 [ForkJoinPool.commonPool-worker-4]
outer 8 [ForkJoinPool.commonPool-worker-2]
outer 9 [ForkJoinPool.commonPool-worker-7]

我们在这里看到的是,对于我的机器,有八个内核,七个工作线程正在为工作做贡献,利用所有内核,对于公共池,调用者线程也将有助于工作,而不仅仅是等待完成.您可以清楚地看到main输出中的线程.

此外,您可以看到外部流获得完全并行性,而一些内部流仅由单个线程完全处理.每个工作线程都对外部流的元素中的至少一个做出贡献.如果将外部流的大小减小到核心数,则很可能只看到一个工作线程处理一个外部流元素,这意味着完全顺序执行所有内部流.

但是我使用的数字与核心数量不匹配,甚至不是它的倍数,以证明另一种行为.由于外部流处理的工作量不均匀,即一些线程仅处理一个项目,其他线程处理两个,这些空闲工作线程执行工作窃取,从而有助于剩余外部元素的内部流处理.

这种行为背后有一个简单的理由.当外部流的处理开始时,它不知道它将是"外部流".它只是一个并行流,没有办法找出这是否是一个外部流,而不是处理它,直到其中一个函数启动另一个流操作.但推迟并行处理是没有意义的,直到这一点可能永远不会到来.

除此之外,我强烈反对你假设"如果内部流首先完全并行完成,它会更高效"[...].对于典型的用例,我宁愿相反地期望它,阅读,期望与它实现完全一样的优势.但是,正如前一段所解释的那样,无论如何都没有合理的方法来实现并行处理内部流的首选项.