java8 stream 源码分析

Java8 Lambda表达式，常用接口，Stream流式处理
java8 stream 源码分析
java8 lambda底层原理

源码分析

补充画图：stage、sink（2024-04-24 18:12:46）

list ArrayList 1293

ArrayListSpliterator 1302

ReferencePipeline$Head 1305 (stream   ReferencePipeline.Head)

ReferencePipeline$2 1309 (filter   ReferencePipeline.StatelessOp)

ReferencePipeline$2 1316 (map   ReferencePipeline.StatelessOp)

------------------------------------------------------------------------------------------------
stream   ------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------
# java.util.Collection
default Stream<E> stream() {
    return StreamSupport.stream(spliterator(), false);
}

        # java.util.ArrayList
        @Override
        public Spliterator<E> spliterator() {
            return new ArrayListSpliterator<>(this, 0, -1, 0);
        }

# java.util.stream.StreamSupport
public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
    // ReferencePipeline$Head 1305 (stream   ReferencePipeline.Head)
    return new ReferencePipeline.Head<>(spliterator,
                                        StreamOpFlag.fromCharacteristics(spliterator),
                                        parallel);
}

        # java.util.stream.ReferencePipeline.Head
        static class Head<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT> {

            Head(Spliterator<?> source, int sourceFlags, boolean parallel) {
                super(source, sourceFlags, parallel);// source   ArrayListSpliterator 1302
            }
        }

        # java.util.stream.ReferencePipeline
        abstract class ReferencePipeline<P_IN, P_OUT>
            extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
            implements Stream<P_OUT>  {

            ReferencePipeline(Spliterator<?> source, int sourceFlags, boolean parallel) {
                super(source, sourceFlags, parallel);// source   ArrayListSpliterator 1302
            }
        }

        # java.util.stream.AbstractPipeline
        abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
                extends PipelineHelper<E_OUT> 
                implements BaseStream<E_OUT, S> {

            AbstractPipeline(Spliterator<?> source, int sourceFlags, boolean parallel) {
                this.previousStage = null;
                this.sourceSpliterator = source;// ArrayListSpliterator 1302
                this.sourceStage = this;// ReferencePipeline$Head 1305
                this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
                this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
                this.depth = 0;
                this.parallel = parallel;
            }
        }

------------------------------------------------------------------------------------------------
filter   ------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------
predicate 1308

# java.util.stream.ReferencePipeline
@Override
public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
    // ReferencePipeline$2 1309 (filter   ReferencePipeline.StatelessOp)
    return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
            return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    if (predicate.test(u))
                        downstream.accept(u);
                }
            };
        }
    };
}

        # java.util.stream.ReferencePipeline.StatelessOp
        abstract static class StatelessOp<E_IN, E_OUT>
        extends ReferencePipeline<E_IN, E_OUT> {

            StatelessOp(AbstractPipeline<?, E_IN, ?> upstream, StreamShape inputShape, int opFlags) {
                super(upstream, opFlags);// upstream ===> ReferencePipeline$Head 1305 (stream   ReferencePipeline.Head)
            }
        }


        abstract class ReferencePipeline<P_IN, P_OUT>
        extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
        implements Stream<P_OUT>  {

            ReferencePipeline(AbstractPipeline<?, P_IN, ?> upstream, int opFlags) {
                super(upstream, opFlags);// upstream   ReferencePipeline$Head 1305 (stream   ReferencePipeline.Head)
            }
        }

        abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
                extends PipelineHelper<E_OUT> 
                implements BaseStream<E_OUT, S> {
            private static final String MSG_STREAM_LINKED = "stream has already been operated upon or closed";

            AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
                if (previousStage.linkedOrConsumed)// false 不被连接
                    throw new IllegalStateException(MSG_STREAM_LINKED);
                previousStage.linkedOrConsumed = true;// true 已被连接(上一个节点 不可再被连接)
                previousStage.nextStage = this;// prev.next = this

                this.previousStage = previousStage;// this.prev = prev
                this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
                this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
                this.sourceStage = previousStage.sourceStage;// ReferencePipeline$Head 1305（sourceStage等同于head节点）
                if (opIsStateful())// false
                    sourceStage.sourceAnyStateful = true;
                this.depth = previousStage.depth + 1;// 1   this.depth = prev.depth+1
            }
        }

