和消费流中的数据。下面展示了IteratorSpliterator的forEachRemaining,tryAdvance 两个方法的实现。可以看到,木有特别的地方,就是遍历元素并将指定操作施加于元素。
@Override
public void forEachRemaining(Consumer<? super T> action) {
if (action == null) throw new NullPointerException();
Iterator<? extends T> i;
if ((i = it) == null) {
i = it = collection.iterator();
est = (long)collection.size();
}
i.forEachRemaining(action);
}
@Override
public boolean tryAdvance(Consumer<? super T> action) {
if (action == null) throw new NullPointerException();
if (it == null) {
it = collection.iterator();
est = (long) collection.size();
}
if (it.hasNext()) {
action.accept(it.next());
return true;
}
return false;
}
整体流程就是这样。回顾一下:
- Collector 定义了必要的聚合操作函数;
- ReduceOps.makeRef 将 Collector 封装成一个计算对象 ReduceOps ,依赖的 ReducingSink 定义了具体的流数据消费过程;
- Spliterator 用于对流中的元素进行分区和遍历以及施加Sink指定的操作。
Pipeline
那么,Spliterator 又是从哪里来的呢?是通过类 java.util.stream.AbstractPipeline 的方法 sourceSpliterator 拿到的:
private Spliterator<?> sourceSpliterator(int terminalFlags) {
// Get the source spliterator of the pipeline
Spliterator<?> spliterator = null;
if (sourceStage.sourceSpliterator != null) {
spliterator = sourceStage.sourceSpliterator;
sourceStage.sourceSpliterator = null;
}
else if (sourceStage.sourceSupplier != null) {
spliterator = (Spliterator<?>) sourceStage.sourceSupplier.get();
sourceStage.sourceSupplier = null;
}
else {
throw new IllegalStateException(MSG_CONSUMED);
}
// code for isParallel
return spliterator;
}
这里的 sourceStage 是一个 AbstractPipeline。 Pipeline 是实现流式计算的流水线抽象,也是Stream的实现类。可以看到,java.util.stream 定义了四种 pipeline: DoublePipeline, IntPipeline, LongPipeline, ReferencePipeline。可以重点看 ReferencePipeline 的实现。比如 filter, map
abstract class ReferencePipeline<P_IN, P_OUT>
extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
implements Stream<P_OUT>
@Override
public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
Objects.requireNonNull(predicate);
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);
}
};
}
};
}
@Override
@SuppressWarnings("unchecked")
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) {
@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));
}
};
}
};
}
套路基本一样,关键点在于 accept 方法。filter 只在满足条件时将值传给下一个 pipeline, 而 map 将计算的值传给下一个 pipeline. StatelessOp 没有什么逻辑,JDK文档解释是:Base class for a stateless intermediate stage of a Stream。相应还有一个 StatefulOp, Head。 这些都是 ReferencePipeline ,负责将值在 pipeline 之间传递,交给 Sink 去计算。
st