java8 Stream的實現原理 (從零開始實現一個stream流)

1.Stream 流的介紹

1.1 java8 stream介紹

　　java8新增了stream流的特性，可以讓用戶以函數式的方式、更爲簡單的操縱集合等數據結構，並實現了用戶無感知的並行計算。

1.2 從零開始實現一個stream流

　　相信不少人在使用過java8的streamAPI接口以後，都會對其實現原理感到好奇，但每每在看到jdk的stream源碼後卻被其複雜的抽象、封裝給弄糊塗了，而沒法很好的理解其背後的原理。究其緣由，是由於jdk的stream源碼是高度工程化的代碼，工程化的代碼爲了效率和知足各式各樣的需求，會將代碼實現的極其複雜，不易理解。

　　在這裏，咱們將拋開jdk的實現思路，從零開始實現一個stream流。

　　咱們的stream流一樣擁有惰性求值，函數式編程接口等特性，並兼容jdk的Collection等數據結構(但不支持並行計算 orz)。

　　相信在親手實現一個stream流的框架以後，你們能更好的理解流計算的原理。

2.stream的優勢

　　在探討探究stream的實現原理和動手實現以前，咱們先要體會stream流計算的獨特之處。

　　舉個例子： 有一個List<Person>列表,咱們須要得到年齡爲70歲的前10個Person的姓名。

過程式的解決方案：

　　稍加思考，咱們很快就寫出了一個過程式的解決方案(僞代碼)：

List<Person> personList = fromDB(); // 得到List<Person>
int limit = 10; // 限制條件
List<String> nameList = new ArrayList(); // 收集的姓名集合
for(Person personItem : personList){
    if(personItem.age == 70){ // 知足條件
        nameList.add(personItem.name); // 加入姓名集合
        if(nameList.size() >= 10){ // 判斷是否超過限制
            break;
        }
    }
}
return nameList;

函數式stream解決方案：

　　下面咱們給出一種基於stream流的解決方案(僞代碼)：

List<Person> personList = fromDB(); // 得到List<Person>
List<String> nameList = personList.stream()
　　　　　　.filter(item->item.age == 70) // 過濾條件
　　　　　　.limit(10)    // limit限制條件
　　　　　　.map(item->item.name) // 得到姓名
　　　　　　.collect(Collector.toList()); // 轉化爲list

return nameList;

兩種方案的不一樣之處：

