推薦一篇博文,很好的介紹了Stream的原理.本文對其進行一些補充更加詳細的講解.

需求:

從"張三","李四","王二","張四五"中選出以張開頭的名字,然後從再從中選出名字最長的一個,輸出其長度.

1.一種直白的實現

缺點:
1. 迭代次數過多
2. 頻繁產生中間結果,效能無法接受

實際想要的效果:
平常的寫法:

int longest = 0;
for(String str : strings){
    if(str.startsWith("張")){// 1. filter(), 保留以張開頭的字串
        int len = str.length();// 2. mapToInt(), 轉換成長度
 

        longest = Math.max(len, longest);// 3. max(), 保留最長的長度
    }
}
System.out.println(longest);

Stream的做法:

  Stream.of("張三","李四","王二","張四五")
        .filter(x -> x.startsWith("張"))
        .mapToInt(String::length)
        .max()
        .ifPresent(System.out::println);

2.Stream是怎麼做到的?

Stream的操作分類

中間操作:返回一個新的Stream
- 有狀態 sorted(),必須等上一步操作完拿到全部元素後才可操作
- 無狀態 filter(),該操作的元素不受上一步操作的影響

  list.stream().filter(x -> x.startWith("張").map(x -> x.length())
  list.stream().filter(x -> x.startWith("張").sorted().map(x -> x.length())

終端操作:返回結果
- 短路操作findFirst(),找到一個則返回,也就是break當前的迴圈
- 非短路操作forEach(),遍歷全部元素

以上操作決定了Stream一定是先構建完畢再執行的特點,也就是延遲執行,當需要結果(終端操作時)開始執行流水線.
Stream做到的是對於多次呼叫合併到一次迭代中處理完所有的呼叫方式.換句話說就是解決了上述的兩個缺點.大概思路是記錄下每一步的操作,然後終端操作時對其迭代依次執行每一步的操作,最後再一次迴圈中處理.

問題:
1. 操作是如何記錄下來的?
2. 操作是如何疊加的?
3. 疊加完如何執行的?
4. 執行完如何收集結果的?

Stream結構示意圖:

示例程式碼:

    List<String> data = new ArrayList<>();
    data.add("張三");
    data.add("李四");
    data.add("王三");
    data.add("馬六");

    data.stream()
        .filter(x -> x.length() == 2)
        .map(x -> x.replace("三","五"))
        .sorted()
        .filter(x -> x.contains("五"))
        .forEach(System.out::println);

1. 操作是如何記錄下來的?

1. Head記錄Stream起始操作
2. StatelessOp記錄中間操作
3. StatefulOp記錄有狀態的中間操作
   這三個操作例項化會指向其父類AbstractPipeline,也就是在AbstractPipeline中建立了雙向連結串列

對於Head

    AbstractPipeline(Spliterator<?> source,
                     int sourceFlags, boolean parallel) {
        this.previousStage = null; //首操作上一步為null    
        this.sourceSpliterator = source; //資料
        this.sourceStage = this; //Head操作
        this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
        this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
        this.depth = 0;
        this.parallel = parallel;
    }

對於其他Stage:

    AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
        if (previousStage.linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        previousStage.linkedOrConsumed = true;
        //雙向連結串列的建立
        previousStage.nextStage = this;
        this.previousStage = previousStage;
        this.sourceStage = previousStage.sourceStage;        
        this.depth = previousStage.depth + 1;        

        this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
        this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
        if (opIsStateful())
            sourceStage.sourceAnyStateful = true;
    }

呼叫過程如此用雙向連結串列串聯起來,每一步都得知其上一步與下一步的操作.
data.stream()
.filter(x -> x.length() == 2)
.map(x -> x.replace(“三”,”五”))
.sorted()
.filter(x -> x.contains(“五”))
.forEach(System.out::println);

2.操作是如何疊加的?

Sink<T>介面:
1. void begin(long size),迴圈開始前呼叫,通知每個Stage做好準備
2. void end(),迴圈結束時呼叫,依次呼叫每個Stage的end方法,處理結果
3. boolean cancellationRequested(),判斷是否可以提前結束迴圈
4. void accept(T value),每一步的處理

其子類之一ChainedReference:

    static abstract class ChainedReference<T, E_OUT> implements Sink<T> {
        protected final Sink<? super E_OUT> downstream;

        public ChainedReference(Sink<? super E_OUT> downstream) {
            this.downstream = Objects.requireNonNull(downstream);
        }
        @Override
        public void begin(long size) {
            downstream.begin(size);
        }
        @Override
        public void end() {
            downstream.end();
        }
        @Override
        public boolean cancellationRequested() {
            return downstream.cancellationRequested();
        }
    }

例Filter:

    @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) {
            @Override
            Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> 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) {
                        //條件成立則傳遞給下一個操作,也因為如此所以有狀態的操作必須放到
                        //end方法裡面
                        if (predicate.test(u))
                            downstream.accept(u);
                    }
                };
            }
        };
    }

