Java 8 Stream 原理 - 调试笔记

Java 8 Stream 原理——调试笔记

 

理解的前提:对流的执行步骤有基本认识——中间操作和终止操作

 

示例程序:

Stream<String> stream = Stream.of( “a”, “bdd”, “bdd” );

stream.filter( s -> s.length() > 1 )//

        .sorted()//

        .mapToInt( String::length )//

        .sum();

 

代码解析:

1) stream.filter(xx); // 对应的是ReferencePipeline 类,创建一个新的流,将当前流赋值给新的流 previousStage 字段,并返回新创建的流。

代码:

return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE …

2) sorted()mapToInt() 方法都类似。

3) xx.sum() 对应的是IntPipeline

4) 代码产生流的链结构

sum() = mapToInt(pre) -> sorted(pre) -> filter(pre) -> stream

 

由于流只有调用终止操作才会真正执行,所以在 .sum() 打断点,开始调试。。。

 

断点调试

xx.sum(); // F5 进入 sum() 方法

reduce(0, Integer::sum); // F5 进入

ReduceOps.makeInt(identity, op) // 产生一个终止符 TerminalOp

evaluate(ReduceOps.makeInt(identity, op)); // F5, F7, F5 进入到 AbstractPipeline 

sourceSpliterator(terminalOp.getOpFlags()) // 创建一个 spliterator (暂叫分叉器)

terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags())); // F5, F7, F5, F7, F5 进入到 ReduceOps

helper.wrapAndCopyInto(makeSink(), spliterator).get();

// 1) F5, F7, F5 进入 AbstractPipeline.wrapAndCopyInto(S, Spliterator<P_IN>)

// 2) F5, F7, F5 进入 wrapSink(Sink<E_OUT>) // 此方法返回一个调用链为 filter() -> sorted() -> mapToInt() 收集点(sink)

// 3) F7, F5 进入 copyInto(Sink<P_IN>, Spliterator<P_IN>)

// 3a) wrappedSink.begin(spliterator.getExactSizeIfKnown()); 算是真正开始执行流式操作

// 4) F5 spliterator.forEachRemaining(wrappedSink) 进入 Spliterators$ArraySpliterator 类,此方法是对集合元素进行迭代。

 

action 对应的是 filter()

再看 ReferencePipeline.filter(xx) 源码

                    public void accept(P_OUT u) {

                        if (predicate.test(u)) // 测试通过

                            downstream.accept(u); // 下一个流处理

                }

 

下一个流对应的是 sorted()

跟踪 ReferencePipeline.sorted() -> SortedOps.makeRef(this) -> new OfRef<>(upstream)

comparator 是一个默认的比较器:Comparator.naturalOrder()

SortedOps$OfRef.opWrapSink(int flags, Sink<T> sink) 方法创建 RefSortingSink 对象。

具体看 RefSortingSink

也可以在 spliterator.forEachRemaining(wrappedSink) 里的

do { action.accept((T)a[i]); } while (++i < hi); 一直 F5, F7 也能找到

RefSortingSink 的工作方式很简单:accept(T t) 只是添加元素;end() 方法才排序,然后 for 循环调用 downstream.accept(t),最后调用 downstream.end();

其中 RefSortingSink.end() 是在 AbstractPipeline.copyInto(Sink<P_IN>, Spliterator<P_IN>) 里调用的(代码:wrappedSink.end();

 

下一个流对应的是 mapToInt()

查看 ReferencePipeline.mapToInt(xx),很简单——将值转换成 int 然后传给 downstream

 

最后sum() 调用的是 TerminalOp.get(),里面的 state operator.applyAsInt(state, t) 累加的

 

 

算法的理解

    final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {

        Objects.requireNonNull(sink);

 

        for ( @SuppressWarnings(“rawtypes”) AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {

            sink = p.opWrapSink(p.previousStage.combinedFlags, sink);

        }

        return (Sink<P_IN>) sink;

}

 

sum() = mapToInt(pre) -> sorted(pre) -> filter(pre) -> stream

 

sum(none) -> map(prev) -> sort()

sink = sum(none).opWrapSink(none)

 

sink = sum(sink) // this = sum

sink = map(sum(sink)) // map = sum.prev, sink = sum(sink)

sink = sort(map(sum(sink))) // sort = map.prev, sink = map(sum(sink))

灵感来了一下就能理解,灵感没来纠结了半天也不能理解能用画图理解更好.

教训:super(upstream, opFlags); 没再跟进去,导致字段理解有误!!!

 

    原文作者:fl_zxf
    原文地址: https://blog.csdn.net/fl_zxf/article/details/72550731
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