　　從函數式的角度上看，過程式的代碼實現將收集元素、循環迭代、各類邏輯判斷耦合在一塊兒，暴露了太多細節。當將來需求變更和變得更加複雜的狀況下，過程式的代碼將變得難以理解和維護(須要控制檯打印出 年齡爲70歲的前10個Person中，姓王的Person的名稱）。

　　函數式的解決方案解開了代碼細節和業務邏輯的耦合，相似於sql語句，表達的是"要作什麼"而不是"如何去作"，使程序員能夠更加專一於業務邏輯，寫出易於理解和維護的代碼。

List<Person> personList = fromDB(); // 得到List<Person>
personList.stream()
    .filter(item->item.age == 70) // 過濾條件
    .limit(10)    // limit限制條件
    .filter(item->item.name.startWith("王"))  // 過濾條件
    .map(item->item.name) // 得到姓名
    .forEach(System.out::println);

3.stream API接口介紹

　　stream API的接口是函數式的，儘管java 8也引入了lambda表達式，但java實質上依然是由接口-匿名內部類來實現函數傳參的，因此須要事先定義一系列的函數式接口。

Function: 相似於 y = F(x)

@FunctionalInterface
public interface Function<R,T> {

    /**
     * 函數式接口
     * 相似於 y = F(x)
     * */
    R apply(T t);
}

BiFunction: 相似於 z = F(x,y)

@FunctionalInterface
public interface BiFunction<R, T, U> {

    /**
     * 函數式接口
     * 相似於 z = F(x,y)
     * */
    R apply(T t, U u);
}

ForEach: 遍歷處理

@FunctionalInterface
public interface ForEach <T>{

    /**
     * 迭代器遍歷
     * @param item 被迭代的每一項
     * */
    void apply(T item);
}

Comparator: 比較器

@FunctionalInterface
public interface Comparator<T>  {

    /**
     * 比較方法邏輯
     * @param o1    參數1
     * @param o2    參數2
     * @return      返回值大於0 ---> (o1 > o2)
     *              返回值等於0 ---> (o1 = o2)
     *              返回值小於0 ---> (o1 < o2)
     */
    int compare(T o1, T o2);
}

Predicate: 條件判斷

@FunctionalInterface
public interface Predicate <T>{

    /**
     * 函數式接口
     * @param item 迭代的每一項
     * @return true 知足條件
     *          false 不知足條件
     * */
    boolean satisfy(T item);
}

Supplier：提供初始值

@FunctionalInterface
public interface Supplier<T> {

    /**
     * 提供初始值
     * @return 初始化的值
     * */
    T get();
}

EvalFunction：stream求值函數

@FunctionalInterface
public interface EvalFunction<T> {

    /**
     * stream流的強制求值方法
     * @return 求值返回一個新的stream
     * */
    MyStream<T> apply();
}

stream API接口：

/**
 * stream流的API接口
 */
public interface Stream<T> {

    /**
     * 映射 lazy 惰性求值
     * @param mapper 轉換邏輯 T->R
     * @return 一個新的流
     * */
    <R> MyStream<R> map(Function<R,T> mapper);

    /**
     * 扁平化 映射 lazy 惰性求值
     * @param mapper 轉換邏輯 T->MyStream<R>
     * @return  一個新的流(扁平化以後)
     * */
    <R> MyStream<R> flatMap(Function<? extends MyStream<R>, T> mapper);

    /**
     * 過濾 lazy 惰性求值
     * @param predicate 謂詞判斷
     * @return 一個新的流，其中元素是知足predicate條件的
     * */
    MyStream<T> filter(Predicate<T> predicate);

    /**
     * 截斷 lazy 惰性求值
     * @param n 截斷流，只獲取部分
     * @return 一個新的流，其中的元素不超過 n
     * */
    MyStream<T> limit(int n);

    /**
     * 去重操做 lazy 惰性求值
     * @return 一個新的流，其中的元素不重複(!equals）
     * */
    MyStream<T> distinct();

    /**
     * 窺視 lazy 惰性求值
     * @return 同一個流，peek不改變流的任何行爲
     * */
    MyStream<T> peek(ForEach<T> consumer);

    /**
     * 遍歷 eval 強制求值
     * @param consumer 遍歷邏輯
     * */
    void forEach(ForEach<T> consumer);

    /**
     * 濃縮 eval 強制求值
     * @param initVal 濃縮時的初始值
     * @param accumulator 濃縮時的 累加邏輯
     * @return 濃縮以後的結果
     * */
    <R> R reduce(R initVal, BiFunction<R, R, T> accumulator);

    /**
     * 收集 eval 強制求值
     * @param collector 傳入所需的函數組合子，生成高階函數
     * @return 收集以後的結果
     * */
    <R, A> R collect(Collector<T,A,R> collector);

    /**
     * 最大值 eval 強制求值
     * @param comparator 大小比較邏輯
     * @return 流中的最大值
     * */
    T max(Comparator<T> comparator);

    /**
     * 最小值 eval 強制求值
     * @param comparator 大小比較邏輯
     * @return 流中的最小值
     * */
    T min(Comparator<T> comparator);

    /**
     * 計數 eval 強制求值
     * @return  當前流的個數
     * */
    int count();

    /**
     * 流中是否存在知足predicate的項
     * @return true 存在 匹配項
     *         false 不存在 匹配項
     * */
    boolean anyMatch(Predicate<? super T> predicate);

    /**
     * 流中的元素是否所有知足predicate
     * @return true 所有知足
     *          false 不所有知足
     * */
    boolean allMatch(Predicate<? super T> predicate);

    /**
     * 返回空的 stream
     * @return 空stream
     * */
    static <T> MyStream<T> makeEmptyStream(){
        // isEnd = true
        return new MyStream.Builder<T>().isEnd(true).build();
    }
}

4.MyStream 實現細節

　　簡單介紹了API接口定義以後，咱們開始深刻探討流的內部實現。

　　流由兩個重要的部分所組成，"當前數據項(head)"和"下一數據項的求值函數(nextItemEvalProcess)"。

　　其中，nextItemEvalProcess是流可以實現"惰性求值"的關鍵。

　　

流的基本屬性：

public class MyStream<T> implements Stream<T> {
    /**
     * 流的頭部
     * */
    private T head;

    /**
     * 流的下一項求值函數
     * */
    private NextItemEvalProcess nextItemEvalProcess;

    /**
     * 是不是流的結尾
     * */
    private boolean isEnd;

    public static class Builder<T>{
        private MyStream<T> target;

        public Builder() {
            this.target = new MyStream<>();
        }

        public Builder<T> head(T head){
            target.head = head;
            return this;
        }

        Builder<T> isEnd(boolean isEnd){
            target.isEnd = isEnd;
            return this;
        }

        public Builder<T> nextItemEvalProcess(NextItemEvalProcess nextItemEvalProcess){
            target.nextItemEvalProcess = nextItemEvalProcess;
            return this;
        }

        public MyStream<T> build(){
            return target;
        }
    }

   /**
     * 當前流強制求值
     * @return 求值以後返回一個新的流
     * */
    private MyStream<T> eval(){
        return this.nextItemEvalProcess.eval();
    }

    /**
     * 當前流 爲空
     * */
    private boolean isEmptyStream(){
        return this.isEnd;
    }
}

/**
 * 下一個元素求值過程
 */
public class NextItemEvalProcess {

    /**
     * 求值方法
     * */
    private EvalFunction evalFunction;

    public NextItemEvalProcess(EvalFunction evalFunction) {
        this.evalFunction = evalFunction;
    }

    MyStream eval(){
        return evalFunction.apply();
    }
}

4.1 stream流在使用過程當中的三個階段

　　1.  生成並構造一個流 (List.stream() 等方法)

　　2.  在流的處理過程當中添加、綁定惰性求值流程  (map、filter、limit 等方法)

　　3.  對流使用強制求值函數，生成最終結果 (max、collect、forEach等方法)

4.2 生成並構造一個流

　　流在生成時是"純淨"的，其最初的NextItemEvalProcess求值以後就是指向本身的下一個元素。

　　咱們以一個Integer整數流的生成爲例。IntegerStreamGenerator.getIntegerStream(1,10) 會返回一個流結構，其邏輯上等價於一個從1到10的整數流。但實質是一個惰性求值的stream對象，這裏稱其爲IntStream，其NextItemEvalProcess是一個閉包，方法體是一個遞歸結構的求值函數，其中下界參數low = low + 1。

　　當IntStream第一次被求值時，流開始初始化，isStart = false。當初始化完成以後，每一次求值，都會生成一個新的流對象，其中head(low) = low + 1。當low > high時，流被終止，返回空的流對象。

　　

/**
 * 整數流生成器
 */
public class IntegerStreamGenerator {
    /**
     * 得到一個有限的整數流 介於[low-high]之間
     * @param low 下界
     * @param high 上界
     * */
    public static MyStream<Integer> getIntegerStream(int low, int high){
        return getIntegerStreamInner(low,high,true);
    }

    /**
     * 遞歸函數。配合getIntegerStream(int low,int high)
     * */
    private static MyStream<Integer> getIntegerStreamInner(int low, int high, boolean isStart){
        if(low > high){
            // 到達邊界條件，返回空的流
            return Stream.makeEmptyStream();
        }

        if(isStart){
            return new MyStream.Builder<Integer>()
                    .process(new NextItemEvalProcess(()->getIntegerStreamInner(low,high,false)))
                    .build();
        }else{
            return new MyStream.Builder<Integer>()
                    // 當前元素 low
                    .head(low)
                    // 下一個元素 low+1
                    .process(new NextItemEvalProcess(()->getIntegerStreamInner(low+1,high,false)))
                    .build();
        }
    }
}

　　能夠看到，生成一個流的關鍵在於肯定如何求值下一項元素。對於整數流來講，low = low + 1就是其下一項的求值過程。

　　那麼對於咱們很是關心的jdk集合容器，又該如何生成對應的流呢?

　　答案是Iterator迭代器，jdk的集合容器都實現了Iterator迭代器接口，經過迭代器咱們能夠輕易的取得容器的下一項元素，而不用關心容器內部實現細節。換句話說，只要實現過迭代器接口，就能夠天然的轉化爲stream流，從而得到流計算的全部能力。

/**
 * 集合流生成器
 */
public class CollectionStreamGenerator {
    /**
     * 將一個List轉化爲stream流
     * */
    public static <T> MyStream<T> getListStream(List<T> list){
        return getListStream(list.iterator(),true);
    }

    /**
     * 遞歸函數
     * @param iterator list 集合的迭代器
     * @param isStart 是不是第一次迭代
     * */
    private static <T> MyStream<T> getListStream(Iterator<T> iterator, boolean isStart){
        if(!iterator.hasNext()){
            // 不存在迭代的下一個元素，返回空的流
            return Stream.makeEmptyStream();
        }

        if(isStart){
            // 初始化，只須要設置 求值過程
            return new MyStream.Builder<T>()
                    .nextItemEvalProcess(new NextItemEvalProcess(()-> getListStream(iterator,false)))
                    .build();
        }else{
            // 非初始化，設置head和接下來的求值過程
            return new MyStream.Builder<T>()
                    .head(iterator.next())
                    .nextItemEvalProcess(new NextItemEvalProcess(()-> getListStream(iterator,false)))
                    .build();
        }
    }
}

　　思考一個小問題，如何生成一個無窮的整數流？

4.3 在流的處理過程當中添加、綁定惰性求值流程

　　咱們以map接口舉例說明。API的map接口是一個惰性求值接口，在流執行了map方法後(stream.map())，不會進行任何的求值運算。map在執行時，會生成一個新的求值過程NextItemEvalProcess，新的過程將以前流的求值過程給"包裹"起來了，僅僅是在"流的生成"到"流的最終求值"之間增長了一道處理工序，最終返回了一個新的stream流對象。

　　API.map所依賴的內部靜態map方法是一個惰性求值方法，其每次調用"只會"將當前流的head部分進行map映射操做，而且生成一個新的流。新生成流的NextItemEvalProcess和以前邏輯基本保持一致(遞歸)，惟一的區別是，第二個參數傳入的stream在調用方法以前會被強制求值(eval)後再傳入。

    @Override
    public <R> MyStream<R> map(Function<R, T> mapper) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
                ()->{
                    MyStream myStream = lastNextItemEvalProcess.eval();
                    return map(mapper, myStream);
                }
        );

        // 求值鏈條 加入一個新的process map
        return new MyStream.Builder<R>()
                .nextItemEvalProcess(this.nextItemEvalProcess)
                .build();
    }

   /**
     * 遞歸函數 配合API.map
     * */
    private static <R,T> MyStream<R> map(Function<R, T> mapper, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        R head = mapper.apply(myStream.head);

        return new MyStream.Builder<R>()
                .head(head)
                .nextItemEvalProcess(new NextItemEvalProcess(()->map(mapper, myStream.eval())))
                .build();
    }

