java - 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

预期输出:

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

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

免责声明:这些片段仅用于代码练习，不提供任何有效的用例。

更新:错误 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 is incorrectly assuming that an upstream operation supports and honors cancellation, which unfortunately is not the case for 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 操作进行操作，但是，只需为 {"Sample3", "Sample4", "Sample5"} 在 ArraySpliterator 上调用 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 操作，但从不检查它是否被取消。因此，它会在停止之前急切地遍历该“子流”中的所有元素，甚至可能是 through the rest of the stream .