我正在阅读有关无状态的内容并在 doc 中遇到了这个:

Stream pipeline results may be nondeterministic or incorrect if the behavioral parameters to the stream operations are stateful. A stateful lambda (or other object implementing the appropriate functional interface) is one whose result depends on any state which might change during the execution of the stream pipeline.

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

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

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

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

这段代码会不会有任何问题，因为并行流会拆分输入并且在一个 block 中不同并不一定意味着在整个输入中不同？

编辑(以下答案的快速总结)

distinct 是有状态操作，在有状态中间操作的情况下，并行流可能需要多次传递或大量缓冲开销。如果元素的排序不相关，distinct 也可以更有效地实现。 同样根据 doc :

For ordered streams, the selection of distinct elements is stable (for duplicated elements, the element appearing first in the encounter order is preserved.) For unordered streams, no stability guarantees are made.

但是在并行运行的有序流的情况下，distinct 可能是不稳定的——这意味着它会在重复的情况下保留任意元素，而不一定是 distinct 中预期的第一个元素。

来自link :

Internally, the distinct() operation keeps a Set that contains elements that have been seen previously, but it’s buried inside the operation and we can’t get to it from application code.

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

最佳答案

大致指出了doc的相关部分(强调，我的):

Intermediate operations are further divided into stateless and stateful operations. Stateless operations, such as filter and map, retain no state from previously seen element when processing a new element -- each element can be processed independently of operations on other elements. Stateful operations, such as distinct and sorted, may incorporate state from previously seen elements when processing new elements

Stateful operations may need to process the entire input before producing a result. For example, one cannot produce any results from sorting a stream until one has seen all elements of the stream. As a result, under parallel computation, some pipelines containing stateful intermediate operations may require multiple passes on the data or may need to buffer significant data. Pipelines containing exclusively stateless intermediate operations can be processed in a single pass, whether sequential or parallel, with minimal data buffering

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

Streams may or may not have a defined encounter order. Whether or not a stream has an encounter order depends on the source and the intermediate operations. Certain stream sources (such as List or arrays) are intrinsically ordered, whereas others (such as HashSet) are not. Some intermediate operations, such as sorted(), may impose an encounter order on an otherwise unordered stream, and others may render an ordered stream unordered, such as BaseStream.unordered(). Further, some terminal operations may ignore encounter order, such as forEach().

...

For parallel streams, relaxing the ordering constraint can sometimes enable more efficient execution. Certain aggregate operations, such as filtering duplicates (distinct()) or grouped reductions (Collectors.groupingBy()) can be implemented more efficiently if ordering of elements is not relevant. Similarly, operations that are intrinsically tied to encounter order, such as limit(), may require buffering to ensure proper ordering, undermining the benefit of parallelism. In cases where the stream has an encounter order, but the user does not particularly care about that encounter order, explicitly de-ordering the stream with unordered() may improve parallel performance for some stateful or terminal operations. However, most stream pipelines, such as the "sum of weight of blocks" example above, still parallelize efficiently even under ordering constraints.

总而言之，

• distinct 可以很好地处理并行流，但您可能已经知道，它必须在继续之前消耗掉整个流，这可能会占用大量内存。
• 如果项目的来源是无序集合(例如 hashset)或流是 unordered()，则 distinct 不担心对输出进行排序并且这样会很有效率

如果您不担心顺序并希望看到更多性能，解决方案是将 .unordered() 添加到流管道。

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

唉，在 Java 中没有(可用的内置)并发哈希集(除非他们对 ConcurrentHashMap 很聪明)，所以我只能留给你一个不幸的可能性，即 distinct 是使用常规 Java 集以阻塞方式实现的。在这种情况下，我看不到执行并行 distinct 有任何好处。

编辑:我说得太早了。使用具有不同的并行流可能会有一些好处。看起来 distinct 的实现比我最初想象的要聪明。参见 @Eugene's answer .

关于java - 并行流是否可以在不同的操作下正常工作？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/53645037/