------------------------------------------------------------------------------------------------
map   ------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------
mapper 1315

# java.util.stream.ReferencePipeline
@Override
public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
            return new Sink.ChainedReference<P_OUT, R>(sink) {
                @Override
                public void accept(P_OUT u) {
                    downstream.accept(mapper.apply(u));
                }
            };
        }
    };
}

        AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
            if (previousStage.linkedOrConsumed)
                throw new IllegalStateException(MSG_STREAM_LINKED);
            previousStage.linkedOrConsumed = true;
            previousStage.nextStage = this;// 1316 map

            this.previousStage = previousStage;// 1309 filter
            this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
            this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
            this.sourceStage = previousStage.sourceStage;// 1305 stream
            if (opIsStateful())
                sourceStage.sourceAnyStateful = true;
            this.depth = previousStage.depth + 1;// 2
        }


------------------------------------------------------------------------------------------------
sorted   ------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------
comparator 1318   Comparator.reverseOrder()


    # java.util.stream.ReferencePipeline#sorted
    @Override
    public final Stream<P_OUT> sorted(Comparator<? super P_OUT> comparator) {
        return SortedOps.makeRef(this, comparator);// this=1316 map
    }


    final class SortedOps {
        # java.util.stream.SortedOps#makeRef
        static <T> Stream<T> makeRef(AbstractPipeline<?, T, ?> upstream,
                                    Comparator<? super T> comparator) {
            return new OfRef<>(upstream, comparator);// SortedOps$OfRef 1342
        }

        private static final class OfRef<T> extends ReferencePipeline.StatefulOp<T, T> {
            OfRef(AbstractPipeline<?, T, ?> upstream, Comparator<? super T> comparator) {
                super(upstream, StreamShape.REFERENCE, StreamOpFlag.IS_ORDERED | StreamOpFlag.NOT_SORTED);
                this.isNaturalSort = false;
                this.comparator = Objects.requireNonNull(comparator);
            }
        }
    }




    # java.util.stream.ReferencePipeline.StatefulOp#StatefulOp
    StatefulOp(AbstractPipeline<?, E_IN, ?> upstream, StreamShape inputShape, int opFlags) {
        super(upstream, opFlags);// upstream=1316 map
    }

    AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
        if (previousStage.linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        previousStage.linkedOrConsumed = true;
        previousStage.nextStage = this;// SortedOps$OfRef 1342

        this.previousStage = previousStage;// 1316 map
        this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
        this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
        this.sourceStage = previousStage.sourceStage;// 1305 stream
        if (opIsStateful())
            sourceStage.sourceAnyStateful = true;
        this.depth = previousStage.depth + 1;// 3
    }