　　惰性求值接口的實現大同小異，你們須要體會一下閉包、遞歸、惰性求值等概念，限於篇幅就不一一展開啦。

flatMap：

   @Override
    public <R> MyStream<R> flatMap(Function<? extends MyStream<R>,T> mapper) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
            ()->{
                MyStream myStream = lastNextItemEvalProcess.eval();
                return flatMap(mapper, Stream.makeEmptyStream(), myStream);
            }
        );

        // 求值鏈條 加入一個新的process map
        return new MyStream.Builder<R>()
            .nextItemEvalProcess(this.nextItemEvalProcess)
            .build();
    }

  /**
     * 遞歸函數 配合API.flatMap
     * */
    private static <R,T> MyStream<R> flatMap(Function<? extends MyStream<R>,T> mapper, MyStream<R> headMyStream, MyStream<T> myStream){
        if(headMyStream.isEmptyStream()){
            if(myStream.isEmptyStream()){
                return Stream.makeEmptyStream();
            }else{
                T outerHead = myStream.head;
                MyStream<R> newHeadMyStream = mapper.apply(outerHead);

                return flatMap(mapper, newHeadMyStream.eval(), myStream.eval());
            }
        }else{
            return new MyStream.Builder<R>()
                        .head(headMyStream.head)
                        .nextItemEvalProcess(new NextItemEvalProcess(()-> flatMap(mapper, headMyStream.eval(), myStream)))
                        .build();
        }
    }

