网站建设截图,公司推进企业安全文化建设,品牌网页设计图片,亳州网站建设费用DistinctOpsDistinctOps 是一个专门用于实现 Stream.distinct() 操作的工厂类。正如它的名字所示#xff0c;它的核心职责就是创建能够去除流中重复元素的操作。distinct() 是一个有状态的中间操作 (stateful intermediate operation)#xff0c;这意味着它通常需要看到所有元…DistinctOpsDistinctOps 是一个专门用于实现 Stream.distinct() 操作的工厂类。正如它的名字所示它的核心职责就是创建能够去除流中重复元素的操作。distinct() 是一个有状态的中间操作 (stateful intermediate operation)这意味着它通常需要看到所有元素才能决定哪些元素可以进入下一阶段这使得它的实现比无状态操作如 filter, map要复杂得多。ReferencePipeline这个本身就是顶层的stream中调用Overridepublic final StreamP_OUT distinct() {return DistinctOps.makeRef(this);}下面我们分几个部分来详细解析 DistinctOps 类。类的定位和结构final class DistinctOps: 这是一个 final 类并且构造函数是 private 的表明它是一个纯粹的工具类Utility Class不能被继承或实例化。所有的功能都通过静态方法 makeRef 提供。makeRef(AbstractPipeline?, T, ? upstream): 这是该类的唯一入口。当调用 stream.distinct() 时底层实际上就是调用了这个 DistinctOps.makeRef() 方法。它接收上游的 Pipeline 作为输入然后返回一个新的 Pipeline 阶段这个新阶段就包含了去重逻辑。
// ... existing code ...
static T ReferencePipelineT, T makeRef(AbstractPipeline?, T, ? upstream) {return new ReferencePipeline.StatefulOpT, T(upstream, StreamShape.REFERENCE,StreamOpFlag.IS_DISTINCT | StreamOpFlag.NOT_SIZED) {// ... 具体的实现 ...};
}
// ... existing code ...
这里它创建了一个 ReferencePipeline.StatefulOp 的匿名子类实例。这立即告诉我们几点重要信息distinct() 是一个有状态操作 (StatefulOp)。它向上游声明了 IS_DISTINCT 标志表示从这个阶段输出的流是已经去重的。它还声明了 NOT_SIZED 标志因为去重后元素的数量是未知的除非处理完所有元素。核心实现串行 vs 并行有序 vs 无序distinct() 的复杂性在于它必须根据流的 并行性 (parallel) 和 有序性 (ordered) 采取截然不同的策略。DistinctOps 内部通过重写 StatefulOp 的几个关键方法来处理这些不同情况。串行执行 (opWrapSink)这是最简单的情况对应的是流的串行 (sequential) 执行。逻辑在 opWrapSink 方法中实现。
// ... existing code ...OverrideSinkT opWrapSink(int flags, SinkT sink) {Objects.requireNonNull(sink);if (StreamOpFlag.DISTINCT.isKnown(flags)) {// 如果上游已经去重什么都不做直接返回下游的 Sinkreturn sink;} else if (StreamOpFlag.SORTED.isKnown(flags)) {// 优化如果流已排序只需和前一个元素比较return new Sink.ChainedReference(sink) {boolean seenNull;T lastSeen;// ...Overridepublic void accept(T t) {if (t null) {if (!seenNull) { /* ... */ }} else if (lastSeen null || !t.equals(lastSeen)) {downstream.accept(lastSeen t);}}};} else {// 通用情况使用 Set 来记录见过的元素return new Sink.ChainedReference(sink) {SetT seen;Overridepublic void begin(long size) {seen new HashSet();downstream.begin(-1);}// ...Overridepublic void accept(T t) {if (seen.add(t)) { // Set.add() 返回 true 表示添加成功即之前没有downstream.accept(t);}}};}}
// ... existing code ...
逻辑分析:已去重 (DISTINCT.isKnown): 如果流已经被标记为去重的distinct() 就成了一个空操作no-op直接把元素传给下游。已排序 (SORTED.isKnown): 这是一个非常重要的优化。如果流是排序的那么重复的元素必然是相邻的。因此我们不需要一个 Set 来存储所有见过的元素只需要记住上一个元素 (lastSeen) 即可。这极大地降低了空间复杂度。通用情况: 对于无序的流唯一的办法就是用一个 HashSet 来存储所有已经见过的元素。每次来一个新元素尝试添加到 Set 中。如果 add 方法返回 true说明这是个新元素就把它传递给下游。并行执行 (opEvaluateParallel)这是最复杂的情况对应流的并行 (parallel) 执行。并行执行需要一个屏障 (barrier) 操作即必须处理完所有分片Spliterator的数据合并结果后才能继续。
// ... existing code ...OverrideP_IN NodeT opEvaluateParallel(PipelineHelperT helper,SpliteratorP_IN spliterator,IntFunctionT[] generator) {if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {// 已去重空操作return helper.evaluate(spliterator, false, generator);}else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {// 有序并行退化为串行处理模式来保证顺序return reduce(helper, spliterator);}else {// 无序并行最高效的并行模式AtomicBoolean seenNull new AtomicBoolean(false);ConcurrentHashMapT, Boolean map new ConcurrentHashMap();// ... 并行地将所有元素放入 ConcurrentHashMap 来去重forEachOp.evaluateParallel(helper, spliterator);// ...return Nodes.node(keys);}}
// ... existing code ...
逻辑分析:有序并行 (ORDERED.isKnown): 为了保证元素的顺序并行 distinct 无法做到真正的“懒加载”。它必须收集所有元素在一个地方完成去重然后再把结果交给下游。这里的 reduce 方法内部使用了 LinkedHashSet 来保证顺序这实际上是一个代价高昂的屏障操作。无序并行: 这是并行 distinct 性能最好的场景。因为它不关心顺序所以可以充分利用并行性。它使用 ConcurrentHashMap 作为共享的 Set因为 ConcurrentHashMap 是线程安全的而 HashSet 不是。每个线程都把自己分片中的元素尝试放入这个共享的 map 中。putIfAbsent 是一个原子操作可以保证只有一个线程能成功放入一个特定的元素。所有线程完成后map 的 keySet 就是最终去重后的结果集。ConcurrentHashMap 有一个限制它的 key 和 value 都不能为 null。为了处理流中可能存在的 null 元素代码使用了一个额外的 AtomicBoolean seenNull 标志位。如果流中出现了 null就将这个标志设为 true。并行懒加载 (opEvaluateParallelLazy)这个方法用于支持可以“懒加载”的并行操作。
// ... existing code ...OverrideP_IN SpliteratorT opEvaluateParallelLazy(PipelineHelperT helper, SpliteratorP_IN spliterator) {if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {// ...}else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {// 有序流不支持懒加载必须退化为屏障操作return reduce(helper, spliterator).spliterator();}else {// 无序流可以懒加载return new StreamSpliterators.DistinctSpliterator(helper.wrapSpliterator(spliterator));}}
// ... existing code ...
逻辑分析:对于有序流distinct 无法实现懒加载因为它需要看到所有元素才能确定最终顺序。所以它退化成了 opEvaluateParallel 中的 reduce 逻辑收集所有元素再返回一个新的 Spliterator。对于无序流可以实现懒加载逐个处理。它返回一个特制的 DistinctSpliterator这个 Spliterator 在内部包装了原始的 Spliterator并在 tryAdvance 时进行去重判断。MatchOpsMatchOps 是 Stream API 内部一个至关重要的工厂类它专门负责创建和管理 短路short-circuiting 的终端操作也就是我们常用的 anyMatch(), allMatch(), noneMatch()。首先看类的定义
final class MatchOps {private MatchOps() { }
//...
和 SortedOps、ReduceOps 类似它是一个包可见的、拥有私有构造函数的 final 类。这表明它是一个内部使用的工具工厂专门用于创建 TerminalOp终端操作的实例。它的核心职责是根据用户调用的匹配方法anyMatch 等和传入的谓词Predicate构建一个能够高效执行匹配逻辑的终端操作。短路Short-circuiting是理解 MatchOps 的关键。anyMatch, allMatch, noneMatch 都是短路操作。这意味着它们不一定需要处理流中的所有元素就能得出最终结果。anyMatch(p): 只要找到一个满足谓词 p 的元素结果就确定为 true无需再检查后续元素。allMatch(p): 只要找到一个不满足谓词 p 的元素结果就确定为 false无需再检查后续元素。noneMatch(p): 只要找到一个满足谓词 p 的元素结果就确定为 false无需再检查后续元素。这种“提前退出”的能力就是短路它能极大地提升在某些数据场景下的执行效率。MatchKind 枚举统一匹配逻辑为了用一套统一的逻辑来处理这三种不同的匹配规则MatchOps 设计了一个非常精巧的枚举 MatchKind。
// ... existing code ...enum MatchKind {/** Do any elements match the predicate? */ANY(true, true),/** Do all elements match the predicate? */ALL(false, false),/** Do no elements match the predicate? */NONE(true, false);private final boolean stopOnPredicateMatches;private final boolean shortCircuitResult;private MatchKind(boolean stopOnPredicateMatches,boolean shortCircuitResult) {this.stopOnPredicateMatches stopOnPredicateMatches;this.shortCircuitResult shortCircuitResult;}}
// ... existing code ...
这个枚举通过两个布尔值巧妙地描述了三种匹配行为的共性与差异stopOnPredicateMatches: 当谓词predicate的计算结果为 true 时是否应该停止处理ANY: 是。找到一个匹配的就停。ALL: 否。找到一个匹配的还不够必须继续找直到找到不匹配的或者遍历完。NONE: 是。找到一个匹配的就停因为结果已经确定是 false。shortCircuitResult: 如果发生了短路提前停止那么最终的结果应该是什么ANY: true。ALL: false。NONE: false。通过这个枚举后续的实现代码如下面的 MatchSink就可以写出通用的逻辑而无需为 any, all, none 分别写 if-else 分支。工厂方法与 MatchSinkMatchOps 提供了一系列 make... 工厂方法用于为不同类型的流Stream, IntStream 等创建匹配操作。我们以 makeRef 为例
// ... existing code ...public static T TerminalOpT, Boolean makeRef(Predicate? super T predicate,MatchKind matchKind) {Objects.requireNonNull(predicate);Objects.requireNonNull(matchKind);class MatchSink extends BooleanTerminalSinkT {MatchSink() {super(matchKind);}Overridepublic void accept(T t) {if (!stop predicate.test(t) matchKind.stopOnPredicateMatches) {stop true;value matchKind.shortCircuitResult;}}}return new MatchOp(StreamShape.REFERENCE, matchKind, MatchSink::new);}
// ... existing code ...
它接收一个 Predicate 和一个 MatchKind。内部定义了一个局部类 MatchSink它继承自 BooleanTerminalSink。Sink 是流中处理元素的末端。MatchSink 的 accept(T t) 方法是核心逻辑所在!stop: 检查是否已经有其他元素触发了短路。如果已经 stop则直接忽略当前元素。predicate.test(t) matchKind.stopOnPredicateMatches: 这是一个非常优雅的判断。它将当前元素的匹配结果 (true 或 false) 与 MatchKind 中定义的“停止条件”进行比较。对于 ANY: stopOnPredicateMatches 是 true。当 predicate.test(t) 为 true 时条件成立。对于 ALL: stopOnPredicateMatches 是 false。当 predicate.test(t) 为 false 时条件成立。对于 NONE: stopOnPredicateMatches 是 true。当 predicate.test(t) 为 true 时条件成立。如果条件成立就设置 stop true 来通知上游停止发送数据并设置 value matchKind.shortCircuitResult 来记录短路时的结果。最后它创建一个 MatchOp 实例并返回。MatchOp 是 TerminalOp 的一个具体实现它封装了 MatchKind 和用于创建 MatchSink 的 Supplier。并行执行MatchTask对于并行流MatchOp 的 evaluateParallel 方法会创建一个 MatchTask。
// ... existing code ...SuppressWarnings(serial)private static final class MatchTaskP_IN, P_OUTextends AbstractShortCircuitTaskP_IN, P_OUT, Boolean, MatchTaskP_IN, P_OUT {
// ... (constructors) ...Overrideprotected Boolean doLeaf() {boolean b helper.wrapAndCopyInto(op.sinkSupplier.get(), spliterator).getAndClearState();if (b op.matchKind.shortCircuitResult)shortCircuit(b);return null;}Overrideprotected Boolean getEmptyResult() {return !op.matchKind.shortCircuitResult;}}
// ... existing code ...
MatchTask 继承自 AbstractShortCircuitTask这是一个专为并行短路操作设计的 Fork/Join 任务。doLeaf(): 这是叶子任务执行的逻辑。它会对分配给自己的那一小部分数据执行匹配操作。helper.wrapAndCopyInto(...): 执行匹配得到这部分数据的结果 b。if (b op.matchKind.shortCircuitResult): 判断这个局部结果 b 是否就是最终的短路结果。例如对于 anyMatch如果这个叶子任务发现了一个匹配项它的结果 b 就会是 true这恰好等于 ANY 的 shortCircuitResult。shortCircuit(b): 如果发现了可以导致整个流短路的结果它会调用 shortCircuit。这个方法会通知 Fork/Join 框架一个最终结果已经找到了所有其他正在运行的、或者尚未开始的 MatchTask 都可以被取消了。这实现了并行的短路。getEmptyResult(): 如果所有任务都正常执行完毕没有发生短路那么这个方法返回最终的结果。例如对于 allMatch如果所有元素都匹配没有发生短路最终结果就是 true即 !op.matchKind.shortCircuitResult (!false)。总结MatchOps 是一个设计精良的内部工厂它通过以下方式优雅地实现了 anyMatch, allMatch, noneMatchMatchKind 枚举: 用两个布尔标志统一了三种匹配模式的逻辑避免了重复代码。MatchSink: 作为串行执行的核心它利用 MatchKind 的配置来实现通用的、可短路的元素处理逻辑。MatchTask: 作为并行执行的核心它利用 AbstractShortCircuitTask 的能力在 Fork/Join 框架下实现了高效的并行短路一旦任何一个子任务找到了决定性的结果就能迅速终止整个计算。FindOps它与我们之前讨论的 MatchOps 非常相似但专注于实现 findFirst() 和 findAny() 这两个终端操作。FindOps 是一个内部使用的工厂类其核心职责是创建用于“查找”操作的终端操作TerminalOp实例。
final class FindOps {private FindOps() { }
//...
和 MatchOps 一样它也是一个 final 的、拥有私有构造函数的工具类专门用于 Stream API 内部。它处理的 findFirst() 和 findAny() 都是短路操作一旦找到符合条件的元素就可以立即终止流的处理。findFirst() vs findAny()这是理解 FindOps 的关键区别点findFirst(): 必须返回流中遭遇顺序encounter order的第一个元素。这是一个有序操作。在并行流中即使其他线程更快地找到了一个元素也必须等待遭遇顺序更靠前的任务完成以确保返回的是“最左边”的那个结果。findAny(): 可以返回流中的任意一个元素。这是一个无序操作。在并行流中它允许任何一个线程只要找到了一个元素就可以立即短路整个计算而无需关心这个元素在原始流中的位置。这使得 findAny() 在并行流中的性能通常优于 findFirst()。FindOps 内部通过一个布尔标志 mustFindFirst 来区分这两种行为。FindOps 提供了一系列 make... 方法用于为不同类型的流对象、int、long、double创建查找操作。
// ... existing code ...SuppressWarnings(unchecked)public static T TerminalOpT, OptionalT makeRef(boolean mustFindFirst) {return (TerminalOpT, OptionalT)(mustFindFirst ? FindSink.OfRef.OP_FIND_FIRST : FindSink.OfRef.OP_FIND_ANY);}
// ... existing code ...
这些方法非常简洁。它们根据 mustFindFirst 参数直接返回一个预先创建好的、静态的 TerminalOp 实例。这些实例如 OP_FIND_FIRST被定义在内部类 FindSink 中。这种方式避免了重复创建对象提高了效率。FindSink - 串行执行的核心FindSink 是实现查找逻辑的 TerminalSink。
// ... existing code ...private abstract static class FindSinkT, O implements TerminalSinkT, O {boolean hasValue;T value;// ...Overridepublic void accept(T value) {if (!hasValue) {hasValue true;this.value value;}}Overridepublic boolean cancellationRequested() {return hasValue;}
// ... existing code ...
它的逻辑非常直接accept(T value): 当接收到第一个元素时将其保存在 this.value 中并将 hasValue 标志设为 true。由于 if (!hasValue) 的判断后续所有到达的元素都会被忽略。cancellationRequested(): 一旦 hasValue 变为 true此方法就返回 true。这会通知上游的流管道“我已经找到结果了请不要再发送更多元素了”从而实现串行流的短路。FindSink 有多个静态内部子类OfRef, OfInt 等用于处理不同数据类型并预先创建了 OP_FIND_FIRST 和 OP_FIND_ANY 这两个静态常量。FindOp - 终端操作的封装FindOp 是 TerminalOp 接口的具体实现它封装了查找操作的所有元信息。
// ... existing code ...private static final class FindOpT, O implements TerminalOpT, O {private final StreamShape shape;final int opFlags;final O emptyValue;// ...final SupplierTerminalSinkT, O sinkSupplier;FindOp(boolean mustFindFirst,StreamShape shape,O emptyValue,PredicateO presentPredicate,SupplierTerminalSinkT, O sinkSupplier) {this.opFlags StreamOpFlag.IS_SHORT_CIRCUIT | (mustFindFirst ? 0 : StreamOpFlag.NOT_ORDERED);// ...}
// ... existing code ...
opFlags: 操作标志。所有查找操作都是 IS_SHORT_CIRCUIT。对于 findAny (mustFindFirst 为 false)还会额外加上 NOT_ORDERED 标志告知流管道可以进行无序优化。evaluateSequential(...): 定义了串行执行的逻辑即创建一个 FindSink 并让流把元素送入其中。evaluateParallel(...): 定义了并行执行的逻辑它会创建一个 FindTask 并启动它。FindTask - 并行执行的核心FindTask 是 AbstractShortCircuitTask 的子类负责并行查找。
// ... existing code ...private static final class FindTaskP_IN, P_OUT, Oextends AbstractShortCircuitTaskP_IN, P_OUT, O, FindTaskP_IN, P_OUT, O {private final FindOpP_OUT, O op;private final boolean mustFindFirst;
// ...Overrideprotected O doLeaf() {O result helper.wrapAndCopyInto(op.sinkSupplier.get(), spliterator).get();if (!mustFindFirst) { // findAny 逻辑if (result ! null)shortCircuit(result); // 找到任何一个立即短路return null;}else { // findFirst 逻辑if (result ! null) {foundResult(result); // 找到一个需要通过 cancelLaterNodes 确保有序性return result;}elsereturn null;}}
// ...}
// ... existing code ...
doLeaf() 方法清晰地展示了 findFirst 和 findAny 在并行时的区别findAny (!mustFindFirst): 叶子任务只要处理自己的数据分片并找到了一个元素 (result ! null)就立刻调用 shortCircuit(result)。这会尝试将结果写入共享的 AtomicReference并触发全局的短路所有其他任务都会尽快停止。findFirst (mustFindFirst): 叶子任务找到一个元素后不能直接宣布胜利。它需要调用 foundResult(result)这个方法内部会调用 cancelLaterNodes()。这会取消掉所有处理“更右边”遭遇顺序更靠后数据的任务同时允许“更左边”的任务继续执行。最终只有最左边的那个找到了结果的任务其结果才会被采纳为最终结果。这个过程通过 onCompletion 方法中的逻辑来保证。private void foundResult(O answer) {if (isLeftmostNode())shortCircuit(answer);elsecancelLaterNodes();}AbstractTask::protected boolean isLeftmostNode() {SuppressWarnings(unchecked)K node (K) this;while (node ! null) {K parent node.getParent();if (parent ! null parent.leftChild ! node)return false;node parent;}return true;}总结FindOps 通过一系列精心设计的内部类为 findFirst 和 findAny 提供了统一且高效的实现工厂方法 (make...): 提供简洁的入口并复用预创建的 TerminalOp 实例。FindSink: 实现了串行查找的短路逻辑。FindOp: 封装了操作的元数据并区分了 findFirst 和 findAny 的标志。FindTask: 作为 AbstractShortCircuitTask 的子类它为并行查找提供了核心实现并巧妙地利用父类的短路和取消机制分别实现了 findAny 的“快速响应”和 findFirst 的“有序保证”。
