CH10-Stream

// java.util.Collection部分源码
public interface Collection<E> extends Iterable<E> {

    // 省略其他代码

    @Override
    default Spliterator<E> spliterator() {
        return Spliterators.spliterator(this, 0);
    }

    default Stream<E> stream() {
        return StreamSupport.stream(spliterator(), false);
    }

    default Stream<E> parallelStream() {
        return StreamSupport.stream(spliterator(), true);
    }
}

// java.lang.Iterable部分源码
public interface Iterable<T> {

    // 省略其他代码

    default void forEach(Consumer<? super T> action) {
        Objects.requireNonNull(action);
        for (T t : this) {
            action.accept(t);
        }
    }

    default Spliterator<T> spliterator() {
        return Spliterators.spliteratorUnknownSize(iterator(), 0);
    }
}

// JDK1.7中的java.lang.Iterable
public interface Iterable<T> {

    Iterator<T> iterator();
}

// JDK1.7中的Iterable实现
public MyIterable<Long> implements Iterable<Long>{

    public Iterator<Long> iterator(){
         ....
    }
}

public interface Spliterator<T> {

    // 暂时省略其他代码

    boolean tryAdvance(Consumer<? super T> action);

    default void forEachRemaining(Consumer<? super T> action) {
        do { } while (tryAdvance(action));
    }

    Spliterator<T> trySplit();

    long estimateSize();

    default long getExactSizeIfKnown() {
        return (characteristics() & SIZED) == 0 ? -1L : estimateSize();
    }

    int characteristics();

    default boolean hasCharacteristics(int characteristics) {
        return (characteristics() & characteristics) == characteristics;
    }

    default Comparator<? super T> getComparator() {
        throw new IllegalStateException();
    }

    // 暂时省略其他代码
}

public static void main(String[] args) throws Exception {
    List<Integer> list = new ArrayList<>();
    list.add(2);
    list.add(1);
    list.add(3);
    Spliterator<Integer> spliterator = list.stream().spliterator();
    final AtomicInteger round = new AtomicInteger(1);
    final AtomicInteger loop = new AtomicInteger(1);
    while (spliterator.tryAdvance(num -> System.out.printf("第%d轮回调Action,值:%d\n", round.getAndIncrement(), num))) {
        System.out.printf("第%d轮循环\n", loop.getAndIncrement());
    }
}

// 控制台输出
第1轮回调Action,值:2
第1轮循环
第2轮回调Action,值:1
第2轮循环
第3轮回调Action,值:3
第3轮循环

public static void main(String[] args) {
    List<Integer> list = new ArrayList<>();
    list.add(2);
    list.add(1);
    list.add(3);
    Spliterator<Integer> spliterator = list.stream().spliterator();
    final AtomicInteger round = new AtomicInteger(1);
    spliterator.forEachRemaining(num -> System.out.printf("第%d轮回调Action,值:%d\n", round.getAndIncrement(), num));
}

// 控制台输出
第1轮回调Action,值:2
第2轮回调Action,值:1
第3轮回调Action,值:3

public static void main(String[] args) throws Exception {
    List<Integer> list = new ArrayList<>();
    list.add(2);
    list.add(3);
    list.add(4);
    list.add(1);
    Spliterator<Integer> first = list.stream().spliterator();
    Spliterator<Integer> second = first.trySplit();
    first.forEachRemaining(num -> {
        System.out.printf("first spliterator item: %d\n", num);
    });
    second.forEachRemaining(num -> {
        System.out.printf("second spliterator item: %d\n", num);
    });
}

// 控制台输出
first spliterator item: 4
first spliterator item: 1
second spliterator item: 2
second spliterator item: 3

public static void main(String[] args) throws Exception {
    List<Integer> list = new ArrayList<>();
    list.add(2);
    list.add(3);
    list.add(4);
    list.add(1);
    Spliterator<Integer> spliterator = list.stream().spliterator();
    System.out.println(spliterator.estimateSize());
}

// 控制台输出
4

public static void main(String[] args) throws Exception {
    List<Integer> list = new ArrayList<>();
    list.add(2);
    list.add(3);
    list.add(4);
    list.add(1);
    Spliterator<Integer> spliterator = list.stream().spliterator();
    System.out.println(spliterator.getExactSizeIfKnown());
}

// 控制台输出
4

// ArrayList#spliterator()
public Spliterator<E> spliterator() {
    return new ArrayListSpliterator(0, -1, 0);
}

// ArrayList中内部类ArrayListSpliterator
final class ArrayListSpliterator implements Spliterator<E> {
    
    // 当前的处理的元素索引值，其实是剩余元素的下边界值(包含)，在tryAdvance()或者trySplit()方法中被修改，一般初始值为0
    private int index;
    // 栅栏，其实是元素索引值的上边界值(不包含)，一般初始化的时候为-1，使用时具体值为元素索引值上边界加1
    private int fence;
    // 预期的修改次数，一般初始化值等于modCount
    private int expectedModCount;

    ArrayListSpliterator(int origin, int fence, int expectedModCount) {
        this.index = origin;
        this.fence = fence;
        this.expectedModCount = expectedModCount;
    }
    
    // 获取元素索引值的上边界值，如果小于0，则把hi和fence都赋值为(ArrayList中的)size，expectedModCount赋值为(ArrayList中的)modCount，返回上边界值
    // 这里注意if条件中有赋值语句hi = fence，也就是此方法调用过程中临时变量hi总是重新赋值为fence，fence是ArrayListSpliterator实例中的成员属性
    private int getFence() {
        int hi;
        if ((hi = fence) < 0) {
            expectedModCount = modCount;
            hi = fence = size;
        }
        return hi;
    }
    
    // Spliterator自分割，这里采用了二分法
    public ArrayListSpliterator trySplit() {
        // hi等于当前ArrayListSpliterator实例中的fence变量，相当于获取剩余元素的上边界值
        // lo等于当前ArrayListSpliterator实例中的index变量，相当于获取剩余元素的下边界值
        // mid = (lo + hi) >>> 1，这里的无符号右移动1位运算相当于(lo + hi)/2
        int hi = getFence(), lo = index, mid = (lo + hi) >>> 1;
        // 当lo >= mid的时候为不可分割，返回NULL，否则，以index = lo,fence = mid和expectedModCount = expectedModCount创建一个新的ArrayListSpliterator
        // 这里有个细节之处，在新的ArrayListSpliterator构造参数中，当前的index被重新赋值为index = mid，这一点容易看漏，老程序员都喜欢做这样的赋值简化
        // lo >= mid返回NULL的时候，不会创建新的ArrayListSpliterator，也不会修改当前ArrayListSpliterator中的参数
        return (lo >= mid) ? null : new ArrayListSpliterator(lo, index = mid, expectedModCount);
    }
    