再例如sorted():

        @Override
        public void begin(long size) {
            if (size >= Nodes.MAX_ARRAY_SIZE)
                throw new IllegalArgumentException(Nodes.BAD_SIZE);
            list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();
        }
        @Override
        public void end() {
            list.sort(comparator);
            downstream.begin(list.size());
            if (!cancellationWasRequested) {
                list.forEach(downstream::accept);
            }
            else {
                for (T t : list) {
                    if (downstream.cancellationRequested()) break;
                    downstream.accept(t);
                }
            }
            downstream.end();
            list = null;
        }
        @Override
        public void accept(T t) {
            list.add(t);
        }

疊加後如何執行?

執行操作是由終端操作來觸發的,例如foreach操作

    @Override
    public void forEach(Consumer<? super P_OUT> action) {
        //evaluate就是開關,一旦呼叫就立即執行整個Stream    
        evaluate(ForEachOps.makeRef(action, false));
    }

執行前會對操作從末尾到起始反向包裹起來,得到呼叫鏈

Sink opWrapSink(int flags, Sink<P_OUT> sink) ;

    //這個Sink是終端操作所對應的Sink
    final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
        Objects.requireNonNull(sink);

        for ( AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
            sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
        }
        return (Sink<P_IN>) sink;
    }

    @Override
    final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
        Objects.requireNonNull(wrappedSink);

        if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
            //依次執行呼叫鏈
            wrappedSink.begin(spliterator.getExactSizeIfKnown());
            spliterator.forEachRemaining(wrappedSink);
            wrappedSink.end();
        }
        else {
            copyIntoWithCancel(wrappedSink, spliterator);
        }
    }

有狀態的中間操作何時執行?

例如sorted()操作,其依賴上一次操作的結果集,按照呼叫鏈來說結果集必須在accept()呼叫完才會產生.那也就說明sorted操作需要在end中,然後再重新開啟呼叫鏈.

sorted的end方法:

       @Override
        public void end() {
            list.sort(comparator);
            downstream.begin(list.size());
            if (!cancellationWasRequested) {
                list.forEach(downstream::accept);
            }
            else {
                for (T t : list) {
                    if (downstream.cancellationRequested()) break;
                    downstream.accept(t);
                }
            }
            downstream.end();
            list = null;
        }

那麼就相當於sorted給原有操作斷路了一次,然後又重新接上,再次遍歷.

如何收集到結果?

foreach是不需要收集到結果的,但是對於collect這樣的操作是需要拿到最終end產生的結果.end產生的結果在最後一個Sink中,這樣的操作最終都會提供一個取出資料的get方法.

       @Override
        public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
                                           Spliterator<P_IN> spliterator) {
            return helper.wrapAndCopyInto(makeSink(), spliterator).get();
        }

如此拿到資料返回