Java 8 Stream:limit()和skip()之间的区别

Question

谈论Streams,当我执行这段代码时

public class Main {
    public static void main(String[] args) {
        Stream.of(1,2,3,4,5,6,7,8,9)
        .peek(x->System.out.print("\nA"+x))
        .limit(3)
        .peek(x->System.out.print("B"+x))
        .forEach(x->System.out.print("C"+x));
    }
}

我得到了这个输出

A1B1C1
A2B2C2
A3B3C3

因为将我的流限制为前三个组件会强制执行动作A,B和C三次.

尝试使用skip()方法对最后三个元素执行类似的计算,显示了不同的行为:这

public class Main {
    public static void main(String[] args) {
        Stream.of(1,2,3,4,5,6,7,8,9)
        .peek(x->System.out.print("\nA"+x))
        .skip(6)
        .peek(x->System.out.print("B"+x))
        .forEach(x->System.out.print("C"+x));
    }
}

输出这个

A1
A2
A3
A4
A5
A6
A7B7C7
A8B8C8
A9B9C9

在这种情况下,为什么执行A1到A6的操作？它必须与limit是一个短路状态中间操作这一事实有关,而skip不是,但我不明白这个属性的实际含义.是不是"在跳过之前的所有动作都被执行而不是每个人都在限制之前"？

Answer 1

你在这里有两个流管道.

这些流管道每个都包含一个源,几个中间操作和一个终端操作.

但是中间操作是懒惰的.这意味着除非下游操作需要项目,否则不会发生任何事情.如果是,则中间操作会完成生成所需项目所需的全部操作,然后再次等待直到请求其他项目,依此类推.

终端操作通常是"急切的".也就是说,他们要求流中的所有项目完成所需的项目.

因此,您应该将管道视为forEach向下一个项目请求其后面的流,并且该流询问其后面的流,依此类推,一直到源.

考虑到这一点,让我们看看我们对您的第一个管道有什么:

Stream.of(1,2,3,4,5,6,7,8,9)
        .peek(x->System.out.print("\nA"+x))
        .limit(3)
        .peek(x->System.out.print("B"+x))
        .forEach(x->System.out.print("C"+x));

所以,forEach是要求第一项.这意味着"B" peek需要一个项目,并向limit输出流询问它,这意味着limit需要询问"A" peek,它将转到源.给出一个项目,并一直到达forEach,你得到你的第一行:

A1B1C1

该forEach请求另一个项目,然后另一个.每次,请求都在流上传播并执行.但是当forEach要求第四个项目时,当请求到达时limit,它知道它已经给出了允许给出的所有项目.

因此,它不是要求"A"偷看另一个项目.它立即指示其项目已耗尽,因此不再执行任何操作并forEach终止.

第二个管道会发生什么？

    Stream.of(1,2,3,4,5,6,7,8,9)
    .peek(x->System.out.print("\nA"+x))
    .skip(6)
    .peek(x->System.out.print("B"+x))
    .forEach(x->System.out.print("C"+x));

再次,forEach是要求第一项.这是传播回来的.但是当它到达时skip,它知道它必须从其上游请求6个项目才能通过下游.因此它在"A"上游发出请求,在peek不向下游传递请求的情况下使用它,发出另一个请求,依此类推.因此,"A"偷看获得6个项目请求并产生6个打印,但这些项目不会传递下来.

A1
A2
A3
A4
A5
A6

在第7个请求中skip,该项目被传递到"B"偷看并从它传递到forEach,所以完整的打印完成:

A7B7C7

那就像以前一样.的skip意志现在,一旦进入一个请求,要求一个项目的上游和下游传递,因为它"知道"它已经完成了它的跳跃任务.因此,其余的打印件将通过整个管道,直到源耗尽.

Answer 2

流水管道的流畅表示法正是造成这种混乱的原因.以这种方式思考:

limit(3)

所有流水线操作都是懒惰地进行评估,除了forEach()它是一个终端操作,它触发"执行管道".

在执行管道时,中间流定义不会对"之前"或"之后"发生的事情做出任何假设.他们所做的就是获取输入流并将其转换为输出流:

Stream<Integer> s1 = Stream.of(1,2,3,4,5,6,7,8,9);
Stream<Integer> s2 = s1.peek(x->System.out.print("\nA"+x));
Stream<Integer> s3 = s2.limit(3);
Stream<Integer> s4 = s3.peek(x->System.out.print("B"+x));

s4.forEach(x->System.out.print("C"+x));

