takeWhile() 与平面图的工作方式不同

Question

我正在用 takeWhile 创建 sn-ps 来探索它的可能性。与 flatMap 配合使用时，行为不符合预期。请在下面找到代码 sn-p。

String[][] strArray = {{"Sample1", "Sample2"}, {"Sample3", "Sample4", "Sample5"}};

Arrays.stream(strArray)
        .flatMap(indStream -> Arrays.stream(indStream))
        .takeWhile(ele -> !ele.equalsIgnoreCase("Sample4"))
        .forEach(ele -> System.out.println(ele));

实际输出：

Sample1
Sample2
Sample3
Sample5

预期输出：

Sample1
Sample2
Sample3

期望的原因是 takeWhile 应该一直执行到内部条件变为真为止。我还在 flatmap 中添加了打印输出语句以进行调试。流仅返回两次，符合预期。

但是，如果链中没有平面图，这也可以正常工作。

String[] strArraySingle = {"Sample3", "Sample4", "Sample5"};
Arrays.stream(strArraySingle)
        .takeWhile(ele -> !ele.equalsIgnoreCase("Sample4"))
        .forEach(ele -> System.out.println(ele));

实际输出：

Sample3

此处实际输出与预期输出匹配。

免责声明：这些 sn-ps 仅用于代码练习，不提供任何有效的用例。

更新： 错误JDK-8193856：修复将作为 JDK 10 的一部分提供。 更改将更正whileOps 接收器::接受

@Override 
public void accept(T t) {
    if (take = predicate.test(t)) {
        downstream.accept(t);
    }
}

改变的实现：

@Override
public void accept(T t) {
    if (take && (take = predicate.test(t))) {
        downstream.accept(t);
    }
}

Answer 1

这是 JDK 9 中的一个错误 - 来自 issue #8193856：

takeWhile 错误地假设上游操作支持并尊重取消，不幸的是flatMap 并非如此。

说明

如果流是有序的，takeWhile 应该会显示预期的行为。在您的代码中并非完全如此，因为您使用了forEach，它放弃了订单。如果你关心它，你在这个例子中做了，你应该使用forEachOrdered。有趣的是：这并没有改变任何事情。 ?

所以也许一开始就没有对流进行排序？ （在那种情况下the behavior is ok。）如果你为从strArray创建的流创建一个临时变量，并通过在断点处执行表达式((StatefulOp) stream).isOrdered();来检查它是否有序，你会发现它确实是有序的：

String[][] strArray = {{"Sample1", "Sample2"}, {"Sample3", "Sample4", "Sample5"}};

Stream<String> stream = Arrays.stream(strArray)
        .flatMap(indStream -> Arrays.stream(indStream))
        .takeWhile(ele -> !ele.equalsIgnoreCase("Sample4"));

// breakpoint here
System.out.println(stream);

这意味着这很可能是一个实现错误。

进入代码

正如其他人所怀疑的，我现在也认为这可能与flatMap 的渴望有关。更准确地说，这两个问题可能具有相同的根本原因。

查看WhileOps的来源，我们可以看到这些方法：

@Override
public void accept(T t) {
    if (take = predicate.test(t)) {
        downstream.accept(t);
    }
}

@Override
public boolean cancellationRequested() {
    return !take || downstream.cancellationRequested();
}

takeWhile 使用此代码检查给定的流元素 t 是否满足 predicate：

• 如果是这样，它将元素传递给downstream 操作，在本例中为System.out::println。
• 如果不是，则将take设置为false，因此下次询问是否应该取消管道（即已完成）时，返回true。

这涵盖了takeWhile 操作。你需要知道的另一件事是forEachOrdered导致终端操作执行ReferencePipeline::forEachWithCancel方法：

@Override
final boolean forEachWithCancel(Spliterator<P_OUT> spliterator, Sink<P_OUT> sink) {
    boolean cancelled;
    do { } while (
            !(cancelled = sink.cancellationRequested())
            && spliterator.tryAdvance(sink));
    return cancelled;
}

所有这些都是：

1. 检查管道是否被取消
2. 如果没有，则将接收器前移一个元素
3. 如果这是最后一个元素则停止

