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();
        }
    }

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());
        }
    }

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();
    }

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());
        }
    }

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();
    }

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);
    }

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);
        }
    }

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);
        }
    }

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);
    }

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());
        }
    }

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;
        }
    }

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，存在許多不足之處，請多多指教。