Java Stream過濾實現原理詳解

一、Stream簡介

Stream是Java 8中引入的一種全新的函數式編程方式，可以看做對集合和數組操作的函數化封裝。

Stream可以進行過濾、映射、統計、歸約等多種操作，可以大大簡化代碼複雜度，提高編程效率。

 // Stream是java.util.stream包下的類，需要進行導包  
import java.util.stream.Stream;

二、Stream過濾方法filter()

Stream中的filter方法用於過濾數據，篩選出符合條件的元素。

filter方法接受一個Predicate接口類型的lambda表達式，該表達式用於確定元素是否符合過濾條件。

 // 過濾出大於等於10的數字  
Stream<Integer> stream1 = Stream.of(5, 10, 15, 20);  
Stream<Integer> stream2 = stream1.filter(i -> i >= 10);  
stream2.forEach(System.out::println);  
// 輸出：  
// 10  
// 15  
// 20  

三、Stream過濾實現原理

filter方法實現過濾的關鍵在於，判斷每個元素是否符合條件，只輸出符合條件的元素。

實現這個過程需要滿足兩個條件：

• 每個元素都要被判斷，即需要用到遍歷方法。
• 滿足篩選條件的元素需要輸出，即需要用到輸出方法。

Stream本質上就是對這兩個條件進行封裝，核心就是使用map、reduce等方法實現中間操作，最終使用forEach等方法實現終端操作。

具體流程如下：

1. 獲取數據源——Stream.of或集合的stream方法等。
2. 進行中間操作——如filter方法，進行過濾。
3. 進行終端操作——如forEach方法，輸出符合條件的元素。

四、Stream過濾實現原理示例

通過對Stream源碼的分析，可以看出Stream的具體實現原理。

1. Stream的準備

Stream的創建方式有多種，示例中使用of方法創建一個int類型的Stream：

Stream<Integer> stream = Stream.of(2, 3, 4, 5, 6);

2. filter方法實現

filter方法需要傳入一個Predicate類型的lambda表達式，用於判斷元素是否符合過濾條件。

具體實現過程如下：

public interface Predicate<T> {  
    //進行元素判斷，判斷是否符合過濾條件  
    boolean test(T t);  
}  

public interface Stream<T> extends BaseStream<T, Stream<T>> {  
    // 進行中間操作，如filter方法 
    Stream<T> filter(Predicate<? super T> predicate);  
}  

// 中間操作的具體實現    
@Override  
public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {  
    Objects.requireNonNull(predicate);  
    return new StatelessOp<>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_SIZED) {  
        @Override  
        public Sink<P_OUT> opWrapSink(int flags, Sink<T> sink) {  
            return new Sink.ChainedReference<>(sink) {  
                @Override  
                public void begin(long size) {  
                    downstream.begin(-1);  
                }  
                @Override  
                public void accept(T t) {  
                    if (predicate.test(t)) {  
                        downstream.accept(t);  
                    }  
                }  
            };  
        }  
    };  
}

可以看出filter方法返回了一個新的Stream對象，其中StatelessOp實現了對原Stream的封裝（對外提供的接口還是Stream）。

具體實現了兩個方法：

• opIsStateful：用於判斷Stream是否能重複使用，filter方法中實現為return false。
• opWrapSink：用於返回一個Sink對象，Sink是對元素遍歷輸出的封裝。

opWrapSink返回的是一個匿名類，用於判斷元素是否符合過濾條件：

if (predicate.test(t)) {  
    downstream.accept(t);  
}

3. forEach方法實現

forEach用於對Stream的符合條件的元素進行輸出，具體實現過程如下：

public interface Consumer<T> {  
    void accept(T t);  
}  

public interface Stream<T> extends BaseStream<T, Stream<T>> {  
    // 對符合條件的元素進行逐個操作  
    void forEach(Consumer<? super T> action);  
}  

// 終端操作的具體實現  
@Override  
public final void forEach(Consumer<? super P_OUT> action) {  
    evaluate(ForEachOps.makeRef(action, !isParallel()));  
}

可以看出，forEach方法實際調用了evaluate方法。

evaluate方法的具體實現在PipelineHelper類中，對StatelessOp對象逐個進行輸出操作。

下面是PipelineHelper的部分代碼實現：

 @Override  
        public <E_OUT> TerminalOp<P_OUT, E_OUT> evaluate(TerminalOp<E_OUT, R> terminalOp) {  
            Objects.requireNonNull(terminalOp);  
            if (linkedOrConsumed) {  
                throw new IllegalStateException(MSG_STREAM_LINKED);  
            }  
            linkedOrConsumed = true;  
            return new ChainedTerminal<>(terminalOp).evaluateSequential(this);  
        }

static final class ChainedTerminal<E_IN, E_OUT>  
            implements TerminalOp<E_IN, E_OUT> {  
            private final TerminalOp<?, E_OUT> terminalOp;  

            ChainedTerminal(TerminalOp<?, E_OUT> terminalOp) {  
                this.terminalOp = terminalOp;  
            }  

            @Override  
            public StreamShape inputShape() {  
                return terminalOp.inputShape();  
            }  

            @Override  
            public <S> E_OUT evaluateSequential(PipelineHelper<S> helper, Spliterator<S> spliterator) {  
                return terminalOp.evaluateSequential(helper, ops.wrapSink(terminalOp, helper.wrapSink(terminalOp, terminalSink)  
                ));  
            }  
    }

@Override  
    // Input(depth==0), Output(Vector,X)  
    public Sink<T> wrapSink(PipelineHelper<T> ph, Sink<X> sink) {  
        return opWrapSink(opFlags, sink);  
    }  

五、總結

Java Stream過濾實現原理是在中間操作filter和終端操作forEach的配合下完成的。

Stream的本質是對集合和數組操作的封裝，並不會增加實際的存儲壓力。

Stream的使用可以大大簡化代碼複雜度，提高編程效率。

下面是完整的示例代碼：

import java.util.Objects;  
import java.util.function.Consumer;  
import java.util.function.Predicate;  
import java.util.stream.Stream;  

public class StreamFilterDemo {  
    public static void main(String[] args) {  
        // Stream的創建  
        Stream<Integer> stream = Stream.of(2, 3, 4, 5, 6);  
        // 過濾大於等於4的元素  
        Stream<Integer> stream2 = stream.filter(i -> i >= 4);  
        // 輸出符合條件的元素  
        stream2.forEach(System.out::println);  
    }  
}