------------------------------------------------------------------------------------------------
collect   ------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------
CollectorImpl 1344   Collectors.toList()


        public final class Collectors {
            # java.util.stream.Collectors#toList
            public static <T> Collector<T, ?, List<T>> toList() {
                return new CollectorImpl<>(
                                           (Supplier<List<T>>) ArrayList::new,
                                           List::add,
                                           (left, right) -> { left.addAll(right); return left; },
                                           CH_ID
                );
            }

            static class CollectorImpl<T, A, R> implements Collector<T, A, R> {
                CollectorImpl(Supplier<A> supplier,
                              BiConsumer<A, T> accumulator,
                              BinaryOperator<A> combiner,
                              Set<Characteristics> characteristics) {
                    this(supplier, accumulator, combiner, castingIdentity(), characteristics);
                }

                CollectorImpl(Supplier<A> supplier,
                              BiConsumer<A, T> accumulator,
                              BinaryOperator<A> combiner,
                              Function<A,R> finisher,
                              Set<Characteristics> characteristics) {
                    this.supplier = supplier;// 1345   ArrayList::new
                    this.accumulator = accumulator;// 1346   List::add
                    this.combiner = combiner;// 1347   (left, right) -> { left.addAll(right); return left; }
                    this.finisher = finisher;// 1348   i -> (R) i   T转R
                    this.characteristics = characteristics;
                }
            }

            private static <I, R> Function<I, R> castingIdentity() {
                return i -> (R) i;
            }
        }



    public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) {
        A container;
        if (isParallel()
                && (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
                && (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
            container = collector.supplier().get();
            BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();
            forEach(u -> accumulator.accept(container, u));
        }
        else {
            // 非并行
            container = evaluate(ReduceOps.makeRef(collector));
        }
        return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
               ? (R) container
               : collector.finisher().apply(container);
    }

            final class ReduceOps {
                # java.util.stream.ReduceOps#makeRef
                public static <T, I> TerminalOp<T, I> makeRef(Collector<? super T, I, ?> collector) {
                        Supplier<I> supplier = collector.supplier();// 1345   ArrayList::new
                        BiConsumer<I, ? super T> accumulator = collector.accumulator();// 1346   List::add
                        BinaryOperator<I> combiner = collector.combiner();// 1347   (left, right) -> { left.addAll(right); return left; }

                        # ReduceOps$3ReducingSink Box 1355
                        class ReducingSink extends Box<I> implements AccumulatingSink<T, I, ReducingSink> {
                            @Override
                            public void begin(long size) { state = supplier.get(); }
                            @Override
                            public void accept(T t) { accumulator.accept(state, t); }
                            @Override
                            public void combine(ReducingSink other) { state = combiner.apply(state, other.state); }
                        }
                        // ReduceOps.ReduceOp 1354
                        return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
                            @Override
                            public ReducingSink makeSink() {
                                return new ReducingSink();
                            }

                            @Override
                            public int getOpFlags() {
                                return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
                                       ? StreamOpFlag.NOT_ORDERED
                                       : 0;
                            }
                        };
                }
            }


            private static abstract class ReduceOp<T, R, S extends AccumulatingSink<T, R, S>>
                    implements TerminalOp<T, R> {
                private final StreamShape inputShape;

                ReduceOp(StreamShape shape) {
                    inputShape = shape;// REFERENCE
                }

                public abstract S makeSink();

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

                @Override
                public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
                                                   Spliterator<P_IN> spliterator) {
                    return helper.wrapAndCopyInto(makeSink(), spliterator).get();
                }

                @Override
                public <P_IN> R evaluateParallel(PipelineHelper<T> helper,
                                                 Spliterator<P_IN> spliterator) {
                    return new ReduceTask<>(this, helper, spliterator).invoke().get();
                }
            }

            # java.util.stream.AbstractPipeline#evaluate
            # this=SortedOps$OfRef 1342
            final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
                if (linkedOrConsumed)
                    throw new IllegalStateException(MSG_STREAM_LINKED);
                linkedOrConsumed = true;// true   ===> 当前设为 true（之前都是 prev.link 设为 true）

                return isParallel()
                       ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))

                       // 非并行
                       : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
            }


            // terminalOp.evaluateSequential ===> ReduceOps.ReduceOp#evaluateSequential

        public <P_IN> R evaluateSequential(PipelineHelper<T> helper,// SortedOps$OfRef 1342
                                           Spliterator<P_IN> spliterator) {// ArrayListSpliterator 1302
            // makeSink()
            //      ===> ReduceOps#makeRef
            //              ===> ReduceOps$3ReducingSink Box 1355 collect_sink
            return helper.wrapAndCopyInto(makeSink(), spliterator).get();
            // get java.util.stream.ReduceOps.Box#get    获取 collect ArrayList::new 创建的对象
        }

        # java.util.stream.AbstractPipeline#wrapAndCopyInto
        // this=SortedOps$OfRef 1342
        final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
            copyInto(wrapSink(sink), spliterator);
            return sink;// sink ===> ReduceOps$3ReducingSink Box 1355 collect_sink
        }

        final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
            // AbstractPipeline.this ===> SortedOps$OfRef 1342
            // 1.
            //      sink ===> ReduceOps$3ReducingSink Box 1355 collect_sink
            //      p ===> SortedOps$OfRef 1342
            //      sink ===> SortedOps$RefSortingSink 1366 sort_sink    downstream = collect_sink
            // 2.
            //      p ===> 1316 map
            //      sink ===> ReferencePipeline$3$1 1367 map_sink       downstream = sort_sink
            // 3.
            //      p ===> 1309 filter
            //      sink ===> ReferencePipeline$2$1 1379 filter_sink        downstream = map_sink
            // 
            // 
            // 
            // 
            // list5 ===> ArrayList::new 1424
            // 代码       List<Integer> list5 = list.stream()
            //                                      .filter(e -> e == 11)
            //                                      .map(e -> e * 3)
            //                                      .sorted(Comparator.reverseOrder())
            //                                      .collect(Collectors.toList());
            //
            //
            //
            // 总结
            //      filter_sink
            //          downstream = map_sink
            //              downstream = sort_sink
            //                  downstream = collect_sink
            for (AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
                sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
            }
            return (Sink<P_IN>) sink;
        }

        # 1.opWrapSink
        # java.util.stream.SortedOps.OfRef#opWrapSink
        public Sink<T> opWrapSink(int flags, Sink<T> sink) {
            if (StreamOpFlag.SORTED.isKnown(flags) && isNaturalSort)
                return sink;
            else if (StreamOpFlag.SIZED.isKnown(flags))
                return new SizedRefSortingSink<>(sink, comparator);
            else
                return new RefSortingSink<>(sink, comparator);// REFERENCE stream 排序
        }


        private static final class RefSortingSink<T> extends AbstractRefSortingSink<T> {
            private ArrayList<T> list;

            RefSortingSink(Sink<? super T> sink, Comparator<? super T> comparator) {
                // this = SortedOps$RefSortingSink 1366
                // sink ===> ReduceOps$3ReducingSink Box 1355        downstream
                // comparator 1318   Comparator.reverseOrder()
                super(sink, comparator);
            }

            @Override
            public void begin(long size) {
                // list = ArrayList 1406  （copyInto begin 时触发）
                list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();
            }

            @Override
            public void end() {
                list.sort(comparator);
                downstream.begin(list.size());// downstream = collect_sink
                if (!cancellationWasRequested) {
                    list.forEach(downstream::accept);
                }
                else {
                    for (T t : list) {
                        if (downstream.cancellationRequested()) break;
                        downstream.accept(t);
                    }
                }
                downstream.end();
                list = null;
            }

            @Override
            public void accept(T t) {
                list.add(t);
            }
        }

        # 2.opWrapSink
        # java.util.stream.ReferencePipeline#map
        Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
            return new Sink.ChainedReference<P_OUT, R>(sink) {
                @Override
                public void accept(P_OUT u) {
                    downstream.accept(mapper.apply(u));// downstream = sort_sink
                }
            };
        }

        static abstract class ChainedReference<T, E_OUT> implements Sink<T> {
            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(); }
        }

        # 3.opWrapSink
        # java.util.stream.ReferencePipeline#filter
        Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
            return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    if (predicate.test(u))
                        downstream.accept(u);// downstream = map_sink
                }
            };
        }

        @Override
        // wrappedSink = 1379 filter_sink
        final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
            if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
                wrappedSink.begin(spliterator.getExactSizeIfKnown());
                spliterator.forEachRemaining(wrappedSink);// 循环消费
                wrappedSink.end();
            }
            else {
                copyIntoWithCancel(wrappedSink, spliterator);
            }
        }

        // java.util.ArrayList.ArrayListSpliterator#forEachRemaining
        public void forEachRemaining(Consumer<? super E> action) {
            for (int i = 0; i < list.size; i++) {
                E e = (E) list[i];
                action.accept(e);// action = 1379 filter_sink
            }
        }
                    public void accept(P_OUT u) {// filter_sink
                        if (predicate.test(u))
                            downstream.accept(u);
                    }
                    public void accept(P_OUT u) {// map_sink
                        downstream.accept(mapper.apply(u));
                    }
                    public void accept(T t) {// sort_sink
                        list.add(t);// list = ArrayList 1406
                    }

        
        public void end() {// filter_sink
            downstream.end();
        }
        public void end() {// map_sink
            downstream.end();
        }
        public void end() {// sort_sink
            list.sort(comparator);
            downstream.begin(list.size());// downstream = collect_sink
            if (!cancellationWasRequested) {
                list.forEach(downstream::accept);
            }
            else {
                for (T t : list) {
                    if (downstream.cancellationRequested()) break;
                    downstream.accept(t);
                }
            }
            downstream.end();
            list = null;
        }


            public void begin(long size) {
                state = supplier.get();// state 1424
            }
            public void accept(T t) {
                accumulator.accept(state, t);
            }
            default void end() {}