    // tryAdvance实现
    public boolean tryAdvance(Consumer<? super E> action) {
        if (action == null)
            throw new NullPointerException();
        // 获取迭代的上下边界
        int hi = getFence(), i = index;
        // 由于前面分析下边界是包含关系，上边界是非包含关系，所以这里要i < hi而不是i <= hi
        if (i < hi) {
            index = i + 1;
            // 这里的elementData来自ArrayList中，也就是前文经常提到的元素数组容器，这里是直接通过元素索引访问容器中的数据
            @SuppressWarnings("unchecked") E e = (E)elementData[i];
            // 对传入的Action进行回调
            action.accept(e);
            // 并发修改异常判断
            if (modCount != expectedModCount)
                throw new ConcurrentModificationException();
            return true;
        }
        return false;
    }
    
    // forEachRemaining实现，这里没有采用默认实现，而是完全覆盖实现一个新方法
    public void forEachRemaining(Consumer<? super E> action) {
        // 这里会新建所需的中间变量，i为index的中间变量，hi为fence的中间变量，mc为expectedModCount的中间变量
        int i, hi, mc;
        Object[] a;
        if (action == null)
            throw new NullPointerException();
        // 判断容器数组存在性
        if ((a = elementData) != null) {
            // hi、fence和mc初始化
            if ((hi = fence) < 0) {
                mc = modCount;
                hi = size;
            }
            else
                mc = expectedModCount;
            // 这里就是先做参数合法性校验，再遍历临时数组容器a中中[i,hi)的剩余元素对传入的Action进行回调
            // 这里注意有一处隐蔽的赋值(index = hi)，下界被赋值为上界，意味着每个ArrayListSpliterator实例只能调用一次forEachRemaining()方法
            if ((i = index) >= 0 && (index = hi) <= a.length) {
                for (; i < hi; ++i) {
                    @SuppressWarnings("unchecked") E e = (E) a[i];
                    action.accept(e);
                }
                // 这里校验ArrayList的modCount和mc是否一致，理论上在forEachRemaining()遍历期间，不能对数组容器进行元素的新增或者移除，一旦发生modCount更变会抛出异常
                if (modCount == mc)
                    return;
            }
        }
        throw new ConcurrentModificationException();
    }
    
    // 获取剩余元素估计值，就是用剩余元素索引上边界直接减去下边界
    public long estimateSize() {
        return getFence() - index;
    }
    
    // 具备ORDERED、SIZED和SUBSIZED特性
    public int characteristics() {
        return Spliterator.ORDERED | Spliterator.SIZED | Spliterator.SUBSIZED;
    }
}

public static void main(String[] args) {
    List<Integer> list = new ArrayList<>();
    list.add(2);
    list.add(1);
    list.add(3);
    Spliterator<Integer> spliterator = list.stream().spliterator();
    final AtomicInteger round = new AtomicInteger(1);
    spliterator.forEachRemaining(num -> System.out.printf("[第一次遍历forEachRemaining]第%d轮回调Action,值:%d\n", round.getAndIncrement(), num));
    round.set(1);
    spliterator.forEachRemaining(num -> System.out.printf("[第二次遍历forEachRemaining]第%d轮回调Action,值:%d\n", round.getAndIncrement(), num));
}

// 控制台输出
[第一次遍历forEachRemaining]第1轮回调Action,值:2
[第一次遍历forEachRemaining]第2轮回调Action,值:1
[第一次遍历forEachRemaining]第3轮回调Action,值:3

public class ConcurrentSplitCalculateSum {

    private static class ForkTask extends Thread {

        private int result = 0;

        private final Spliterator<Integer> spliterator;
        private final CountDownLatch latch;

        public ForkTask(Spliterator<Integer> spliterator,
                        CountDownLatch latch) {
            this.spliterator = spliterator;
            this.latch = latch;
        }

        @Override
        public void run() {
            long start = System.currentTimeMillis();
            spliterator.forEachRemaining(num -> result = result + num);
            long end = System.currentTimeMillis();
            System.out.printf("线程[%s]完成计算任务,当前段计算结果:%d,耗时:%d ms\n",
                    Thread.currentThread().getName(), result, end - start);
            latch.countDown();
        }

        public int result() {
            return result;
        }
    }

    private static int join(List<ForkTask> tasks) {
        int result = 0;
        for (ForkTask task : tasks) {
            result = result + task.result();
        }
        return result;
    }

    private static final int THREAD_NUM = 4;

    public static void main(String[] args) throws Exception {
        List<Integer> source = new ArrayList<>();
        for (int i = 1; i < 101; i++) {
            source.add(i);
        }
        Spliterator<Integer> root = source.stream().spliterator();
        List<Spliterator<Integer>> spliteratorList = new ArrayList<>();
        Spliterator<Integer> x = root.trySplit();
        Spliterator<Integer> y = x.trySplit();
        Spliterator<Integer> z = root.trySplit();
        spliteratorList.add(root);
        spliteratorList.add(x);
        spliteratorList.add(y);
        spliteratorList.add(z);
        List<ForkTask> tasks = new ArrayList<>();
        CountDownLatch latch = new CountDownLatch(THREAD_NUM);
        for (int i = 0; i < THREAD_NUM; i++) {
            ForkTask task = new ForkTask(spliteratorList.get(i), latch);
            task.setName("fork-task-" + (i + 1));
            tasks.add(task);
        }
        tasks.forEach(Thread::start);
        latch.await();
        int result = join(tasks);
        System.out.println("最终计算结果为:" + result);
    }
}

// 控制台输出结果
线程[fork-task-4]完成计算任务,当前段计算结果:1575,耗时:0 ms
线程[fork-task-2]完成计算任务,当前段计算结果:950,耗时:1 ms
线程[fork-task-3]完成计算任务,当前段计算结果:325,耗时:1 ms
线程[fork-task-1]完成计算任务,当前段计算结果:2200,耗时:1 ms
最终计算结果为:5050