View Code

filter：

    @Override
    public MyStream<T> filter(Predicate<T> predicate) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
                ()-> {
                    MyStream myStream = lastNextItemEvalProcess.eval();
                    return filter(predicate, myStream);
                }
        );

        // 求值鏈條 加入一個新的process filter
        return this;
    }

  /**
     * 遞歸函數 配合API.filter
     * */
    private static <T> MyStream<T> filter(Predicate<T> predicate, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        if(predicate.satisfy(myStream.head)){
            return new Builder<T>()
                    .head(myStream.head)
                    .nextItemEvalProcess(new NextItemEvalProcess(()->filter(predicate, myStream.eval())))
                    .build();
        }else{
            return filter(predicate, myStream.eval());
        }
    }

View Code

limit：

   @Override
    public MyStream<T> limit(int n) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
                ()-> {
                    MyStream myStream = lastNextItemEvalProcess.eval();
                    return limit(n, myStream);
                }
        );

        // 求值鏈條 加入一個新的process limit
        return this;
    }

  /**
     * 遞歸函數 配合API.limit
     * */
    private static <T> MyStream<T> limit(int num, MyStream<T> myStream){
        if(num == 0 || myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        return new MyStream.Builder<T>()
                .head(myStream.head)
                .nextItemEvalProcess(new NextItemEvalProcess(()->limit(num-1, myStream.eval())))
                .build();
    }

View Code

distinct：

   @Override
    public MyStream<T> distinct() {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
            ()-> {
                MyStream myStream = lastNextItemEvalProcess.eval();
                return distinct(new HashSet<>(), myStream);
            }
        );

        // 求值鏈條 加入一個新的process limit
        return this;
    }

  /**
     * 遞歸函數 配合API.distinct
     * */
    private static <T> MyStream<T> distinct(Set<T> distinctSet,MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        if(!distinctSet.contains(myStream.head)){
            // 加入集合
            distinctSet.add(myStream.head);

            return new Builder<T>()
                .head(myStream.head)
                .nextItemEvalProcess(new NextItemEvalProcess(()->distinct(distinctSet, myStream.eval())))
                .build();
        }else{
            return distinct(distinctSet, myStream.eval());
        }
    }