手写

MyArrayList

package com.taopanfeng.junit.mystream;

import java.util.ArrayList;

/**
 * 描述
 *
 * @author 陶攀峰
 * @date 2024-04-03 15:18
 */
public class MyArrayList extends ArrayList<Integer> {

    public MyArrayList(Integer... arr) {
        for (Integer i : arr) {
            this.add(i);
        }
    }

    public MyStream myStream() {
        return new MyStream(this);
    }

}

MyConsumer

package com.taopanfeng.junit.mystream;

import java.util.function.Consumer;

/**
 * 描述
 *
 * @author 陶攀峰
 * @date 2024-04-03 15:46
 */
public abstract class MyConsumer implements Consumer<Integer> {
    MyArrayList container;

    public void begin(long size) {
    }

    public abstract void accept(Integer t);

    public void end() {
    }

    public static abstract class LinkedConsumer extends MyConsumer {
        MyConsumer next;

        public LinkedConsumer(MyConsumer next) {
            this.next = next;
        }

        @Override
        public void begin(long size) {
            next.begin(size);
        }

        @Override
        public void end() {
            next.end();
        }
    }
}

MyStream

package com.taopanfeng.junit.mystream;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Console;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * 描述
 *
 * @author 陶攀峰
 * @date 2024-04-03 15:13
 */
