集合 Collection.stream()
数组 Arrays.stream
静态方法 Stream.of

default Stream stream() : 返回一个串行流。
default Stream parallelStream() : 返回一个并行流。
List<Integer> integerList = new ArrayList<>();
integerList.add(1);
integerList.add(2);
Stream<Integer> stream = integerList.stream();
Stream<Integer> stream1 = integerList.parallelStream();

int[] array = {1,2,3};
Stream<Integer> stream = Arrays.stream(array);

Stream<Integer> integerStream = Stream.of(1, 2, 3, 4, 5, 6, 7, 8);

List<String> list = Arrays.asList("b","b","c","a");
list.forEach(System.out::print); //bbca
list.stream().distinct().forEach(System.out::print);//bca

//筛选1501班的学生
computerClub.stream().filter(e -> e.getClassNum().equals("1501")).forEach(System.out::println);
//筛选年龄大于15的学生
List<Student> collect = computerClub.stream().filter(e -> e.getAge() > 15).collect(Collectors.toList());

//篮球俱乐部所有成员名 + 暂时住上商标^_^,并且获取所有队员名
List<String> collect1 = basketballClub.stream()
        .map(e -> e.getName() + "^_^")
        .collect(Collectors.toList());
        collect1.forEach(System.out::println);

//篮球俱乐部所有成员名 + 赞助商商标^_^,并且获取所有队员详细内容
List<Student> collect = basketballClub.stream()
.peek(e -> e.setName(e.getName() + "^_^"))
.collect(Collectors.toList());
collect.forEach(System.out::println);
//Student{idNum='2015134012', name='小c^_^', age=13, classNum='1503'}
//Student{idNum='2015134013', name='小s^_^', age=14, classNum='1503'}
//Student{idNum='2015134015', name='小d^_^', age=15, classNum='1504'}
//Student{idNum='2015134018', name='小y^_^', age=16, classNum='1505'}

List<String> list = Arrays.asList("a","b","c");
//获取list中top2即截断取前两个
List<String> collect1 = list.stream().limit(2).collect(Collectors.toList());
collect1.forEach(System.out::print);//ab

List<String> list = Arrays.asList("a","b","c");
//获取list中top2即截断取前两个
List<String> collect1 = list.stream().skip(2).collect(Collectors.toList());
collect1.forEach(System.out::print);//c

List<String> list = Arrays.asList("b","c","a");
//获取list中top2即截断取前两个
List<String> collect1 = list.stream().sorted().collect(Collectors.toList());
collect1.forEach(System.out::print);//abc

//此处只是演示 此类需求直接用List构造器即可
List<Student> collect = computerClub.stream().collect(Collectors.toList());
Set<Student> collect1 = pingpongClub.stream().collect(Collectors.toSet());

//注意key必须是唯一的 如果不是唯一的会报错而不是像普通map那样覆盖
        Map<String, String> collect2 = pingpongClub.stream()
        .collect(Collectors.toMap(Student::getIdNum, Student::getName));

//分组 类似于数据库中的group by
        Map<String, List<Student>> collect3 = pingpongClub.stream()
        .collect(Collectors.groupingBy(Student::getClassNum));

//字符串拼接 第一个参数是分隔符 第二个参数是前缀 第三个参数是后缀
        String collect4 = pingpongClub.stream().map(Student::getName).collect(Collectors.joining(",", "【", "】")); //【小u,小i,小m,小n】

//三个俱乐部符合年龄要求的按照班级分组
        Map<String, List<Student>> collect5 = Stream.of(basketballClub, pingpongClub, computerClub)
        .flatMap(e -> e.stream().filter(s -> s.getAge() < 17))
        .collect(Collectors.groupingBy(Student::getClassNum));

//按照是否年龄>16进行分组 key为true和false
        ConcurrentMap<Boolean, List<Student>> collect6 = Stream.of(basketballClub, pingpongClub, computerClub)
        .flatMap(Collection::stream)
        .collect(Collectors.groupingByConcurrent(s -> s.getAge() > 16));

Optional<Student> first = basketballClub.stream().findFirst();
if (first.isPresent()) {
    Student student = first.get();
    System.out.println(student);
}

Optional<Student> any = basketballClub.stream().findAny();
if (any.isPresent()) {
    Student student2 = any.get();
    System.out.println(student2);
}
Optional<Student> any1 = basketballClub.stream().parallel().findAny();
System.out.println(any1);

IntStream.range(1, 10)
   .peek(x -> System.out.print("\nA" + x))
   .limit(3)
   .peek(x -> System.out.print("B" + x))
   .forEach(x -> System.out.print("C" + x));

/**
 * Constructor for the head of a stream pipeline.
 *
 * @param source {@code Spliterator} describing the stream source
 * @param sourceFlags the source flags for the stream source, described in
 * {@link StreamOpFlag}
 * @param parallel {@code true} if the pipeline is parallel
 */
AbstractPipeline(Spliterator<?> source, int sourceFlags, boolean parallel) {
    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);
    this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
    this.depth = 0;
    this.parallel = parallel;
}

/**
 * Constructor for appending an intermediate operation stage onto an
 * existing pipeline.
 *
 * @param previousStage the upstream pipeline stage
 * @param opFlags the operation flags for the new stage, described in
 * {@link StreamOpFlag}
 */
AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
    if (previousStage.linkedOrConsumed)
        throw new IllegalStateException(MSG_STREAM_LINKED);
    previousStage.linkedOrConsumed = true;
    previousStage.nextStage = this;

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

interface Sink<T> extends Consumer<T> {
    /**
     * Resets the sink state to receive a fresh data set.  This must be called
     * before sending any data to the sink.  After calling {@link #end()},
     * you may call this method to reset the sink for another calculation.
     * @param size The exact size of the data to be pushed downstream, if
     * known or {@code -1} if unknown or infinite.
     *
     * <p>Prior to this call, the sink must be in the initial state, and after
     * this call it is in the active state.
     *
     * 开始遍历前调用，通知Sink做好准备
     */
    default void begin(long size) {}

    /**
     * Indicates that all elements have been pushed.  If the {@code Sink} is
     * stateful, it should send any stored state downstream at this time, and
     * should clear any accumulated state (and associated resources).
     *
     * <p>Prior to this call, the sink must be in the active state, and after
     * this call it is returned to the initial state.
     *
     * 所有元素遍历完成后调用，通知Sink没有更多元素了
     */
    default void end() {}
    
    /**
     * Indicates that this {@code Sink} does not wish to receive any more data.
     *
     * @implSpec The default implementation always returns false.
     *
     * @return true if cancellation is requested
     *
     * 是否可以结束操作，可以让短路操作尽早结束
     */
    default boolean cancellationRequested() {}

    /**
     * Accepts a value.
     *
     * @implSpec The default implementation throws IllegalStateException.
     *
     * @throws IllegalStateException if this sink does not accept values
     *
     * 遍历时调用，接收的一个待处理元素，并对元素进行处理。
     * Stage把自己包含的操作和回调方法封装到该方法里，前一个Stage只需要调用当前
     * Stage.accept方法即可
     */
    default void accept(T value) {}
}

// Stream.map 将生成一个新Stream
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) {
        // 该方法将回调函数(处理逻辑)包装成Sink
        @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) {
                    // 接收数据，使用当前包装的回调函数处理数据，并传递给下游Sink
                    downstream.accept(mapper.apply(u));
                }
            };
        }
    };
}

private static final class RefSortingSink<T> extends AbstractRefSortingSink<T> {
    // 存放用于排序的元素
    private ArrayList<T> list;

    RefSortingSink(Sink<? super T> sink, Comparator<? super T> comparator) {
        super(sink, comparator);
    }

    @Override
    public void begin(long size) {
        if (size >= Nodes.MAX_ARRAY_SIZE)
            throw new IllegalArgumentException(Nodes.BAD_SIZE);
        // 创建用于存放排序元素的列表
        list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();
    }

    @Override
    public void end() {
        // 只有在接收到所有元素后才开始排序
        list.sort(comparator);
        downstream.begin(list.size());
        // 排序完成后，将数据传递给下游Sink
        if (!cancellationWasRequested) {
            // 下游Sink不包含短路操作，将数据依次传递给下游Sink
            list.forEach(downstream::accept);
        }
        else {
            // 下游Sink包含短路操作
            for (T t : list) {
                // 对于每一个元素，都要询问是否可以结束处理
                if (downstream.cancellationRequested()) break;
                // 将元素传递给下游Sink
                downstream.accept(t);
            }
        }
        // 告知下游Sink数据传递完毕
        downstream.end();
        list = null;
    }

    @Override
    public void accept(T t) {
        // 依次将需要排序的元素加入到临时列表中
        list.add(t);
    }
}

/**
 * Evaluate the pipeline with a terminal operation to produce a result.
 *
 * @param <R> the type of result
 * @param terminalOp the terminal operation to be applied to the pipeline.
 * @return the result
 */
final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp /* 各种终止操作 */) {
    assert getOutputShape() == terminalOp.inputShape();
    if (linkedOrConsumed)
        throw new IllegalStateException(MSG_STREAM_LINKED);
    linkedOrConsumed = true;

    return isParallel()
           ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags())) /* 并发执行 */
           : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags())); /* 串行执行 */
}

final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink /* TerminalSink */, Spliterator<P_IN> spliterator) {
    copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
    return sink;
}

final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
    Objects.requireNonNull(wrappedSink);

    if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
        // 不包含短路操作
        
        // 1. begin
        wrappedSink.begin(spliterator.getExactSizeIfKnown());
        // 2. 遍历调用 sink.accept
        spliterator.forEachRemaining(wrappedSink);
        // 3. end
        wrappedSink.end();
    }
    else {
        // 包含短路操作
        copyIntoWithCancel(wrappedSink, spliterator);
    }
}

final <P_IN> void copyIntoWithCancel(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
    @SuppressWarnings({"rawtypes","unchecked"})
    AbstractPipeline p = AbstractPipeline.this;
    while (p.depth > 0) {
        p = p.previousStage;
    }
    // 1. begin
    wrappedSink.begin(spliterator.getExactSizeIfKnown());
    // 2. 遍历调用 sink.accept
    //    每一次遍历都询问cancellationRequested结果
    //    如果cancellationRequested为true，则中断遍历
    p.forEachWithCancel(spliterator, wrappedSink);
    // 3. end
    wrappedSink.end();
}