View Code

peek：

    @Override
    public MyStream<T> peek(ForEach<T> consumer) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
            ()-> {
                MyStream myStream = lastNextItemEvalProcess.eval();
                return peek(consumer,myStream);
            }
        );

        // 求值鏈條 加入一個新的process peek
        return this;
    }

  /**
     * 遞歸函數 配合API.peek
     * */
    private static <T> MyStream<T> peek(ForEach<T> consumer,MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        consumer.apply(myStream.head);

        return new MyStream.Builder<T>()
            .head(myStream.head)
            .nextItemEvalProcess(new NextItemEvalProcess(()->peek(consumer, myStream.eval())))
            .build();
    }

View Code

4.4 對流使用強制求值函數，生成最終結果

　　咱們以forEach方法舉例說明。強制求值方法forEach會不斷的對當前stream進行求值並讓consumer接收處理，直到當前流成爲空流。

有兩種可能的狀況會致使遞歸傳入的流參數成爲空流(empty-stream)：

　　1. 最初生成流的求值過程返回了空流(整數流，low > high 時，返回空流 )

　　2. limit之類的短路操做，會提早終止流的求值返回空流(n == 0 時，返回空流)

    @Override
    public void forEach(ForEach<T> consumer) {
        // 終結操做 直接開始求值
        forEach(consumer,this.eval());
    }
    
    /**
     * 遞歸函數 配合API.forEach
     * */
    private static <T> void forEach(ForEach<T> consumer, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return;
        }

        consumer.apply(myStream.head);
        forEach(consumer, myStream.eval());
    }

