takeWhile()对平台图的工作方式不同

Question

我正在用takeWhile创建代码片段，以探索它的可能性。当与flatMap一起使用时，这种行为不符合预期。请找到下面的代码片段。

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

ExpectedOutput：

Sample1
Sample2
Sample3

产生这种期望的原因是，takeWhile应该一直执行到内部条件变为真为止。我还在平台图中添加了printout语句以进行调试。流只返回两次，这与预期一致。

然而，这只是一个很好的没有平面地图在链中。

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

实际产出：

Sample3

这里，实际输出与预期输出匹配。

免责声明:这些代码片段只是用于代码实践，并不提供任何有效的应用程序。

更新： Bug JDK-8193856：fix将作为JDK 10的一部分提供。

@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中的一个bug -来自第8193856期

takeWhile错误地假设上游操作支持和荣誉取消，不幸的是，flatMap的情况并非如此。

解释

如果流是有序的，takeWhile应该显示预期的行为。这在您的代码中并不完全是这样的，因为您使用的是forEach，它放弃订单。如果您关心它(在本例中是这样做的)，则应该使用forEachOrdered。有趣的是:这不会改变任何事情。

所以也许这条溪流一开始就没有被点菜？(在这种情况下，这种行为是可以的.)如果您为从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作为sink直接在WhileOp上操作，您可以看到这是如何实现的：

• ReferencePipeline::forEachWithCancel做它的循环：
  • WhileOps::accept给出了每个流元素
  • 在每个元素之后查询WhileOps::cancellationRequested

​

• 在某个时候，"Sample4"失败了谓词，流被取消。

耶！

用flatMap

但是，在“坏情况”(使用flatMap；您的第一个示例)中，forEachWithCancel操作flatMap操作，它只是在ArraySpliterator for {"Sample3", "Sample4", "Sample5"}上调用forEachRemaining，这样做：

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操作，，但从不检查是否取消了。因此，在停止之前，它将急切地遍历“子流”中的所有元素，甚至可能是穿过小溪的其余部分。

Answer 2

不管我怎么看，这都是个bug --谢谢霍格的评论。我不想把这个答案放在这里(认真的！)，但是没有一个答案清楚地说明这是一个错误。

人们说，这必须与有序/非有序，这是不正确的，因为这将报告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

如果你看看takeWhile

如果此流是有序的，则返回一个由与给定谓词匹配的该流中提取的元素的最长前缀组成的流。 如果该流是无序的，则返回一个流，该流由从该流获取的与给定谓词匹配的元素子集组成。

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

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

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