上篇Stream Head的构建(一)我们用Collection.stream构建了源头，长相如下图

stream_source_head.png

此篇以集合为源，看一个简单的流的中间操作。

List<String> names = Arrays.asList("one", "two", "three", "four");
 Stream<Integer> stream = names.stream()
                //具体Lambda表达式或方法引用先不用管
                .filter(s -> s.length() > 2)
                .map(String::toUpperCase);

在上图中把源头设为stream0，接下来filter(...)在stream0上操作，返回一个中间流StatelessOp,和Head一样也是ReferencePipeline子类

ReferencePipeline.java
public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
        Objects.requireNonNull(predicate);
        //这里this就是stream0
        return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                     StreamOpFlag.NOT_SIZED) {
            @Override
            public Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
            //这里实现了重要的抽象方法opWrapSink，以一个Sink作为参数，又返回一个Sink，先忽略
               ...
            }
        };
    }

我们只看StatelessOp的构造方法

public StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,inputShape,opFlags) {
            super(upstream, opFlags); 
        }

和Head一样也是调用的AbstractPipeline构造方法，只不过Head调用的是专门构造源头的，StatelessOp或StatefulOp调用的是专门构造中间管道的。

AbstractPipeline.java
//previousStage就是stream0
AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
        if (previousStage.linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        previousStage.linkedOrConsumed = true;
        //上个stream的next指向自己
        previousStage.nextStage = this;
        //自己的previous指向上个 stream
        this.previousStage = previousStage;
        this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
        this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
        this.sourceStage = previousStage.sourceStage;
        if (opIsStateful())
            sourceStage.sourceAnyStateful = true;
        this.depth = previousStage.depth + 1;
    }

filter方法调用完后，如下图

stream_filter.png

再看map，在stream1上操作，和filter雷同

ReferencePipeline.java
public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
        Objects.requireNonNull(mapper);
        return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
            @Override
            public Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
                ...同上
            }
        };
    }

stream_map.png

stream_structure.png