　　強制求值的接口的實現也都大同小異，限於篇幅就不一一展開啦。

reduce：

  /**
     * 遞歸函數 配合API.reduce
     * */
    private static <R,T> R reduce(R initVal, BiFunction<R,R,T> accumulator, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return initVal;
        }

        T head = myStream.head;
        R result = reduce(initVal,accumulator, myStream.eval());

        return accumulator.apply(result,head);
    }

  /**
     * 遞歸函數 配合API.reduce
     * */
    private static <R,T> R reduce(R initVal, BiFunction<R,R,T> accumulator, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return initVal;
        }

        T head = myStream.head;
        R result = reduce(initVal,accumulator, myStream.eval());

        return accumulator.apply(result,head);
    }

View Code

max：

   @Override
    public T max(Comparator<T> comparator) {
        // 終結操做 直接開始求值
        MyStream<T> eval = this.eval();

        if(eval.isEmptyStream()){
            return null;
        }else{
            return max(comparator,eval,eval.head);
        }
    }

    /**
     * 遞歸函數 配合API.max
     * */
    private static <T> T max(Comparator<T> comparator, MyStream<T> myStream, T max){
        if(myStream.isEnd){
            return max;
        }

        T head = myStream.head;
        // head 和 max 進行比較
        if(comparator.compare(head,max) > 0){
            // head 較大 做爲新的max傳入
            return max(comparator, myStream.eval(),head);
        }else{
            // max 較大 不變
            return max(comparator, myStream.eval(),max);
        }
    }

View Code

min：

   @Override
    public T min(Comparator<T> comparator) {
        // 終結操做 直接開始求值
        MyStream<T> eval = this.eval();

        if(eval.isEmptyStream()){
            return null;
        }else{
            return min(comparator,eval,eval.head);
        }
    }

  /**
     * 遞歸函數 配合API.min
     * */
    private static <T> T min(Comparator<T> comparator, MyStream<T> myStream, T min){
        if(myStream.isEnd){
            return min;
        }

        T head = myStream.head;
        // head 和 min 進行比較
        if(comparator.compare(head,min) < 0){
            // head 較小 做爲新的min傳入
            return min(comparator, myStream.eval(),head);
        }else{
            // min 較小 不變
            return min(comparator, myStream.eval(),min);
        }
    }

View Code

count：

   @Override
    public int count() {
        // 終結操做 直接開始求值
        return count(this.eval(),0);
    }

  /**
     * 遞歸函數 配合API.count
     * */
    private static <T> int count(MyStream<T> myStream, int count){
        if(myStream.isEmptyStream()){
            return count;
        }

        // count+1 進行遞歸
        return count(myStream.eval(),count+1);
    }

View Code

anyMatch：

   @Override
    public boolean anyMatch(Predicate<? super T> predicate) {
        // 終結操做 直接開始求值
        return anyMatch(predicate,this.eval());
    }
    
  /**
     * 遞歸函數 配合API.anyMatch
     * */
    private static <T> boolean anyMatch(Predicate<? super T> predicate,MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            // 截止末尾，不存在任何匹配項
            return false;
        }

        // 謂詞判斷
        if(predicate.satisfy(myStream.head)){
            // 匹配 存在匹配項 返回true
            return true;
        }else{
            // 不匹配，繼續檢查，直到存在匹配項
            return anyMatch(predicate,myStream.eval());
        }
    }

View Code

allMatch：

   @Override
    public boolean allMatch(Predicate<? super T> predicate) {
        // 終結操做 直接開始求值
        return allMatch(predicate,this.eval());
    }
    
    /**
     * 遞歸函數 配合API.anyMatch
     * */
    private static <T> boolean allMatch(Predicate<? super T> predicate,MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            // 所有匹配
            return true;
        }

        // 謂詞判斷
        if(predicate.satisfy(myStream.head)){
            // 當前項匹配，繼續檢查
            return allMatch(predicate,myStream.eval());
        }else{
            // 存在不匹配的項，返回false
            return false;
        }
    }

View Code

4.5 collect方法

