Java流操作解析：深度剖析中间操作、终端操作与并行处理机制

static <T> ReferencePipeline<T, T> makeRef(AbstractPipeline<?, T, ?> upstream) {
    // 创建一个新的 StatefulOp 实例，表示去重操作的流水线，使用 REFERENCE 类型的流形状
    return new ReferencePipeline.StatefulOp<T, T>(upstream, StreamShape.REFERENCE,
                                                  StreamOpFlag.IS_DISTINCT | StreamOpFlag.NOT_SIZED) {

        // reduce 方法用于将并行处理的元素归约为单个结果
        <P_IN> Node<T> reduce(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {
            // 如果流是有序的，则保持排序顺序
            TerminalOp<T, LinkedHashSet<T>> reduceOp
                    = ReduceOps.<T, LinkedHashSet<T>>makeRef(LinkedHashSet::new, LinkedHashSet::add,
                                                             LinkedHashSet::addAll);
            // 使用 reduceOp 对元素进行归约操作，并将结果封装为 Node
            return Nodes.node(reduceOp.evaluateParallel(helper, spliterator));
        }

        // opEvaluateParallel 方法用于并行评估操作
        @Override
        <P_IN> Node<T> opEvaluateParallel(PipelineHelper<T> helper,
                                          Spliterator<P_IN> spliterator,
                                          IntFunction<T[]> generator) {
            // 如果流中已经包含了 DISTINCT 标志，表示已经进行了去重操作，则直接返回
            if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {
                // 不进行任何操作，直接返回流的结果
                return helper.evaluate(spliterator, false, generator);
            }
            // 如果流中已经包含了 ORDERED 标志，表示流是有序的
            else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {
                // 调用 reduce 方法进行归约操作
                return reduce(helper, spliterator);
            }
            // 如果流不是有序的
            else {
                // 用于标记是否有 null 值出现的原子布尔值
                AtomicBoolean seenNull = new AtomicBoolean(false);
                // 使用 ConcurrentHashMap 存储元素，保证线程安全
                ConcurrentHashMap<T, Boolean> map = new ConcurrentHashMap<>();
                // 使用 ForEachOps 进行并行遍历并添加元素到 ConcurrentHashMap
                TerminalOp<T, Void> forEachOp = ForEachOps.makeRef(t -> {
                    if (t == null)
                        seenNull.set(true);  // 如果元素为 null，则设置标志为 true
                    else
                        map.putIfAbsent(t, Boolean.TRUE);  // 如果元素不为 null，则添加到 ConcurrentHashMap 中
                }, false);
                forEachOp.evaluateParallel(helper, spliterator);

                // 如果出现 null 元素，则将其加入到结果中
                Set<T> keys = map.keySet();
                if (seenNull.get()) {
                    // 如果有 null 元素，则创建一个支持 null 元素的 HashSet 并添加到结果中
                    keys = new HashSet<>(keys);
                    keys.add(null);
                }
                // 返回包含去重结果的 Node
                return Nodes.node(keys);
            }
        }
    };
}

/**
* ArrayList::new作为一个Supplier，以及List::add作为一个累加器函数，可以说是使用了工厂模式和策略模式
*/
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);
}

public long count() {
    return mapToLong(e -> 1L).sum();
}