假设：byte characteristics()  => 也就是最多8个位用于表示特性集合，如果每个位只表示一种特性，那么可以总共表示8种特性
特性X：0000 0001
特性Y：0000 0010
以此类推
假设：characteristics = X | Y = 0000 0001 | 0000 0010 = 0000 0011
那么：characteristics & X = 0000 0011 & 0000 0001 = 0000 0001
判断characteristics是否包含X：(characteristics & X) == X

// 返回特性集合
int characteristics();

// 基于位运算判断特性集合中是否存在输入的特性
default boolean hasCharacteristics(int characteristics) {
    return (characteristics() & characteristics) == characteristics;
}

public class CharacteristicsCheck {

    public static void main(String[] args) {
        System.out.printf("是否存在ORDERED特性:%s\n", hasCharacteristics(Spliterator.ORDERED));
        System.out.printf("是否存在SIZED特性:%s\n", hasCharacteristics(Spliterator.SIZED));
        System.out.printf("是否存在DISTINCT特性:%s\n", hasCharacteristics(Spliterator.DISTINCT));
    }

    private static int characteristics() {
        return Spliterator.ORDERED | Spliterator.SIZED | Spliterator.SORTED;
    }

    private static boolean hasCharacteristics(int characteristics) {
        return (characteristics() & characteristics) == characteristics;
    }
}

// 控制台输出
是否存在ORDERED特性:true
是否存在SIZED特性:true
是否存在DISTINCT特性:false

enum StreamOpFlag {

    // 暂时忽略其他代码

    // 类型枚举，Stream相关实体类型
    enum Type {
         
        // SPLITERATOR类型，关联所有和Spliterator相关的特性
        SPLITERATOR,

        // STREAM类型，关联所有和Stream相关的标志
        STREAM,

        // STREAM类型，关联所有和Stream中间操作相关的标志
        OP,

        // TERMINAL_OP类型，关联所有和Stream终结操作相关的标志
        TERMINAL_OP,

        // UPSTREAM_TERMINAL_OP类型，关联所有在最后一个有状态操作边界上游传播的终止操作标志
        // 这个类型的意义直译有点拗口，不过实际上在JDK11源码中，这个类型没有被流相关功能引用，暂时可以忽略
        UPSTREAM_TERMINAL_OP
    }

    // 设置/注入标志的bit模式，二进制数0001，十进制数1
    private static final int SET_BITS = 0b01;

    // 清除标志的bit模式，二进制数0010，十进制数2
    private static final int CLEAR_BITS = 0b10;

    // 保留标志的bit模式，二进制数0011，十进制数3
    private static final int PRESERVE_BITS = 0b11;
    
    // 掩码建造器工厂方法，注意这个方法用于实例化MaskBuilder
    private static MaskBuilder set(Type t) {
        return new MaskBuilder(new EnumMap<>(Type.class)).set(t);
    }
    
    // 私有静态内部类，掩码建造器，里面的map由上面的set(Type t)方法得知是EnumMap实例
    private static class MaskBuilder {
        // Type -> SET_BITS|CLEAR_BITS|PRESERVE_BITS|0
        final Map<Type, Integer> map;

        MaskBuilder(Map<Type, Integer> map) {
            this.map = map;
        }
       
        // 设置类型和对应的掩码
        MaskBuilder mask(Type t, Integer i) {
            map.put(t, i);
            return this;
        }
        
        // 对类型添加/inject
        MaskBuilder set(Type t) {
            return mask(t, SET_BITS);
        }

        MaskBuilder clear(Type t) {
            return mask(t, CLEAR_BITS);
        }

        MaskBuilder setAndClear(Type t) {
            return mask(t, PRESERVE_BITS);
        }
        
        // 这里的build方法对于类型中的NULL掩码填充为0，然后把map返回
        Map<Type, Integer> build() {
            for (Type t : Type.values()) {
                map.putIfAbsent(t, 0b00);
            }
            return map;
        }
    }
    
    // 类型->掩码映射
    private final Map<Type, Integer> maskTable;
    
    // bit的起始偏移量，控制下面set、clear和preserve的起始偏移量
    private final int bitPosition;

    // set/inject的bit set(map)，其实准确来说应该是一个表示set/inject的bit map
    private final int set;

    // clear的bit set(map)，其实准确来说应该是一个表示clear的bit map
    private final int clear;

    // preserve的bit set(map)，其实准确来说应该是一个表示preserve的bit map
    private final int preserve;

    private StreamOpFlag(int position, MaskBuilder maskBuilder) {
        // 这里会基于MaskBuilder初始化内部的EnumMap
        this.maskTable = maskBuilder.build();
        // Two bits per flag <= 这里会把入参position放大一倍
        position *= 2;
        this.bitPosition = position;
        this.set = SET_BITS << position; // 设置/注入标志的bit模式左移2倍position
        this.clear = CLEAR_BITS << position; // 清除标志的bit模式左移2倍position
        this.preserve = PRESERVE_BITS << position; // 保留标志的bit模式左移2倍position
    }

    // 省略中间一些方法

    // 下面这些静态变量就是直接返回标志对应的set/injec、清除和保留的bit map
    /**
     * The bit value to set or inject {@link #DISTINCT}.
     */
    static final int IS_DISTINCT = DISTINCT.set;

    /**
     * The bit value to clear {@link #DISTINCT}.
     */
    static final int NOT_DISTINCT = DISTINCT.clear;

    /**
     * The bit value to set or inject {@link #SORTED}.
     */
    static final int IS_SORTED = SORTED.set;

    /**
     * The bit value to clear {@link #SORTED}.
     */
    static final int NOT_SORTED = SORTED.clear;

    /**
     * The bit value to set or inject {@link #ORDERED}.
     */
    static final int IS_ORDERED = ORDERED.set;

    /**
     * The bit value to clear {@link #ORDERED}.
     */
    static final int NOT_ORDERED = ORDERED.clear;

    /**
     * The bit value to set {@link #SIZED}.
     */
    static final int IS_SIZED = SIZED.set;

    /**
     * The bit value to clear {@link #SIZED}.
     */
    static final int NOT_SIZED = SIZED.clear;

