并行流中的Fork-Join池

Question

我已经搜索了网上的各种文章和堆栈溢出问题，但我不能找到这个完美的答案。有许多问题与此相近，但略有不同。

我们知道Java8Streams API在内部使用Fork-Join池。

现在我的问题是如何使用Fork-Join池来划分流管道中的任务？

假设我们有以下内容：

List myList =  inputList.parallelStream().filter( x -> x>0 )
    .map(x -> x+100 ).collect(Collectors.toList());

null

List myList2 = inputList.parallelStream().filter( x -> x>0 )
    .map(x -> x+5 ).sorted().map(x -> x+5 ).collect(Collectors.toList());

Answer 1

首先，您必须使用Parallel使fork-join pooly处于活动状态。这个答案稍微解释了spliterator是如何执行拆分的；但简单地说，拆分是使用流元素的源完成的，并且并行处理整个流水线。在您的示例中，它是filter和map（当然，它还包括terminal操作）。

对于有状态操作--事情更加复杂。让我们以distince为例，首先看看它是如何处理顺序情况的。

通常，您可能认为非并行不同可以使用哈希集实现--您的想法是正确的。hashset可以保存已经看到的所有值，而不处理（发送到下一个操作）其他元素--理论上可以使用非并行的distince操作。但是，如果流已知是排序的怎么办？想想看，这意味着我们可以保留一个将标记为seen的元素（与前面的hashset)。基本上，如果你愿意：

 1,1,2,2,3

这意味着您的有状态操作可以在单个元素之上实现--而不是hashset；代码类似于：

T seen = null;
....
if(seen == null || (!currentElement.equals(seen)){
    seen = currentElement;
    // process seen;
}

但是只有当您知道流是sorted时，这种优化才是可能的，因为这样您就知道下一个元素要么与您已经看到的元素相同，要么是一个新的元素，这对于您以前在其他操作中看到的元素是不可能的--排序操作保证了这一点。

现在是如何实现Parallel Distinct的。你基本上问这个问题：

那么如何创建线程池

同样，从流的角度看，没有任何改变，forkjoinpool使用相同数量的线程--唯一改变的显然是流实现。

简单地说，如果您的流是有序的，内部实现将使用LinkedHashSet（实际上是它的多个实例，因为它在这种情况下确实会减少）来保留您的顺序，如果您不关心顺序，它将使用ConcurrentHashMap--也就是说，如果源代码不是有序的（就像Set)或者您使用了显式的称为Unordered的Unordered。如果您真的想知道它是如何完成的，也可以在实现中查找sorted。

因此，底线是fork Join pool不会改变基于流的实现，它使用相同的模型。另一方面，根据您所拥有的操作，Stream API可能将一些有状态数据用于有状态的中间操作，例如hashset/concurrenthashmap或单个元素等。