为什么溪流不能减少不平行？/stream已被操作或关闭

Question

我偶然发现了一个相当烦人的问题：我有一个程序，它有很多数据源，可以流式传输相同类型的元素，而我想要“映射”程序中每个可用的元素（元素顺序不重要）。

由于元素顺序对我来说并不重要，所以我尝试使用.Parallel():Streamoft=StreamofStreamoft.Parallel（）.reduce(stream.empty(),stream::Concat)；来并行化reduce操作，但这会触发java.lang.IllegalStateException：stream已经被操作或关闭

要亲自体验它，只需通过注释/取消注释.parallel()来使用以下main(java 1.8U20)即可

public static void main(String[] args) {
    // GIVEN
    List<Stream<Integer>> listOfStreamOfInts = new ArrayList<>();
    for (int j = 0; j < 10; j++) {
        IntStream intStreamOf10Ints = IntStream.iterate(0, i -> i + 1)
                .limit(10);
        Stream<Integer> genericStreamOf10Ints = StreamSupport.stream(
                intStreamOf10Ints.spliterator(), true);
        listOfStreamOfInts.add(genericStreamOf10Ints);
    }
    Stream<Stream<Integer>> streamOfStreamOfInts = listOfStreamOfInts
            .stream();
    // WHEN
    Stream<Integer> streamOfInts = streamOfStreamOfInts
            // ////////////////
            // PROBLEM
            //    |
            //    V
            .parallel()
            .reduce(Stream.empty(), Stream::concat);

    // THEN
    System.out.println(streamOfInts.map(String::valueOf).collect(
            joining(", ")));
}

有人能解释一下这种限制吗？ <罢工> /找到一种更好的处理流的并行缩减的方法

在@smutje和@louiswasserman注释之后，.flatmap(function.identity())似乎是一个更好的选项，可以容忍.parallel()流

Answer 1

您正在使用的reduce的形式采用一个标识值和一个关联组合函数。但stream.empty()不是值；它有状态。流不是像数组或集合那样的数据结构；它们是通过可能并行的聚合操作推动数据的载体，并且它们具有某种状态（例如流是否已被消耗）想想这是如何运作的；你要建立一棵树，在那里，相同的“空”流出现在多个叶子中。当您尝试使用这个有状态的not-an-identity两次时（这不会顺序发生，而是并行发生），第二次尝试并遍历该空流时，将非常正确地看到它已经被使用了。

所以问题是，您只是错误地使用了reduce方法。问题不在于并行性；这只是因为并行性暴露了潜在的问题。

其次，即使这“工作”的方式你认为它应该，你只会得到并行构建树，代表的是扁平的溪流；当你去做连接时，那是一个顺序流管道。哎呀。

第三，即使这种方法按照您认为的方式“工作”，您也会通过建立串联流而增加大量的元素访问开销，而且您不会得到您所寻求的并行性的好处。

简单的答案就是把溪流弄平：

String joined = streamOfStreams.parallel()
                               .flatMap(s -> s)
                               .collect(joining(", "));