    /**
     * The bit value to inject {@link #SHORT_CIRCUIT}.
     */
    static final int IS_SHORT_CIRCUIT = SHORT_CIRCUIT.set;
}

// 纵向看
DISTINCT Flag:
maskTable: {
    SPLITERATOR:           0000 0001,
    STREAM:                0000 0001,
    OP:                    0000 0011,
    TERMINAL_OP:           0000 0000,
    UPSTREAM_TERMINAL_OP:  0000 0000
}
position(input): 0
bitPosition:     0
set:             1 => 0000 0000 0000 0000 0000 0000 0000 0001
clear:           2 => 0000 0000 0000 0000 0000 0000 0000 0010
preserve:        3 => 0000 0000 0000 0000 0000 0000 0000 0011

SORTED Flag:
maskTable: {
    SPLITERATOR:           0000 0001,
    STREAM:                0000 0001,
    OP:                    0000 0011,
    TERMINAL_OP:           0000 0000,
    UPSTREAM_TERMINAL_OP:  0000 0000
}
position(input): 1 
bitPosition:     2
set:             4 => 0000 0000 0000 0000 0000 0000 0000 0100
clear:           8 => 0000 0000 0000 0000 0000 0000 0000 1000
preserve:       12 => 0000 0000 0000 0000 0000 0000 0000 1100

ORDERED Flag:
maskTable: {
    SPLITERATOR:           0000 0001,
    STREAM:                0000 0001,
    OP:                    0000 0011,
    TERMINAL_OP:           0000 0010,
    UPSTREAM_TERMINAL_OP:  0000 0010
}
position(input): 2
bitPosition:     4 
set:            16 => 0000 0000 0000 0000 0000 0000 0001 0000
clear:          32 => 0000 0000 0000 0000 0000 0000 0010 0000
preserve:       48 => 0000 0000 0000 0000 0000 0000 0011 0000

SIZED Flag:
maskTable: {
    SPLITERATOR:           0000 0001,
    STREAM:                0000 0001,
    OP:                    0000 0010,
    TERMINAL_OP:           0000 0000,
    UPSTREAM_TERMINAL_OP:  0000 0000
}
position(input): 3
bitPosition:     6 
set:            64 => 0000 0000 0000 0000 0000 0000 0100 0000
clear:         128 => 0000 0000 0000 0000 0000 0000 1000 0000
preserve:      192 => 0000 0000 0000 0000 0000 0000 1100 0000

SHORT_CIRCUIT Flag:
maskTable: {
    SPLITERATOR:           0000 0000,
    STREAM:                0000 0000,
    OP:                    0000 0001,
    TERMINAL_OP:           0000 0001,
    UPSTREAM_TERMINAL_OP:  0000 0000
}
position(input): 12
bitPosition:     24 
set:       16777216 => 0000 0001 0000 0000 0000 0000 0000 0000
clear:     33554432 => 0000 0010 0000 0000 0000 0000 0000 0000   
preserve:  50331648 => 0000 0011 0000 0000 0000 0000 0000 0000

static int DISTINCT_SET = 0b0001;
static int SORTED_CLEAR = 0b1000;

public static void main(String[] args) throws Exception {
    // 支持DISTINCT标志和不支持SORTED标志
    int flags = DISTINCT_SET | SORTED_CLEAR;
    System.out.println(Integer.toBinaryString(flags));
    System.out.printf("支持DISTINCT标志:%s\n", DISTINCT_SET == (DISTINCT_SET & flags));
    System.out.printf("不支持SORTED标志:%s\n", SORTED_CLEAR == (SORTED_CLEAR & flags));
}

// 控制台输出
1001
支持DISTINCT标志:true
不支持SORTED标志:true

public static void main(String[] args) {
    int flags = StreamOpFlag.DISTINCT.set | StreamOpFlag.SORTED.clear;
    System.out.println(StreamOpFlag.DISTINCT.set == (StreamOpFlag.DISTINCT.set & flags));
    System.out.println(StreamOpFlag.SORTED.clear == (StreamOpFlag.SORTED.clear & flags));
}

// 输出

true
true

enum StreamOpFlag {

    // 暂时忽略其他代码

    // 返回当前StreamOpFlag的set/inject的bit map
    int set() {
        return set;
    }

    // 返回当前StreamOpFlag的清除的bit map
    int clear() {
        return clear;
    }

    // 这里判断当前StreamOpFlag类型->标记映射中Stream类型的标记，如果大于0说明不是初始化状态，那么当前StreamOpFlag就是Stream相关的标志
    boolean isStreamFlag() {
        return maskTable.get(Type.STREAM) > 0;
    }

    // 这里就用到按位与判断输入的flags中是否设置当前StreamOpFlag(StreamOpFlag.set)
    boolean isKnown(int flags) {
        return (flags & preserve) == set;
    }

    // 这里就用到按位与判断输入的flags中是否清除当前StreamOpFlag(StreamOpFlag.clear)
    boolean isCleared(int flags) {
        return (flags & preserve) == clear;
    }

    // 这里就用到按位与判断输入的flags中是否保留当前StreamOpFlag(StreamOpFlag.clear)
    boolean isPreserved(int flags) {
        return (flags & preserve) == preserve;
    }

    // 判断当前的Stream实体类型是否可以设置本标志，要求Stream实体类型的标志位为set或者preserve，按位与要大于0
    boolean canSet(Type t) {
        return (maskTable.get(t) & SET_BITS) > 0;
    }

    // 暂时忽略其他代码
}

(flags & set) == set => (flags & preserve) == set
(flags & clear) == clear => (flags & preserve) == clear
(flags & preserve) == preserve => (flags & preserve) == preserve

// 横向看(位移动运算符优先级高于与或，例如<<的优先级比|高)
SPLITERATOR_CHARACTERISTICS_MASK:
mask(init) = 0
mask(DISTINCT,SPLITERATOR[DISTINCT]=01,bitPosition=0) = 0000 0000 | 0000 0001 << 0 = 0000 0000 | 0000 0001 = 0000 0001
mask(SORTED,SPLITERATOR[SORTED]=01,bitPosition=2) = 0000 0001 | 0000 0001 << 2 = 0000 0001 | 0000 0100 = 0000 0101
mask(ORDERED,SPLITERATOR[ORDERED]=01,bitPosition=4) = 0000 0101 | 0000 0001 << 4 = 0000 0101 | 0001 0000 = 0001 0101
mask(SIZED,SPLITERATOR[SIZED]=01,bitPosition=6) = 0001 0101 | 0000 0001 << 6 = 0001 0101 | 0100 0000 = 0101 0101
mask(SHORT_CIRCUIT,SPLITERATOR[SHORT_CIRCUIT]=00,bitPosition=24) = 0101 0101 | 0000 0000 << 24 = 0101 0101 | 0000 0000 = 0101 0101
mask(final) = 0000 0000 0000 0000 0000 0000 0101 0101(二进制)、85(十进制)