public class MyStream {

    private boolean linked;
    private int depth;
    private MyArrayList list;

    private MyStream prevStage;// prev
    private MyStream nextStage;// next

    public MyStream(MyArrayList list) {
        this.linked = false;
        this.depth = 0;
        this.list = list;

        this.prevStage = null;
        this.nextStage = null;
    }

    public MyStream(MyStream prevStage) {
        if (prevStage.linked)
            throw new IllegalStateException("流已被操作或关闭");
        prevStage.linked = true;
        prevStage.nextStage = this;

        this.depth = prevStage.depth + 1;
        this.list = prevStage.list;

        this.prevStage = prevStage;
        this.nextStage = null;
    }

    public final MyStream __peek(Consumer<Integer> consumer) {
        return new MyStream(this) {
            @Override
            public MyConsumer wrapConsumer(MyConsumer next) {
                return new MyConsumer.LinkedConsumer(next) {
                    @Override
                    public void accept(Integer i) {
                        consumer.accept(i);
                        next.accept(i);
                    }
                };
            }
        };
    }

    public final MyStream __map(Function<Integer, Integer> mapper) {
        return new MyStream(this) {
            @Override
            public MyConsumer wrapConsumer(MyConsumer next) {
                return new MyConsumer.LinkedConsumer(next) {
                    @Override
                    public void accept(Integer i) {
                        next.accept(mapper.apply(i));
                    }
                };
            }
        };
    }

    public final MyStream __flatMap(Function<Integer, MyStream> mapper) {
        return new MyStream(this) {
            @Override
            public MyConsumer wrapConsumer(MyConsumer next) {
                return new MyConsumer.LinkedConsumer(next) {
                    @Override
                    public void accept(Integer i) {
                        mapper.apply(i).list.forEach(next);
                    }
                };
            }
        };
    }

    public final MyStream __filter(Predicate<Integer> predicate) {
        return new MyStream(this) {
            @Override
            public MyConsumer wrapConsumer(MyConsumer next) {
                return new MyConsumer.LinkedConsumer(next) {
                    @Override
                    public void accept(Integer i) {
                        if (predicate.test(i)) {
                            next.accept(i);
                        }
                    }
                };
            }
        };
    }

    public final MyStream __distinct() {
        return new MyStream(this) {
            @Override
            public MyConsumer wrapConsumer(MyConsumer next) {
                return new MyConsumer.LinkedConsumer(next) {

                    HashSet<Integer> set;

                    @Override
                    public void begin(long size) {
                        set = new HashSet<>();
                        next.begin(size);
                    }

                    @Override
                    public void accept(Integer i) {
                        if (set.add(i)) {
                            next.accept(i);
                        }
                    }

                    @Override
                    public void end() {
                        set = null;
                        next.end();
                    }
                };
            }
        };
    }

