Java 8-stream實現原理分析(一)

阿新 • • 發佈：2019-01-14

背景介紹

Java 8中引入了lambda和stream,極大的簡化了程式碼編寫工作,但是簡單的程式碼為何能實現如何豐富的功能,其背後又是如何實現的呢?

Spliterator和Iterator

Iterator

Iterator是Java中的第二個迭代器介面,在Java 1.2就已存在,相比Enumeration方法名更簡短,還增加了remove()方法,但是Iterator始終存在兩個問題:

Iterator 有兩個方法：hasNext() 和 next()；訪問下一個元素可能涉及到（但不需要）呼叫這兩個方法。因此，正確編寫 Iterator 需要一定量的防禦性和重複性編碼。（如果客戶端沒有在呼叫 next() 之前呼叫 hasNext() 會怎麼樣？如果它呼叫 hasNext() 兩次會怎麼樣？）
此外，這種兩方法協議通常需要一定水平的有狀態性，比如前窺 (peek ahead ) 一個元素（並跟蹤您是否已前窺）。這些要求累積形成了大量的每元素訪問開銷。

第一個問題比較容易理解,第二個問題就是因為Iterator中有remove()方法存在,若要在Iterator迴圈中使用remove(),就一定要記錄前一個元素(即前窺一個元素),比如在ArrayList.Itr中除了有cursor代表下一個要訪問的元素下標外,還有lastRet記錄上一個訪問元素的下標.

Spliterator

Spliterator（splitable iterator可分割迭代器）是Java 中引進的第三個迭代器介面

使用boolean tryAdvance(Consumer);代替hasNext()和next()
不再提供remove()方法
提供Spliterator trySplit();將自身一分為二,支援併發

流來源

Spliterator即為流來源

實現方式

本來還想繼續寫下去,但是發現再如何寫也無法超越我當時學習時看的部落格,暫時放棄.
深入理解Java Stream流水線

程式碼分析

下面以一段程式碼示例分析下stream的原始碼

public class StreamDemo {
    public static void main(String[] args) {
        List<String> strings = List.of("Apple", "bug", "ABC", "Dog");
        strings = new 
 ArrayList<>(strings);
        OptionalInt max
                = strings.stream()
                //無狀態中間操作
                .filter(s -> s.startsWith("A"))
                //無狀態中間操作
                .mapToInt(String::length)
                //有狀態中間操作
                .sorted()
                //非短路終端操作
                .max();
    }
}

由於直到終端操作才會執行真正的運算,直接看到max(),max其實是reduce操作,最後會呼叫ReduceOp.evaluateSequential()

    /**
     * @param helper 終端操作的前一箇中間操作,通過呼叫"helper.wrapSink()"將"sink"構造成鏈
     * @param spliterator 流來源,即ArrayList.Spliterator()
     */
    @Override
    public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
                                       Spliterator<P_IN> spliterator) {
        return helper.wrapAndCopyInto(makeSink(), spliterator).get();
    }

    /**
     * 
     * @param sink ReduceOp呼叫makeSink()獲得的Sink
     */
    @Override
    @SuppressWarnings("unchecked")
    final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
        Objects.requireNonNull(sink);
        //從最後一個stage直到第一個stage向前呼叫每個stage的opWrapSink()將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;
    }

opWrapSink()是個抽象方法,我們看下filter()的opWrapSink()如何實現

    @Override
    public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
        Objects.requireNonNull(predicate);
        return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                StreamOpFlag.NOT_SIZED) {
            /**
             *
             * @param flags 下一個sink的標誌位,供優化使用
             * @param sink 下一個sink,通過此引數將sink構造成單鏈
             * @return 當前中間操作關聯的sink
             */
            @Override
            Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
                //Sink.ChainedReference是Sink介面的預設實現,僅呼叫下一個sink的相應方法
                return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                    @Override
                    public void begin(long size) {
                        downstream.begin(-1);
                    }

                    @Override
                    public void accept(P_OUT u) {
                        if (predicate.test(u))
                            downstream.accept(u);
                    }
                };
            }
        };
    }