STREAM_MASK:
mask(init) = 0
mask(DISTINCT,SPLITERATOR[DISTINCT]=01,bitPosition=0) = 0000 0000 | 0000 0001 << 0 = 0000 0000 | 0000 0001 = 0000 0001
mask(SORTED,SPLITERATOR[SORTED]=01,bitPosition=2) = 0000 0001 | 0000 0001 << 2 = 0000 0001 | 0000 0100 = 0000 0101
mask(ORDERED,SPLITERATOR[ORDERED]=01,bitPosition=4) = 0000 0101 | 0000 0001 << 4 = 0000 0101 | 0001 0000 = 0001 0101
mask(SIZED,SPLITERATOR[SIZED]=01,bitPosition=6) = 0001 0101 | 0000 0001 << 6 = 0001 0101 | 0100 0000 = 0101 0101
mask(SHORT_CIRCUIT,SPLITERATOR[SHORT_CIRCUIT]=00,bitPosition=24) = 0101 0101 | 0000 0000 << 24 = 0101 0101 | 0000 0000 = 0101 0101
mask(final) = 0000 0000 0000 0000 0000 0000 0101 0101(二进制)、85(十进制)

OP_MASK:
mask(init) = 0
mask(DISTINCT,SPLITERATOR[DISTINCT]=11,bitPosition=0) = 0000 0000 | 0000 0011 << 0 = 0000 0000 | 0000 0011 = 0000 0011
mask(SORTED,SPLITERATOR[SORTED]=11,bitPosition=2) = 0000 0011 | 0000 0011 << 2 = 0000 0011 | 0000 1100 = 0000 1111
mask(ORDERED,SPLITERATOR[ORDERED]=11,bitPosition=4) = 0000 1111 | 0000 0011 << 4 = 0000 1111 | 0011 0000 = 0011 1111
mask(SIZED,SPLITERATOR[SIZED]=10,bitPosition=6) = 0011 1111 | 0000 0010 << 6 = 0011 1111 | 1000 0000 = 1011 1111
mask(SHORT_CIRCUIT,SPLITERATOR[SHORT_CIRCUIT]=01,bitPosition=24) = 1011 1111 | 0000 0001 << 24 = 1011 1111 | 0100 0000 0000 0000 0000 0000 0000 = 0100 0000 0000 0000 0000 1011 1111
mask(final) = 0000 0000 1000 0000 0000 0000 1011 1111(二进制)、16777407(十进制)

TERMINAL_OP_MASK:
mask(init) = 0
mask(DISTINCT,SPLITERATOR[DISTINCT]=00,bitPosition=0) = 0000 0000 | 0000 0000 << 0 = 0000 0000 | 0000 0000 = 0000 0000
mask(SORTED,SPLITERATOR[SORTED]=00,bitPosition=2) = 0000 0000 | 0000 0000 << 2 = 0000 0000 | 0000 0000 = 0000 0000
mask(ORDERED,SPLITERATOR[ORDERED]=10,bitPosition=4) = 0000 0000 | 0000 0010 << 4 = 0000 0000 | 0010 0000 = 0010 0000
mask(SIZED,SPLITERATOR[SIZED]=00,bitPosition=6) = 0010 0000 | 0000 0000 << 6 = 0010 0000 | 0000 0000 = 0010 0000
mask(SHORT_CIRCUIT,SPLITERATOR[SHORT_CIRCUIT]=01,bitPosition=24) = 0010 0000 | 0000 0001 << 24 = 0010 0000 | 0001 0000 0000 0000 0000 0000 0000 = 0001 0000 0000 0000 0000 0010 0000
mask(final) = 0000 0001 0000 0000 0000 0000 0010 0000(二进制)、16777248(十进制)

UPSTREAM_TERMINAL_OP_MASK:
mask(init) = 0
mask(DISTINCT,SPLITERATOR[DISTINCT]=00,bitPosition=0) = 0000 0000 | 0000 0000 << 0 = 0000 0000 | 0000 0000 = 0000 0000
mask(SORTED,SPLITERATOR[SORTED]=00,bitPosition=2) = 0000 0000 | 0000 0000 << 2 = 0000 0000 | 0000 0000 = 0000 0000
mask(ORDERED,SPLITERATOR[ORDERED]=10,bitPosition=4) = 0000 0000 | 0000 0010 << 4 = 0000 0000 | 0010 0000 = 0010 0000
mask(SIZED,SPLITERATOR[SIZED]=00,bitPosition=6) = 0010 0000 | 0000 0000 << 6 = 0010 0000 | 0000 0000 = 0010 0000
mask(SHORT_CIRCUIT,SPLITERATOR[SHORT_CIRCUIT]=00,bitPosition=24) = 0010 0000 | 0000 0000 << 24 = 0010 0000 | 0000 0000 = 0010 0000
mask(final) = 0000 0000 0000 0000 0000 0000 0010 0000(二进制)、32(十进制)

enum StreamOpFlag {

    // SPLITERATOR类型的标志bit map
    static final int SPLITERATOR_CHARACTERISTICS_MASK = createMask(Type.SPLITERATOR);

    // STREAM类型的标志bit map
    static final int STREAM_MASK = createMask(Type.STREAM);

    // OP类型的标志bit map
    static final int OP_MASK = createMask(Type.OP);

    // TERMINAL_OP类型的标志bit map
    static final int TERMINAL_OP_MASK = createMask(Type.TERMINAL_OP);

    // UPSTREAM_TERMINAL_OP类型的标志bit map
    static final int UPSTREAM_TERMINAL_OP_MASK = createMask(Type.UPSTREAM_TERMINAL_OP);