    public final MyStream __sorted(Comparator<Integer> comparator) {
        return new MyStream(this) {
            @Override
            public MyConsumer wrapConsumer(MyConsumer next) {
                return new MyConsumer.LinkedConsumer(next) {

                    ArrayList<Integer> array;

                    @Override
                    public void begin(long size) {
                        array = new MyArrayList();
                    }

                    @Override
                    public void accept(Integer i) {
                        array.add(i);
                    }

                    @Override
                    public void end() {
                        array.sort(comparator);

                        next.begin(array.size());
                        array.forEach(next);
                        next.end();

                        array = null;
                    }
                };
            }
        };
    }

    public final MyStream __skip(Integer skip) {
        return slice(skip, Integer.MAX_VALUE);
    }

    public final MyStream __limit(Integer limit) {
        return slice(0, limit);
    }

    private MyStream slice(Integer skip, Integer limit) {
        return new MyStream(this) {
            @Override
            public MyConsumer wrapConsumer(MyConsumer next) {
                return new MyConsumer.LinkedConsumer(next) {

                    int _skip;
                    int _limit;

                    @Override
                    public void begin(long size) {
                        this._skip = skip;
                        this._limit = limit;
                        next.begin(size);
                    }

                    @Override
                    public void accept(Integer i) {
                        if (_skip == 0) {
                            if (_limit > 0) {
                                _limit--;
                                next.accept(i);
                            }
                        } else {
                            _skip--;
                        }
                    }

                    @Override
                    public void end() {
                        next.end();
                    }
                };
            }
        };
    }


    public final MyArrayList __collect(Supplier<MyArrayList> supplier, BiConsumer<MyArrayList, Integer> accumulator) {
        MyConsumer collect_consumer = new MyConsumer() {
            @Override
            public void begin(long size) {
                container = supplier.get();
            }

            @Override
            public void accept(Integer e) {
                Objects.requireNonNull(container, "容器未初始化");
                accumulator.accept(container, e);
            }
        };

        // 构建消费链表 => 返回 头节点
        MyConsumer head = linkedConsumer(collect_consumer);

        // 依次消费 => head、 head.next、 head.next.next、...
        foreach(head);

        // 返回容器
        return collect_consumer.container;
    }

    final MyConsumer linkedConsumer(MyConsumer consumer) {
        for (MyStream p = this; p.depth > 0; p = p.prevStage) {
            consumer = p.wrapConsumer(consumer);
        }
        return consumer;
    }

    final void foreach(MyConsumer head) {
        head.begin(list.size());
        list.forEach(head);
        head.end();
    }

    MyConsumer wrapConsumer(MyConsumer sink) {
        throw new RuntimeException();
    }

    // 2024-04-03 15:13
    public static void main(String[] args) throws Exception {
        ArrayList<Integer> list = CollUtil.newArrayList(11, 22, 33);
        // 11 22 33

        List<Integer> list2 = list.stream().map(e -> e * 2).collect(Collectors.toList());
        // 22 44 66

        {
            MyArrayList myArrayList = new MyArrayList(11, 22, 33);
            MyArrayList myArrayList2 = myArrayList.myStream().__map(e -> e * 2).__collect(MyArrayList::new, ArrayList::add);
            System.out.println();
        }


        {
            MyArrayList myArrayList = new MyArrayList(11, 22, 33, 22);

            MyArrayList myArrayList2 = myArrayList
                    .myStream()
                    .__filter(e -> e > 11)
                    .__map(e -> e * 2)
                    .__distinct()
                    .__sorted(Comparator.reverseOrder())
                    .__peek(e -> Console.log("排序后：{}", e))
                    .__flatMap(e -> new MyArrayList(e, e * 10).myStream())
                    .__skip(1)
                    .__limit(2)
                    .__collect(MyArrayList::new, ArrayList::add)
                    // .collect(MyArrayList::new, (container, e) -> container.add(e))
                    ;
            System.out.println();
        }
    }

}

并行处理

2024-04-24 18:13:12 补充

前置知识：parallel、spliterator、ForkJoinTask

图片-java8 stream 并行处理