　　collect方法是強制求值方法中，最複雜也最強大的接口，其做用是將流中的元素收集(collect)起來，並轉化成特定的數據結構。

　　從函數式編程的角度來看，collect方法是一個高階函數，其接受三個函數做爲參數(supplier，accumulator，finisher)，最終生成一個更增強大的函數。在java中，三個函數參數以Collector實現對象的形式呈現。

　　supplier 方法：用於提供收集collect的初始值。

　　accumulator 方法：用於指定收集過程當中，初始值和流中個體元素聚合的邏輯。

　　finnisher 方法：用於指定在收集完成以後的收尾轉化操做(例如：StringBuilder.toString() ---> String)。

collect接口實現：

    @Override
    public <R, A> R collect(Collector<T, A, R> collector) {
        // 終結操做 直接開始求值
        A result = collect(collector,this.eval());

        // 經過finish方法進行收尾
        return collector.finisher().apply(result);
    }

    /**
     * 遞歸函數 配合API.collect
     * */
    private static <R, A, T> A collect(Collector<T, A, R> collector, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return collector.supplier().get();
        }

        T head = myStream.head;
        A tail = collect(collector, myStream.eval());

        return collector.accumulator().apply(tail,head);
    }

collector接口：

/**
 * collect接口 收集器
 * 經過傳入組合子，生成高階過程
 */
public interface Collector<T, A, R> {

    /**
     * 收集時，提供初始化的值
     * */
    Supplier<A> supplier();

    /**
     * A = A + T
     * 累加器，收集時的累加過程
     * */
    BiFunction<A, A, T> accumulator();

    /**
     * 收集完成以後的收尾操做
     * */
    Function<A, R> finisher();
}

　　瞭解jdk源碼的讀者可能會注意到，jdk的stream實現中collector接口多了一個combiner接口，combiner接口用於指定並行計算以後的結果集合並的邏輯，因爲咱們的實現不支持並行計算，所以也不須要添加combiner接口了。

　　同時，jdk還提供了一個Collectors工具類，很好的知足了平時常見的需求(Collector.toList()、Collctor.groupingBy())等等。但特殊時刻仍是須要用戶本身指定collect傳入的參數，精細的控制處理邏輯的，所以仍是有必要了解一下collect方法內部原理的。

stream.collect()參數經常使用工具類：

/**
 * stream.collect() 參數經常使用工具類
 */
public class CollectUtils {
    /**
     * stream 轉換爲 List
     * */
    public static <T> Collector<T, List<T>, List<T>> toList(){
        return new Collector<T, List<T>, List<T>>() {
            @Override
            public Supplier<List<T>> supplier() {
                return ArrayList::new;
            }

            @Override
            public BiFunction<List<T>, List<T>, T> accumulator() {
                return (list, item) -> {
                    list.add(item);
                    return list;
                };
            }

            @Override
            public Function<List<T>, List<T>> finisher() {
                return list -> list;
            }
        };
    }

    /**
     * stream 轉換爲 Set
     * */
    public static <T> Collector<T, Set<T>, Set<T>> toSet(){
        return new Collector<T, Set<T>, Set<T>>() {
            @Override
            public Supplier<Set<T>> supplier() {
                return HashSet::new;
            }

            @Override
            public BiFunction<Set<T>, Set<T>, T> accumulator() {
                return (set, item) -> {
                    set.add(item);
                    return set;
                };
            }

            @Override
            public Function<Set<T>, Set<T>> finisher() {
                return set -> set;
            }
        };
    }
}

4.6 舉例分析

　　咱們選擇一個簡單而又不失通常性的例子，串聯起這些內容。經過完整的描述一個流求值的全過程，加深你們對流的理解。

public static void main(String[] args){
        Integer sum = IntegerStreamGenerator.getIntegerStream(1,10)
                    .filter(item-> item%2 == 0) // 過濾出偶數
                    .map(item-> item * item)    // 映射爲平方
                    .limit(2)                   // 截取前兩個
                    .reduce(0,(i1,i2)-> i1+i2); // 最終結果累加求和(初始值爲0)

        System.out.println(sum); // 20
    }