    // 基于Stream类型，创建对应类型填充所有标志的bit map
    private static int createMask(Type t) {
        int mask = 0;
        for (StreamOpFlag flag : StreamOpFlag.values()) {
            mask |= flag.maskTable.get(t) << flag.bitPosition;
        }
        return mask;
    }

    // 构造一个标志本身的掩码，就是所有标志都采用保留位表示，目前作为flags == 0时候的初始值
    private static final int FLAG_MASK = createFlagMask();
    
    // 构造一个包含全部标志中的preserve位的bit map，按照目前来看是暂时是一个固定值，二进制表示为0011 0000 0000 0000 0000 1111 1111
    private static int createFlagMask() {
        int mask = 0;
        for (StreamOpFlag flag : StreamOpFlag.values()) {
            mask |= flag.preserve;
        }
        return mask;
    }

    // 构造一个Stream类型包含全部标志中的set位的bit map，这里直接使用了STREAM_MASK，按照目前来看是暂时是一个固定值，二进制表示为0000 0000 0000 0000 0000 0000 0101 0101
    private static final int FLAG_MASK_IS = STREAM_MASK;

    // 构造一个Stream类型包含全部标志中的clear位的bit map，按照目前来看是暂时是一个固定值，二进制表示为0000 0000 0000 0000 0000 0000 1010 1010
    private static final int FLAG_MASK_NOT = STREAM_MASK << 1;

    // 初始化操作的标志bit map，目前来看就是Stream的头节点初始化时候需要合并在flags里面的初始化值，照目前来看是暂时是一个固定值，二进制表示为0000 0000 0000 0000 0000 0000 1111 1111
    static final int INITIAL_OPS_VALUE = FLAG_MASK_IS | FLAG_MASK_NOT;
}

enum StreamOpFlag {
    
    // 这个方法完全没有注释，只使用在下面的combineOpFlags()方法中
    // 从源码来看
    // 入参flags == 0的时候，那么直接返回0011 0000 0000 0000 0000 1111 1111
    // 入参flags != 0的时候，那么会把当前flags的所有set/inject、clear和preserve所在位在bit map中全部置为0，然后其他位全部置为1
    private static int getMask(int flags) {
        return (flags == 0)
               ? FLAG_MASK
               : ~(flags | ((FLAG_MASK_IS & flags) << 1) | ((FLAG_MASK_NOT & flags) >> 1));
    }
    
    // 合并新的flags和前一个flags，这里还是用到老套路先和Mask按位与，再进行一次按位或
    // 作为Stream的头节点的时候，prevCombOpFlags必须为INITIAL_OPS_VALUE
    static int combineOpFlags(int newStreamOrOpFlags, int prevCombOpFlags) {
        // 0x01 or 0x10 nibbles are transformed to 0x11
        // 0x00 nibbles remain unchanged
        // Then all the bits are flipped
        // Then the result is logically or'ed with the operation flags.
        return (prevCombOpFlags & StreamOpFlag.getMask(newStreamOrOpFlags)) | newStreamOrOpFlags;
    }
    
    // 通过合并后的flags，转换出Stream类型的flags
    static int toStreamFlags(int combOpFlags) {
        // By flipping the nibbles 0x11 become 0x00 and 0x01 become 0x10
        // Shift left 1 to restore set flags and mask off anything other than the set flags
        return ((~combOpFlags) >> 1) & FLAG_MASK_IS & combOpFlags;
    }
    
    // Stream的标志转换为Spliterator的characteristics
    static int toCharacteristics(int streamFlags) {
        return streamFlags & SPLITERATOR_CHARACTERISTICS_MASK;
    }
    
    // Spliterator的characteristics转换为Stream的标志，入参是Spliterator实例
    static int fromCharacteristics(Spliterator<?> spliterator) {
        int characteristics = spliterator.characteristics();
        if ((characteristics & Spliterator.SORTED) != 0 && spliterator.getComparator() != null) {
            // Do not propagate the SORTED characteristic if it does not correspond
            // to a natural sort order
            return characteristics & SPLITERATOR_CHARACTERISTICS_MASK & ~Spliterator.SORTED;
        }
        else {
            return characteristics & SPLITERATOR_CHARACTERISTICS_MASK;
        }
    }

    // Spliterator的characteristics转换为Stream的标志，入参是Spliterator的characteristics
    static int fromCharacteristics(int characteristics) {
        return characteristics & SPLITERATOR_CHARACTERISTICS_MASK;
    }
}

int left = ORDERED.set | DISTINCT.set;
int right = SIZED.clear | SORTED.clear;
System.out.println("left:" + Integer.toBinaryString(left));
System.out.println("right:" + Integer.toBinaryString(right));
System.out.println("right mask:" + Integer.toBinaryString(getMask(right)));
System.out.println("combine:" + Integer.toBinaryString(combineOpFlags(right, left)));

// 输出结果
left:1010001
right:10001000
right mask:11111111111111111111111100110011
combine:10011001

// 这里简单点只展示8 bit
SPLITERATOR_CHARACTERISTICS_MASK: 0101 0101
Spliterator.ORDERED:              0001 0000
StreamOpFlag.ORDERED.set:         0001 0000

public interface Stream<T> extends BaseStream<T, Stream<T>> {
    
    // 忽略其他代码

    // 过滤Op
    Stream<T> filter(Predicate<? super T> predicate);

    // 映射Op
    <R> Stream<R> map(Function<? super T, ? extends R> mapper);
    
    // 终结操作 - 遍历
    void forEach(Consumer<? super T> action);

    // 忽略其他代码
}

// init
Stream x = buildStream();
// chain: head -> filter(Op) -> map(Op) -> forEach(Terminal Op)
x.filter().map().forEach()

// T是元素类型，S是BaseStream<T, S>类型
// 流的基础接口，这里的流指定的支持同步执行和异步执行的聚合操作的元素序列
public interface BaseStream<T, S extends BaseStream<T, S>> extends AutoCloseable {
    
    // 返回一个当前Stream实例中所有元素的迭代器
    // 这是一个终结操作
    Iterator<T> iterator();
    
    // 返回一个当前Stream实例中所有元素的可拆分迭代器
    Spliterator<T> spliterator();
    
    // 当前的Stream实例是否支持并发
    boolean isParallel();
    
    // 返回一个等效的同步处理的Stream实例
    S sequential();
    
    // 返回一个等效的并发处理的Stream实例
    S parallel();
    