看起来很有希望，对吧？

没有flatMap

在“好的情况下”（没有flatMap；您的第二个示例）forEachWithCancel 直接在WhileOp 上操作为sink，您可以看到这是如何进行的：

• ReferencePipeline::forEachWithCancel 循环：
  • WhileOps::accept 被赋予每个流元素
  • 在每个元素之后查询WhileOps::cancellationRequested
• 在某些时候"Sample4" 使谓词失败并且流被取消

耶！

与flatMap

在“坏情况”（flatMap；您的第一个示例）中，forEachWithCancel 对 flatMap 操作进行操作，不过，它只是在 ArraySpliterator 上调用 forEachRemaining 以获取 {"Sample3", "Sample4", "Sample5"}，这是做什么的：

if ((a = array).length >= (hi = fence) &&
    (i = index) >= 0 && i < (index = hi)) {
    do { action.accept((T)a[i]); } while (++i < hi);
}

忽略所有hi 和fence 的东西，它们仅在数组处理被拆分为并行流时使用，这是一个简单的for 循环，它将每个元素传递给takeWhile 操作， 但从不检查是否已取消。因此，它会在停止之前急切地遍历该“子流”中的所有元素，甚至可能是through the rest of the stream。

Answer 2

无论我怎么看，这都是一个错误 - 感谢 Holger 提供的 cmets。我不想把这个答案放在这里（说真的！），但没有一个答案明确指出这是一个错误。

人们说这与有序/无序有关，但事实并非如此，因为这将报告 true 3 次：

Stream<String[]> s1 = Arrays.stream(strArray);
System.out.println(s1.spliterator().hasCharacteristics(Spliterator.ORDERED));

Stream<String> s2 = Arrays.stream(strArray)
            .flatMap(indStream -> Arrays.stream(indStream));
System.out.println(s2.spliterator().hasCharacteristics(Spliterator.ORDERED));

Stream<String> s3 = Arrays.stream(strArray)
            .flatMap(indStream -> Arrays.stream(indStream))
            .takeWhile(ele -> !ele.equalsIgnoreCase("Sample4"));
System.out.println(s3.spliterator().hasCharacteristics(Spliterator.ORDERED));

如果你把它改成这样也很有趣：

String[][] strArray = { 
         { "Sample1", "Sample2" }, 
         { "Sample3", "Sample5", "Sample4" }, // Sample4 is the last one here
         { "Sample7", "Sample8" } 
};

那么Sample7 和Sample8 将不会成为输出的一部分，否则它们会。似乎flatmap 忽略 dropWhile 将引入的取消标志。

Answer 3

如果你看the documentation for takeWhile：

如果这个流是有序的，[返回]一个由 从此流中获取的与给定匹配的元素的最长前缀 谓词。

如果这个流是无序的，[返回]一个由一个子集组成的流 从此流中获取的与给定谓词匹配的元素。

您的流是巧合排序的，但takeWhile 不知道它是。因此，它返回第二个条件 - 子集。你的takeWhile 就像filter。

如果您在takeWhile 之前添加对sorted 的调用，您将看到预期的结果：

Arrays.stream(strArray)
      .flatMap(indStream -> Arrays.stream(indStream))
      .sorted()
      .takeWhile(ele -> !ele.equalsIgnoreCase("Sample4"))
      .forEach(ele -> System.out.println(ele));

Answer 4

原因是flatMap 操作也是一个intermediate operations，其中一个有状态短路中间操作 takeWhile 是用过。

Holger 在this answer 中指出的flatMap 的行为当然是一个参考，人们不应错过了解此类短路操作的意外输出。

您的预期结果可以通过拆分这两个中间操作来实现，方法是引入终端操作以进一步确定性地使用有序流并将它们作为样本执行：

List<String> sampleList = Arrays.stream(strArray).flatMap(Arrays::stream).collect(Collectors.toList());
sampleList.stream().takeWhile(ele -> !ele.equalsIgnoreCase("Sample4"))
            .forEach(System.out::println);

另外，似乎有一个相关的Bug#JDK-8075939 来跟踪已经注册的这种行为。

编辑：这可以在JDK-8193856 被接受为错误进一步跟踪。