• s1包含9个不同的Integer值.
• s2 窥视所有通过它并打印它们的值.
• s3将前3个值传递给第3个值s4并在第3个值之后中止管道.没有进一步的价值s3.这并不意味着管道中没有更多的值.s2仍然会产生(并打印)更多的值,但没有人请求这些值,因此执行停止.
• s4 再次偷看传递它的所有值并打印它们.
• forEach消耗并打印任何s4传递给它的东西.

这样想吧.整个流是完全懒惰的.只有终端操作主动从管道中提取新值.从中拉出3个值后s4 <- s3 <- s2 <- s1,s3将不再生成新值,并且不再从中提取任何值s2 <- s1.虽然s1 -> s2仍然能够生产4-9,但这些价值从未从管道中拉出,因此从未打印过s2.

skip(6)

有了skip()同样的事情发生:

Stream<Integer> s1 = Stream.of(1,2,3,4,5,6,7,8,9);
Stream<Integer> s2 = s1.peek(x->System.out.print("\nA"+x));
Stream<Integer> s3 = s2.skip(6);
Stream<Integer> s4 = s3.peek(x->System.out.print("B"+x));

s4.forEach(x->System.out.print("C"+x));

• s1包含9个不同的Integer值.
• s2 窥视所有通过它并打印它们的值.
• s3消耗前6个值,"跳过它们",这意味着前6个值不传递给s4,只有后续值.
• s4 再次偷看传递它的所有值并打印它们.
• forEach消耗并打印任何s4传递给它的东西.

这里重要的是,s2不知道剩余的管道跳过任何值.s2窥视所有价值,而不是事后发生的事情.

另一个例子:

请考虑此博客文章中列出的此管道

IntStream.iterate(0, i -> ( i + 1 ) % 2)
         .distinct()
         .limit(10)
         .forEach(System.out::println);

执行上述操作时,程序将永远不会停止.为什么？因为:

IntStream i1 = IntStream.iterate(0, i -> ( i + 1 ) % 2);
IntStream i2 = i1.distinct();
IntStream i3 = i2.limit(10);

i3.forEach(System.out::println);

意思是:

• i1产生交变值的无限量:0,1,0,1,0,1,...
• i2消耗之前遇到的所有值,仅传递"新"值,即总共有2个值i2.
• i3 传递10个值,然后停止.

该算法将永远不会停止,因为i3对于等待i2后,产生8个值0和1,但这些值永远不会出现,而i1从来没有停止进纸值i2.

在管道中的某个点上,产生了超过10个值并不重要.重要的是i3从未见过这10个价值观.

回答你的问题:

是不是"在跳过之前的所有动作都被执行而不是每个人都在限制之前"？

不.之前skip()或之前的所有操作limit().在你的两次执行中,你得到A1- A3.但是limit()可能会使管道短路,一旦发生了感兴趣的事件(达到限制)就中止价值消耗.

Answer 3

单独查看蒸汽操作完全是亵渎神灵,因为这不是评估流的方式.

谈到极限(3) ,它是一种短路操作,这是有道理的,因为想着它,无论操作之前和之后的limit,在流有限制将越来越之后停止迭代ñ元素,直到极限操作,但这并不意味着只处理n个流元素.以此不同的流操作为例

public class App 
{
    public static void main(String[] args) {
        Stream.of(1,2,3,4,5,6,7,8,9)
        .peek(x->System.out.print("\nA"+x))
        .filter(x -> x%2==0)
        .limit(3)
        .peek(x->System.out.print("B"+x))
        .forEach(x->System.out.print("C"+x));
    }
}

会输出

A1
A2B2C2
A3
A4B4C4
A5
A6B6C6

这似乎是正确的,因为限制正在等待3个流元素通过操作链,尽管处理了6个流元素.

Answer 4

所有的流都是基于spliterators，它基本上有两个操作：advanced（向前移动一个元素，类似于iterator）和split（将自己分割在任意位置，适合并行处理）。您可以随时停止获取输入元素（由 完成limit），但您不能只是跳转到任意位置（Spliterator界面中没有此类操作）。因此skip操作需要从源中实际读取第一个元素来忽略它们。请注意，在某些情况下，您可以执行实际跳转：

List<Integer> list = Arrays.asList(1,2,3,4,5,6,7,8,9);

list.stream().skip(3)... // will read 1,2,3, but ignore them
list.subList(3, list.size()).stream()... // will actually jump over the first three elements