    // 返回一个等效的不支持StreamOpFlag.ORDERED特性的Stream实例
    // 或者说支持StreamOpFlag.NOT_ORDERED的特性，也就返回的变体Stream在处理元素的时候不需要顺序处理
    S unordered();
    
    // 返回一个添加了close处理器的Stream实例，close处理器会在下面的close方法中回调
    S onClose(Runnable closeHandler);
    
    // 关闭当前Stream实例，回调关联本Stream的所有close处理器
    @Override
    void close();
}

abstract class PipelineHelper<P_OUT> {

    // 获取流的流水线的数据源的"形状"，其实就是数据源元素的类型
    // 主要有四种类型：REFERENCE（除了int、long和double之外的引用类型）、INT_VALUE、LONG_VALUE和DOUBLE_VALUE
    abstract StreamShape getSourceShape();

    // 获取合并流和流操作的标志，合并的标志包括流的数据源标志、中间操作标志和终结操作标志
    // 从实现上看是当前流管道节点合并前面所有节点和自身节点标志的所有标志
    abstract int getStreamAndOpFlags();

    // 如果当前的流管道节点的合并标志集合支持SIZED，则调用Spliterator.getExactSizeIfKnown()返回数据源中的准确元素数量，否则返回-1
    abstract<P_IN> long exactOutputSizeIfKnown(Spliterator<P_IN> spliterator);

    // 相当于调用下面的方法组合：copyInto(wrapSink(sink), spliterator)
    abstract<P_IN, S extends Sink<P_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator);

    // 发送所有来自Spliterator中的元素到Sink中，如果支持SHORT_CIRCUIT标志，则会调用copyIntoWithCancel
    abstract<P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator);

    // 发送所有来自Spliterator中的元素到Sink中，Sink处理完每个元素后会检查Sink#cancellationRequested()方法的状态去判断是否中断推送元素的操作
    abstract <P_IN> boolean copyIntoWithCancel(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator);

    // 创建接收元素类型为P_IN的Sink实例，实现PipelineHelper中描述的所有中间操作，用这个Sink去包装传入的Sink实例（传入的Sink实例的元素类型为PipelineHelper的输出类型P_OUT）
    abstract<P_IN> Sink<P_IN> wrapSink(Sink<P_OUT> sink);

    // 包装传入的spliterator，从源码来看，在Stream链的头节点调用会直接返回传入的实例，如果在非头节点调用会委托到StreamSpliterators.WrappingSpliterator()方法进行包装
    // 这个方法在源码中没有API注释
    abstract<P_IN> Spliterator<P_OUT> wrapSpliterator(Spliterator<P_IN> spliterator);

    // 构造一个兼容当前Stream元素"形状"的Node.Builder实例
    // 从源码来看直接委托到Nodes.builder()方法
    abstract Node.Builder<P_OUT> makeNodeBuilder(long exactSizeIfKnown,
                                                 IntFunction<P_OUT[]> generator);

    // Stream流水线所有阶段(节点)应用于数据源Spliterator，输出的元素作为结果收集起来转化为Node实例
    // 此方法应用于toArray()方法的计算，本质上是一个终结操作
    abstract<P_IN> Node<P_OUT> evaluate(Spliterator<P_IN> spliterator,
                                        boolean flatten,
                                        IntFunction<P_OUT[]> generator);
}

public class WrapperApp {

    interface Wrapper {

        void doAction();
    }

    public static void main(String[] args) {
        AtomicInteger counter = new AtomicInteger(0);
        Wrapper first = () -> System.out.printf("wrapper [depth => %d] invoke\n", counter.incrementAndGet());
        Wrapper second = () -> {
            first.doAction();
            System.out.printf("wrapper [depth => %d] invoke\n", counter.incrementAndGet());
        };
        second.doAction();
    }
}

// 控制台输出
wrapper [depth => 1] invoke
wrapper [depth => 2] invoke

public interface Sink<T> extends Consumer<T> {

    default void begin(long size) {

    }

    default void end() {

    }

    abstract class ChainedReference<T, OUT> implements Sink<T> {

        protected final Sink<OUT> downstream;

        public ChainedReference(Sink<OUT> downstream) {
            this.downstream = downstream;
        }
    }
}

@SuppressWarnings({"unchecked", "rawtypes"})
public class ReferenceChain<OUT, R> {

    /**
     * sink chain
     */
    private final List<Supplier<Sink<?>>> sinkBuilders = new ArrayList<>();

    /**
     * current sink
     */
    private final AtomicReference<Sink> sinkReference = new AtomicReference<>();

    public ReferenceChain<OUT, R> filter(Predicate<OUT> predicate) {
        //filter
        sinkBuilders.add(() -> {
            Sink<OUT> prevSink = (Sink<OUT>) sinkReference.get();
            Sink.ChainedReference<OUT, OUT> currentSink = new Sink.ChainedReference<>(prevSink) {

                @Override
                public void accept(OUT out) {
                    if (predicate.test(out)) {
                        downstream.accept(out);
                    }
                }
            };
            sinkReference.set(currentSink);
            return currentSink;
        });
        return this;
    }

    public ReferenceChain<OUT, R> map(Function<OUT, R> function) {
        // map
        sinkBuilders.add(() -> {
            Sink<R> prevSink = (Sink<R>) sinkReference.get();
            Sink.ChainedReference<OUT, R> currentSink = new Sink.ChainedReference<>(prevSink) {

                @Override
                public void accept(OUT in) {
                    downstream.accept(function.apply(in));
                }
            };
            sinkReference.set(currentSink);
            return currentSink;
        });
        return this;
    }

    public void forEachPrint(Collection<OUT> collection) {
        forEachPrint(collection, false);
    }

    public void forEachPrint(Collection<OUT> collection, boolean reverse) {
        Spliterator<OUT> spliterator = collection.spliterator();
        // 这个是类似于terminal op
        Sink<OUT> sink = System.out::println;
        sinkReference.set(sink);
        Sink<OUT> stage = sink;
        // 反向包装 -> 正向遍历
        if (reverse) {
            for (int i = 0; i <= sinkBuilders.size() - 1; i++) {
                Supplier<Sink<?>> supplier = sinkBuilders.get(i);
                stage = (Sink<OUT>) supplier.get();
            }
        } else {
            // 正向包装 -> 反向遍历
            for (int i = sinkBuilders.size() - 1; i >= 0; i--) {
                Supplier<Sink<?>> supplier = sinkBuilders.get(i);
                stage = (Sink<OUT>) supplier.get();
            }
        }
        Sink<OUT> finalStage = stage;
        spliterator.forEachRemaining(finalStage);
    }