　　因爲咱們的stream實現採用的是鏈式編程的方式，不太好理解，將其展開爲邏輯等價的形式。

public static void main(String[] args){
　　　　 // 生成整數流 1-10
        Stream<Integer> intStream = IntegerStreamGenerator.getIntegerStream(1,10);

        // intStream基礎上過濾出偶數
        Stream<Integer> filterStream =  intStream.filter(item-> item%2 == 0);

        // filterStream基礎上映射爲平方
        Stream<Integer> mapStream = filterStream.map(item-> item * item);

        // mapStream基礎上截取前兩個
        Stream<Integer> limitStream = mapStream.limit(2);

        // 最終結果累加求和(初始值爲0)
        Integer sum = limitStream.reduce(0,(i1,i2)-> i1+i2);

        System.out.println(sum); // 20
    }

reduce強制求值操做以前的執行過程圖：

 

reduce強制求值過程當中的執行過程圖 ：

　　能夠看到，stream的求值過程並不會一口氣將初始的流所有求值，而是按需的、一個一個的進行求值。

　　stream的一次求值過程至多隻會遍歷流中元素一次；若是存在短路操做(limit、anyMatch等)，實際迭代的次數會更少。

　　所以沒必要擔憂多層的map、filter處理邏輯的嵌套會讓流進行屢次迭代，致使效率急劇降低。

　　

5.stream 總結

5.1 當前版本缺陷

1. 遞歸調用效率較低

　　爲了代碼的簡潔性和更加的函數式，當前實現中不少地方都用遞歸代替了循環迭代。

　　雖然邏輯上遞歸和迭代是等價的，但在目前的計算機硬件上，每一層的遞歸調用都會使得函數調用棧增大，而即便是明顯的尾遞歸調用，java目前也沒有能力進行優化。當流須要處理的數據量很大時，將會出現棧溢出，棧空間不足之類的系統錯誤。

　　將遞歸優化爲迭代可以顯著提升當前版本流的執行效率。

2. API接口較少

　　限於篇幅，咱們只提供了一些較爲經常使用的API接口。在jdk中，Collector工具類提供了不少方便易用的接口；對於同一API接口也提供了多種重載函數給用戶使用。

　　以目前已有的功能爲基礎，提供一些更加方便的接口並不困難。

3. 不支持並行計算

　　因爲流在求值計算時生成的是對象的副本，是無反作用的，很適合經過數據分片執行並行計算。限於我的水平，在設計之初並無考慮將並行計算這一特性加入進來。

5.2 函數式編程

　　仔細分析整個流的執行過程，與其說流是一個對象，不如說流是一個高階函數(higher-order function)。每當map、filter綁定了一個流，新生成的流實際上是一個更加複雜的函數；每一層封裝，都會使新生成的流這一高階函數比起原基礎變得更增強大和複雜。map、filter就像一個個的基礎算子，在接收對應的過程後(filter(過濾出偶數)、map(平方映射))，能夠不斷的疊加，完成許許多多很是複雜的操做。

　　這也是函數式編程的中心思想之一：將計算過程轉化爲一系列嵌套函數的調用。

5.3 總結

　　最初是在學習《計算機程序的構造和解釋》(SICP)中stream流計算時突發奇想的，想着能不能用java來實現一個和書上相似的流計算框架，能和jdk的stream流功能大體相同，最終，經過反覆地思考和嘗試纔將心中所想以java代碼的形式呈現出來。

　　SICP是一本小衆但別具一格的計算機書籍，許多人認爲它不太實用。我我的認爲，雖然計算機技術發展突飛猛進，可是計算機技術的基礎理論卻每每變化緩慢，若是可以抓住技術發展背後那不變的元知識，就不容易在技術的浪潮中失去方向。SICP就是這樣一本教授計算機科學元知識的書籍，雖然一開始有點枯燥，卻能慢慢品味出其美妙之處。

　　但願你們在閱讀完這篇博客以後，能更好的理解流計算，更好的理解函數式編程。

　　SICP公開課視頻(中英字幕)：https://www.bilibili.com/video/av8515129。

　　這篇博客的完整代碼在個人github上：https://github.com/1399852153/Streamjava，存在許多不足之處，請多多指教。