并行流是否可以正常运行？

Question

我正在阅读有关无国籍状态的文章并在文档中遇到过这个问题:

如果流操作的行为参数是有状态的,则流管道结果可能是不确定的或不正确的.有状态lambda(或实现适当功能接口的其他对象)的结果取决于在流管道执行期间可能发生变化的任何状态.

现在,如果我有一个字符串列表(strList比如说),然后尝试使用并行流从中删除重复的字符串,方法如下:

List<String> resultOne = strList.parallelStream().distinct().collect(Collectors.toList());

或者如果我们想要不区分大小写:

List<String> result2 = strList.parallelStream().map(String::toLowerCase)
                       .distinct().collect(Collectors.toList());

此代码是否有任何问题,因为并行流将分割输入并且在一个块中不同并不一定意味着在整个输入中是不同的？

编辑(以下答案的快速摘要)

这distinct是有状态操作,并且在有状态中间操作的情况下,并行流可能需要多次通过或大量缓冲开销.distinct如果元素的排序不相关,也可以更有效地实现.另外根据文件:

对于有序流,不同元素的选择是稳定的(对于重复元素,保留在遇到顺序中首先出现的元素.)对于无序流,不进行稳定性保证.

但是在有序流并行运行的情况下,不同可能是不稳定的 - 意味着它将在重复的情况下保留任意元素,而不一定是distinct其他情况下预期的第一个元素.

从链接:

在内部,distinct()操作保持一个包含先前已经看到的元素的Set,但它隐藏在操作中,我们无法从应用程序代码中获取它.

因此,在并行流的情况下,它可能会消耗整个流,或者可能使用CHM(类似于ConcurrentHashMap.newKeySet()).对于有序的,很可能是使用LinkedHashSet或类似的结构.

Answer 1

大致指出doc(Emphasis mine)的相关部分:

中间操作进一步分为无状态操作和有状态操作.无状态操作(例如过滤器和映射)在处理新元素时不保留先前看到的元素的状态 - 每个元素都可以独立于其他元素上的操作进行处理.有状态操作(例如distinct和sorted)可以在处理新元素时包含先前看到的元素的状态

有状态操作可能需要在生成结果之前处理整个输入.例如,在查看流的所有元素之前,不能通过对流进行排序来产生任何结果.因此,在并行计算下,某些包含有状态中间操作的管道可能需要对数据进行多次传递,或者可能需要缓冲重要数据.仅包含无状态中间操作的管道可以在一次通过中处理,无论是顺序还是并行,具有最小的数据缓冲

如果您进一步阅读(订购部分):

流可能有也可能没有已定义的遭遇顺序.流是否具有遭遇顺序取决于源和中间操作.某些流源(例如List或数组)本质上是有序的,而其他流(例如HashSet)则不是.某些中间操作(例如sorted())可能会在其他无序流上强制执行遭遇顺序,而其他中间操作可能会呈现无序的有序流,例如BaseStream.unordered().此外,一些终端操作可以忽略遭遇顺序,例如forEach().

...

对于并行流,放宽排序约束有时可以实现更高效的执行.如果元素的排序不相关,则可以更有效地实现某些聚合操作,例如过滤重复(distinct())或分组缩减(Collectors.groupingBy()).类似地,与遇到订单本质上相关的操作(例如limit())可能需要缓冲以确保正确排序,从而破坏并行性的好处.在流具有遭遇顺序但用户不特别关心该遭遇顺序的情况下,使用无序()明确地对流进行排序可以改善某些有状态或终端操作的并行性能.然而,大多数流管道,例如上面的"块的权重总和"示例,即使在排序约束下仍然有效地并行化.

结论,

• distinct对于并行流可以正常工作,但正如您可能已经知道的那样,它必须在继续之前使用整个流,这可能会占用大量内存.
• 如果项目的源是无序集合(例如hashset)或流是unordered(),distinct则不担心排序输出因此会有效

.unordered()如果您不担心订单并且希望看到更多性能,解决方案是添加到流管道.

List<String> result2 = strList.parallelStream()
                              .unordered()
                              .map(String::toLowerCase)
                              .distinct()
                              .collect(Collectors.toList());

唉,Java中没有(可用的内置)并发哈希集(除非他们很聪明ConcurrentHashMap),所以我只能给你带来不幸的可能性,即使用常规Java集以阻塞方式实现distinct.在这种情况下,我认为做并行不同的任何好处.

编辑:我说得太早了.使用具有不同的并行流可能会有一些好处.看起来它的distinct实现比我最初想象的更聪明.请参阅@ Eugene的 回答.