    public static void main(String[] args) {
        List<Integer> list = new ArrayList<>();
        list.add(1);
        list.add(2);
        list.add(3);
        list.add(12);
        ReferenceChain<Integer, Integer> chain = new ReferenceChain<>();
        // filter -> map -> for each
        chain.filter(item -> item > 10)
                .map(item -> item * 2)
                .forEachPrint(list);
    }
}

// 输出结果
24

// 目标顺序：filter -> map
Sink mapSink = new Sink(inputSink){

    private Function mapper;

    public void accept(E ele) {
        inputSink.accept(mapper.apply(ele))
    }

}
Sink filterSink = new Sink(mapSink){

    private Predicate predicate;

    public void accept(E ele) {
        if(predicate.test(ele)){
            mapSink.accept(ele);
        }
    }
}

// 继承自Consumer，主要是继承函数式接口方法void accept(T t)
interface Sink<T> extends Consumer<T> {
    
    // 重置当前Sink的状态（为了接收一个新的数据集），传入的size是推送到downstream的准确数据量，无法评估数据量则传入-1
    default void begin(long size) {}

    // 
    default void end() {}

    // 返回true的时候表示当前的Sink不会接收数据
    default boolean cancellationRequested() {
        return false;
    }

    // 特化方法，接受一个int类型的值
    default void accept(int value) {
        throw new IllegalStateException("called wrong accept method");
    }

    // 特化方法，接受一个long类型的值
    default void accept(long value) {
        throw new IllegalStateException("called wrong accept method");
    }

    // 特化方法，接受一个double类型的值
    default void accept(double value) {
        throw new IllegalStateException("called wrong accept method");
    }
    
    // 引用类型链，准确来说是Sink链
    abstract static class ChainedReference<T, E_OUT> implements Sink<T> {
        
        // 下一个Sink
        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();
        }
    }
    // 暂时忽略Int、Long、Double的特化类型场景
}

abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
        extends PipelineHelper<E_OUT> implements BaseStream<E_OUT, S> {
    
    // 流管道链式结构的头节点（只有当前的AbstractPipeline引用是头节点，此变量才会被赋值，非头节点为NULL）
    @SuppressWarnings("rawtypes")
    private final AbstractPipeline sourceStage;
    
    // 流管道链式结构的upstream，也就是上一个节点，如果是头节点此引用为NULL
    @SuppressWarnings("rawtypes")
    private final AbstractPipeline previousStage;
    
    // 合并数据源的标志和操作标志的掩码
    protected final int sourceOrOpFlags;
    
    // 流管道链式结构的下一个节点，如果是头节点此引用为NULL
    @SuppressWarnings("rawtypes")
    private AbstractPipeline nextStage;
    
    // 流的深度
    // 串行执行的流中，表示当前流管道实例中中间操作节点的个数（除去头节点和终结操作）
    // 并发执行的流中，表示当前流管道实例中中间操作节点和前一个有状态操作节点之间的节点个数
    private int depth;
     
    // 合并了所有数据源的标志、操作标志和当前的节点(AbstractPipeline)实例的标志，也就是当前的节点可以基于此属性得知所有支持的标志
    private int combinedFlags;
    
    // 数据源的Spliterator实例
    private Spliterator<?> sourceSpliterator;
    
    // 数据源的Spliterator实例封装的Supplier实例
    private Supplier<? extends Spliterator<?>> sourceSupplier;
    
    // 标记当前的流节点是否被连接或者消费掉，不能重复连接或者重复消费
    private boolean linkedOrConsumed;

    // 标记当前的流管道链式结构中是否存在有状态的操作节点，这个属性只会在头节点中有效
    private boolean sourceAnyStateful;
    
    // 数据源关闭动作，这个属性只会在头节点中有效，由sourceStage持有
    private Runnable sourceCloseAction;
    
    // 标记当前流是否并发执行
    private boolean parallel;

    // 流管道结构头节点的父构造方法，使用数据源的Spliterator实例封装的Supplier实例
    AbstractPipeline(Supplier<? extends Spliterator<?>> source,
                     int sourceFlags, boolean parallel) {
        // 头节点的前驱节点置为NULL
        this.previousStage = null;
        this.sourceSupplier = source;
        this.sourceStage = this;
        // 合并传入的源标志和流标志的掩码
        this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
        // The following is an optimization of:
        // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
        // 初始化合并标志集合为sourceOrOpFlags和所有流操作标志的初始化值
        this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
        // 深度设置为0
        this.depth = 0;
        this.parallel = parallel;
    }

    // 流管道结构头节点的父构造方法，使用数据源的Spliterator实例
    AbstractPipeline(Spliterator<?> source,
                     int sourceFlags, boolean parallel) {
        // 头节点的前驱节点置为NULL
        this.previousStage = null;
        this.sourceSpliterator = source;
        this.sourceStage = this;
        // 合并传入的源标志和流标志的掩码
        this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
        // The following is an optimization of:
        // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
        // 初始化合并标志集合为sourceOrOpFlags和所有流操作标志的初始化值
        this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
        this.depth = 0;
        this.parallel = parallel;
    }
    
    // 流管道结构中间操作节点的父构造方法
    AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
        if (previousStage.linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        previousStage.linkedOrConsumed = true;
        // 设置前驱节点的后继节点引用为当前的AbstractPipeline实例
        previousStage.nextStage = this;
        // 设置前驱节点引用为传入的前驱节点实例
        this.previousStage = previousStage;
        // 合并传入的中间操作标志和流操作标志的掩码
        this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
        // 合并标志集合为传入的标志和前驱节点的标志集合
        this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
        // 赋值sourceStage为前驱节点的sourceStage
        this.sourceStage = previousStage.sourceStage;
        if (opIsStateful())
            // 标记当前的流存在有状态操作
            sourceStage.sourceAnyStateful = true;
        // 深度设置为前驱节点深度加1
        this.depth = previousStage.depth + 1;
    }

    // 省略其他方法
}