opWrapSink()根據下一個sink和flags構造當前sink,並將當前sink返回,便於構成sink鏈

    /**
     *
     * @param wrappedSink 呼叫wrapSink()返回的結果,即"第一個sink"
     * @param spliterator 流來源
     */
    @Override
    final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
        Objects.requireNonNull(wrappedSink);

        if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
            //依次呼叫begin()-->accept()-->end(),由於sink已經連結在一起,可以呼叫下一個sink的相應方法
            wrappedSink.begin(spliterator.getExactSizeIfKnown());
            spliterator.forEachRemaining(wrappedSink);
            wrappedSink.end();
        }
        else {
            copyIntoWithCancel(wrappedSink, spliterator);
        }
    }

copyInto()依次呼叫第一個sink的begin(),accept(),end(),此時終端操作結果存放在最後一個sink中,返回終端操作結果即可.

Sink執行分析

接下來我們分析下4個sink(3箇中間操作+1個終端操作)之間是如何協作的

filter的begin()

                    @Override
                    public void begin(long size) {
                        //什麼都不做,僅僅呼叫dowStream.begin(),由於不確定傳給downStream的元素個數是多少,因此用引數"-1"代表不確定
                        downstream.begin(-1);
                    }

map的begin()

        @Override
        public void begin(long size) {
            //也是什麼都不做,但是map不會減少傳給downStream的資料個數,引數依然是"size"
            downstream.begin(size);
        }

sorted的begin()

        @Override
        public void begin(long size) {
            if (size >= Nodes.MAX_ARRAY_SIZE)
                throw new IllegalArgumentException(Nodes.BAD_SIZE);
            //初始化相關的引數,不在傳給downStream
            b = (size > 0) ? new SpinedBuffer.OfInt((int) size) : new SpinedBuffer.OfInt();
        }

filter的accpet()

                    @Override
                    public void accept(P_OUT u) {
                        //只有通過predicate.test(u)才會傳遞給downStream
                        if (predicate.test(u))
                            downstream.accept(u);
                    }

map的accpet()

                    @Override
                    public void accept(P_OUT u) {
                        //執行map操作
                        downstream.accept(mapper.applyAsInt(u));
                    }

sorted的accpet()

        @Override
        public void accept(int t) {
            //呼叫b.accept(t),b是SpinedBuffer型別,暫不分析
            //同樣並未傳遞給downStream
            b.accept(t);
        }

可能到這裡就有些疑惑了,為什麼sorted stage一直沒有呼叫ReduceOp sink的相關方法?我們帶著這個疑問繼續.

filter的end()

        @Override
        public void end() {
            //簡單傳遞
            downstream.end();
        }

map的end()

        @Override
        public void end() {
            //同樣是簡單傳遞
            downstream.end();
        }

sorted的end()

        @Override
        public void end() {
            int[] ints = b.asPrimitiveArray();
            Arrays.sort(ints);
            //傳遞
            downstream.begin(ints.length);
            if (!cancellationWasRequested) {
                for (int anInt : ints)
                    //傳遞
                    downstream.accept(anInt);
            }
            else {
                for (int anInt : ints) {
                    if (downstream.cancellationRequested()) break;
                    downstream.accept(anInt);
                }
            }
            //傳遞
            downstream.end();
        }

看到這裡,我們就可以解釋上面的問題了:
由於sorted是一個有狀態的中間操作,在sorted完成之前,不能傳遞給downStream,只有在sorted的end()中才可傳遞給downStream

ReduceOp的相關方法

            ...

            public void begin(long size) {
                empty = true;
                //state儲存最後的max結果,初始為0
                state = 0;
            }

            @Override
            public void accept(int t) {
                if (empty) {
                    empty = false;
                    state = t;
                }
                else {
                    //operator就是Math::max,state存放最大值
                    state = operator.applyAsInt(state, t);
                }
            }
            ...

總結

stream的中間操作會構造一個stage鏈,在遇到終端操作時才會真正執行
wrapSink()–>當遇到終端操作時,最後一個stage呼叫wrapSink(),將最後一個stage直到第一個stage向前呼叫每個stage的opWrapSink()將sink構造成鏈
copyInto–>sink鏈構造完畢後,依次呼叫第一個sink的begin(),accept(),end(),執行完畢後終端操作結果存放在最後一個sink中,返回終端操作結果即可

不足

尚未分析並行時如何執行
示例程式碼中的終端操作時非短路操作,沒有看出Sink.cancellationRequested()的作用