import cn.hutool.core.collection.CollUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Spliterator;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class MySpliterator {
    static Logger logger = LoggerFactory.getLogger(MySpliterator.class.getName());

    // 2024-04-24 16:09
    public static void main(String[] args) throws Throwable {
        // 1、获取 Stream
        // List<Integer> list = CollUtil.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        // Integer[] array = {1, 2, 3, 4, 5, 6, 7, 8, 9};
        // Stream<Integer> s1 = list.stream();
        // Stream<Integer> s2 = Arrays.stream(array);

        // 2、获取 Spliterator
        //      tryAdvance 获取一个
        //      forEachRemaining 获取剩余所有的
        // List<Integer> list = CollUtil.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        // Spliterator<Integer> sp1 = list.spliterator();
        // sp1.tryAdvance(System.out::println);// 1
        // sp1.tryAdvance(System.out::println);// 2
        // sp1.tryAdvance(System.out::println);// 3
        // System.out.println("======================");
        // sp1.forEachRemaining(System.out::println);// 依次输出剩余元素   4, 5, 6, 7, 8, 9

        // 3、trySplit 分割 ===> 用于并行流
        // List<Integer> list = CollUtil.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        // Spliterator<Integer> sp1 = list.spliterator();
        // Spliterator<Integer> sp2 = sp1.trySplit();
        // Spliterator<Integer> sp3 = sp2.trySplit();
        // System.out.println("sp1:");
        // sp1.forEachRemaining(System.out::println);// [4, 9)  依次输出：5 6 7 8 9
        // System.out.println("sp2:");
        // sp2.forEachRemaining(System.out::println);// [2, 4)  依次输出：3 4
        // System.out.println("sp3:");
        // sp3.forEachRemaining(System.out::println);// [0, 2)  依次输出：1 2

        // 4、多线程测试
        // List<Integer> list = CollUtil.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        // Spliterator<Integer> sp1 = list.spliterator();
        // Spliterator<Integer> sp2 = sp1.trySplit();
        // Spliterator<Integer> sp3 = sp2.trySplit();
        // CompletableFuture.supplyAsync(() -> StreamSupport.stream(sp1, false).reduce(0, Integer::sum))
        //         .thenAccept(x -> logger.info("sp1 累加：{}", x));// sp1 累加：35
        // CompletableFuture.supplyAsync(() -> StreamSupport.stream(sp2, false).reduce(0, Integer::sum))
        //         .thenAccept(x -> logger.info("sp2 累加：{}", x));// sp2 累加：7
        // CompletableFuture.supplyAsync(() -> StreamSupport.stream(sp3, false).reduce(0, Integer::sum))
        //         .thenAccept(x -> logger.info("sp3 累加：{}", x));// sp3 累加：3

        // 5、LinkedList 测试
        // LinkedList<Integer> linkedList = CollUtil.newLinkedList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        // Spliterator<Integer> sp1 = linkedList.spliterator();// LLSpliterator
        // Spliterator<Integer> sp2 = sp1.trySplit();// Object[] a = new Object[9];    ArraySpliterator
        // sp1.forEachRemaining(System.out::println);// 无输出
        // sp2.forEachRemaining(System.out::println);// 全部输出  依次输出：1, 2, 3, 4, 5, 6, 7, 8, 9

        // 6、HashSet 测试
        // HashSet<Integer> hashSet = CollUtil.newHashSet(1, 2, 3, 4, 5, 6, 7, 8, 9);
        // Spliterator<Integer> sp1 = hashSet.spliterator();// HashMap.KeySpliterator   [8, 16)
        // Spliterator<Integer> sp2 = sp1.trySplit();// KeySpliterator                  [0, 8)

        // 7、TreeSet 测试
        // TreeSet<Integer> treeSet = new TreeSet<>();
        // treeSet.addAll(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9));
        // Spliterator<Integer> sp1 = treeSet.spliterator();// TreeMap.KeySpliterator
        // Spliterator<Integer> sp2 = sp1.trySplit();// TreeMap.KeySpliterator
        // sp1.forEachRemaining(System.out::println);// [root, max) 右树,包含root   依次输出：4, 5, 6, 7, 8, 9
        // sp2.forEachRemaining(System.out::println);// [min, root) 左树,不含root   依次输出：1, 2, 3

        // 8、源码分析 -> https://img-blog.csdnimg.cn/direct/92a32dcf619e4a1ab7d7a752c09cb860.png
        List<Integer> list = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .parallel()
                .map(x -> x * 2)
                .peek(x -> logger.info("{}", x))
                .collect(Collectors.toList());

        System.in.read();
    }
}