Java 8 Stream流底层原理

• 函数式接口
  • 操作
  • 流程
  • Collection
  • AbstractPipeline
  • ReferencePipeline
  • Head
  • StatelessOp
  • StatefulOp
  • TerminalOp
  • ReduceOp
  • MatchOp
  • FindOp
  • ForEachOp
  • Sink
  • ChainedReference
  • TerminalSink
  • Collector
  • 并行流
  • ForkJoinTask
  • AbstractTask

---

函数式接口

初识lambda呢，函数式接口肯定是绕不过去的，函数式接口就是一个有且仅有一个抽象方法，但是可以有多个非抽象方法的接口。函数式接口可以被隐式转换为lambda表达式。

@FunctionalInterface
publicinterfaceCloseable{

voidclose();
}

在java.util.function它包含了很多类，用来支持Java的函数式编程，该包中的函数式接口有：

操作

流程

Stream相关接口继承图：

Stream流水线组织结构示意图（图是盗的）：

Collection

类路径java.util.colltction

@Override
defaultSpliteratorspliterator(){
returnSpliterators.spliterator(this,0);
}
//常用Stream流转换
defaultStreamstream(){
returnStreamSupport.stream(spliterator(),false);
}
//并行流
defaultStreamparallelStream(){
returnStreamSupport.stream(spliterator(),true);
}

//java.util.stream.StreamSupport#stream(java.util.Spliterator,boolean)
publicstaticStreamstream(Spliteratorspliterator,booleanparallel){
Objects.requireNonNull(spliterator);
returnnewReferencePipeline.Head<>(spliterator,StreamOpFlag.fromCharacteristics(spliterator),parallel);
}

AbstractPipeline

类路径java.util.stream.AbstractPipeline

//反向链接到管道链的头部（如果是源阶段，则为自身）。
privatefinalAbstractPipelinesourceStage;

//“上游”管道，如果这是源阶段，则为null。
privatefinalAbstractPipelinepreviousStage;

//此管道对象表示的中间操作的操作标志。
protectedfinalintsourceOrOpFlags;

//管道中的下一个阶段；如果这是最后一个阶段，则为null。在链接到下一个管道时有效地结束。
privateAbstractPipelinenextStage;

//如果是顺序的，则此管道对象与流源之间的中间操作数；如果是并行的，则为先前有状态的中间操作数。在管道准备进行评估时有效。
privateintdepth;

//源和所有操作的组合源标志和操作标志，直到此流水线对象表示的操作为止（包括该流水线对象所代表的操作）。在管道准备进行评估时有效。
privateintcombinedFlags;

//源拆分器。仅对头管道有效。如果管道使用非null值，那么在使用管道之前，sourceSupplier必须为null。在使用管道之后，如果非null，则将其设置为null。
privateSpliteratorsourceSpliterator;

//来源供应商。仅对头管道有效。如果非null，则在使用管道之前，sourceSpliterator必须为null。在使用管道之后，如果非null，则将其设置为null。
privateSupplier>sourceSupplier;

//如果已链接或使用此管道，则为True
privatebooleanlinkedOrConsumed;

//如果正在执行任何有状态操作，则为true；否则为true。仅对源阶段有效。
privatebooleansourceAnyStateful;

privateRunnablesourceCloseAction;

//如果管道是并行的，则为true；否则，管道为顺序的；否则为true。仅对源阶段有效。
privatebooleanparallel;

ReferencePipeline

类路径：java.util.stream.ReferencePipeline

filter

//java.util.stream.ReferencePipeline#filter
@Override
publicfinalStreamfilter(PredicatesuperP_OUT>predicate){
Objects.requireNonNull(predicate);
//返回一个匿名无状态的管道
returnnewStatelessOp(this,StreamShape.REFERENCE,StreamOpFlag.NOT_SIZED){
//下游生产线所需要的回调接口
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
@Override
publicvoidbegin(longsize){
downstream.begin(-1);
}
//真正执行操作的方法，依靠ChainedReference内置ReferencePipeline引用下游的回调
@Override
publicvoidaccept(P_OUTu){
//只有满足条件的元素才能被下游执行
if(predicate.test(u))
downstream.accept(u);
}
};
}
};
}

map

//java.util.stream.ReferencePipeline#map
publicfinalStreammap(FunctionsuperP_OUT,?extendsR>mapper){
Objects.requireNonNull(mapper);
//返回一个匿名无状态的管道
returnnewStatelessOp(this,StreamShape.REFERENCE,StreamOpFlag.NOT_SORTED|StreamOpFlag.NOT_DISTINCT){
//下游生产线所需要的回调接口
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
//真正执行操作的方法，依靠ChainedReference内置ReferencePipeline引用下游的回调
@Override
publicvoidaccept(P_OUTu){
//执行转换后提供给下游执行
downstream.accept(mapper.apply(u));
}
};
}
};
}

flatMap

//java.util.stream.ReferencePipeline#flatMap
@Override
publicfinalStreamflatMap(FunctionsuperP_OUT,?extendsStream>mapper){
Objects.requireNonNull(mapper);
//返回一个匿名无状态的管道
returnnewStatelessOp(this,StreamShape.REFERENCE,
StreamOpFlag.NOT_SORTED|StreamOpFlag.NOT_DISTINCT|StreamOpFlag.NOT_SIZED){
//下游生产线所需要的回调接口
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
@Override
publicvoidbegin(longsize){
downstream.begin(-1);
}
//真正执行操作的方法，依靠ChainedReference内置ReferencePipeline引用下游的回调
@Override
publicvoidaccept(P_OUTu){
try(Streamresult=mapper.apply(u)){
//划分为多个流执行下游（分流）
if(result!=null)
result.sequential().forEach(downstream);
}
}
};
}
};
}

peek

//java.util.stream.ReferencePipeline#peek
@Override
publicfinalStreampeek(ConsumersuperP_OUT>action){
Objects.requireNonNull(action);
//返回一个匿名无状态的管道
returnnewStatelessOp(this,StreamShape.REFERENCE,0){
//下游生产线所需要的回调接口
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
//真正执行操作的方法，依靠ChainedReference内置ReferencePipeline引用下游的回调
@Override
publicvoidaccept(P_OUTu){
//先执行自定义方法，在执行下游方法
action.accept(u);
downstream.accept(u);
}
};
}
};
}

sorted

@Override
publicfinalStreamsorted(){
//不提供Comparator，会使用元素自实现Comparator的compareTo方法
returnSortedOps.makeRef(this);
}

@Override
publicfinalStreamsorted(ComparatorsuperP_OUT>comparator){
returnSortedOps.makeRef(this,comparator);
}
//Sorted.makeRef
staticStreammakeRef(AbstractPipelineupstream,
ComparatorsuperT>comparator){
returnnewOfRef<>(upstream,comparator);
}
//ofRef类
privatestaticfinalclassOfRef<T>extendsReferencePipeline.StatefulOp<T,T>{

privatefinalbooleanisNaturalSort;
privatefinalComparatorsuperT>comparator;

@Override
publicSinkopWrapSink(intflags,Sinksink){
Objects.requireNonNull(sink);
//根据不同的flag进行不同排序
if(StreamOpFlag.SORTED.isKnown(flags)&&isNaturalSort)
returnsink;
elseif(StreamOpFlag.SIZED.isKnown(flags))
returnnewSizedRefSortingSink<>(sink,comparator);
else
returnnewRefSortingSink<>(sink,comparator);
}

}

distinct

@Override
publicfinalStreamdistinct(){
returnDistinctOps.makeRef(this);
}
staticReferencePipelinemakeRef(AbstractPipelineupstream){
//返回一个匿名有状态的管道
returnnewReferencePipeline.StatefulOp(upstream,StreamShape.REFERENCE,StreamOpFlag.IS_DISTINCT|StreamOpFlag.NOT_SIZED){

@Override
SinkopWrapSink(intflags,Sinksink){
Objects.requireNonNull(sink);

if(StreamOpFlag.DISTINCT.isKnown(flags)){
//已经是去重过了
returnsink;
}elseif(StreamOpFlag.SORTED.isKnown(flags)){
//有序流
returnnewSink.ChainedReference(sink){
booleanseenNull;
//这个为先执行的前序元素
TlastSeen;

@Override
publicvoidbegin(longsize){
seenNull=false;
lastSeen=null;
downstream.begin(-1);
}

@Override
publicvoidend(){
seenNull=false;
lastSeen=null;
downstream.end();
}
//这里通过有序的特性，前序元素与后序元素比较，如果相等则跳过执行后序的元素
@Override
publicvoidaccept(Tt){
if(t==null){
//这里控制元素为null只有一个
if(!seenNull){
seenNull=true;
downstream.accept(lastSeen=null);
}
}elseif(lastSeen==null||!t.equals(lastSeen)){
//这里将前序元素赋值给lastSeen
downstream.accept(lastSeen=t);
}
}
};
}else{
//底层通过Set进行去重，所以该元素需要重写hashCode和equals方法
returnnewSink.ChainedReference(sink){
Setseen;

@Override
publicvoidbegin(longsize){
seen=newHashSet<>();
downstream.begin(-1);
}

@Override
publicvoidend(){
seen=null;
downstream.end();
}

@Override
publicvoidaccept(Tt){
if(!seen.contains(t)){
seen.add(t);
downstream.accept(t);
}
}
};
}
}
};
}

skip、limit

publicstaticStreammakeRef(AbstractPipelineupstream,
longskip,longlimit){
if(skip< 0)
thrownewIllegalArgumentException("Skipmustbenon-negative:"+skip);
//返回一个匿名有状态的管道
returnnewReferencePipeline.StatefulOp(upstream,StreamShape.REFERENCE,flags(limit)){
SpliteratorunorderedSkipLimitSpliterator(Spliterators,longskip,longlimit,longsizeIfKnown){
if(skip<= sizeIfKnown) {
                    limit = limit >=0?Math.min(limit,sizeIfKnown-skip):sizeIfKnown-skip;
skip=0;
}
returnnewStreamSpliterators.UnorderedSliceSpliterator.OfRef<>(s,skip,limit);
}
//自己实现真正操作的方法
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
longn=skip;
longm=limit>=0?limit:Long.MAX_VALUE;

@Override
publicvoidbegin(longsize){
downstream.begin(calcSize(size,skip,m));
}

@Override
publicvoidaccept(Tt){
if(n==0){
//limit
if(m>0){
m--;
downstream.accept(t);
}
}
//skip
else{
n--;
}
}

@Override
publicbooleancancellationRequested(){
returnm==0||downstream.cancellationRequested();
}
};
}
};
}

reduce

//java.util.stream.ReferencePipeline#reduce(P_OUT,java.util.function.BinaryOperator)
@Override
publicfinalP_OUTreduce(finalP_OUTidentity,finalBinaryOperatoraccumulator){
returnevaluate(ReduceOps.makeRef(identity,accumulator,accumulator));
}
//java.util.stream.ReferencePipeline#reduce(java.util.function.BinaryOperator)
@Override
publicfinalOptionalreduce(BinaryOperatoraccumulator){
returnevaluate(ReduceOps.makeRef(accumulator));
}
//java.util.stream.ReferencePipeline#reduce(R,java.util.function.BiFunction,java.util.function.BinaryOperator)
@Override
publicfinalRreduce(Ridentity,BiFunctionsuperP_OUT,R>accumulator,BinaryOperatorcombiner){
returnevaluate(ReduceOps.makeRef(identity,accumulator,combiner));
}

//java.util.stream.AbstractPipeline#evaluate(java.util.stream.TerminalOp)
finalRevaluate(TerminalOpterminalOp){
assertgetOutputShape()==terminalOp.inputShape();
if(linkedOrConsumed)
thrownewIllegalStateException(MSG_STREAM_LINKED);
linkedOrConsumed=true;

returnisParallel()
?terminalOp.evaluateParallel(this,sourceSpliterator(terminalOp.getOpFlags()))
:terminalOp.evaluateSequential(this,sourceSpliterator(terminalOp.getOpFlags()));
}

collect

//java.util.stream.ReferencePipeline#collect(java.util.stream.Collector)
@Override
@SuppressWarnings("unchecked")
publicfinalRcollect(CollectorsuperP_OUT,A,R>collector){
Acontainer;
if(isParallel()
&&(collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
&&(!isOrdered()||collector.characteristics().contains(Collector.Characteristics.UNORDERED))){
container=collector.supplier().get();
BiConsumersuperP_OUT>accumulator=collector.accumulator();
forEach(u->accumulator.accept(container,u));
}
else{
container=evaluate(ReduceOps.makeRef(collector));
}
//具有特定转换的使用finisher处理
returncollector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
?(R)container
:collector.finisher().apply(container);
}
//java.util.stream.ReferencePipeline#collect(java.util.function.Supplier,java.util.function.BiConsumer,java.util.function.BiConsumer)
@Override
publicfinalRcollect(Suppliersupplier,BiConsumersuperP_OUT>accumulator,BiConsumercombiner){
returnevaluate(ReduceOps.makeRef(supplier,accumulator,combiner));
}

//java.util.stream.AbstractPipeline#evaluate(java.util.stream.TerminalOp)
finalRevaluate(TerminalOpterminalOp){
assertgetOutputShape()==terminalOp.inputShape();
if(linkedOrConsumed)
thrownewIllegalStateException(MSG_STREAM_LINKED);
linkedOrConsumed=true;

returnisParallel()
?terminalOp.evaluateParallel(this,sourceSpliterator(terminalOp.getOpFlags()))
:terminalOp.evaluateSequential(this,sourceSpliterator(terminalOp.getOpFlags()));
}

forEach

//java.util.stream.ReferencePipeline#forEach
@Override
publicvoidforEach(ConsumersuperP_OUT>action){
evaluate(ForEachOps.makeRef(action,false));
}

//java.util.stream.ForEachOps#makeRef
publicstaticTerminalOpmakeRef(ConsumersuperT>action,booleanordered){
Objects.requireNonNull(action);
returnnewForEachOp.OfRef<>(action,ordered);
}

//java.util.stream.ForEachOps.ForEachOp.OfRef
staticfinalclassOfRef<T>extendsForEachOp<T>{
finalConsumersuperT>consumer;

OfRef(ConsumersuperT>consumer,booleanordered){
super(ordered);
this.consumer=consumer;
}

//只是简单的消费
@Override
publicvoidaccept(Tt){
consumer.accept(t);
}
}

Head

流的数据元的头，类路径java.util.stream.ReferencePipeline.Head

//java.util.stream.ReferencePipeline.Head
staticclassHead<E_IN,E_OUT>extendsReferencePipeline<E_IN,E_OUT>{

Head(Supplier>source,intsourceFlags,booleanparallel){
super(source,sourceFlags,parallel);
}

Head(Spliteratorsource,intsourceFlags,booleanparallel){
super(source,sourceFlags,parallel);
}

@Override
finalbooleanopIsStateful(){
thrownewUnsupportedOperationException();
}

@Override
finalSinkopWrapSink(intflags,Sinksink){
thrownewUnsupportedOperationException();
}
//Optimizedsequentialterminaloperationsfortheheadofthepipeline
@Override
publicvoidforEach(ConsumersuperE_OUT>action){
if(!isParallel()){
sourceStageSpliterator().forEachRemaining(action);
}
else{
super.forEach(action);
}
}

@Override
publicvoidforEachOrdered(ConsumersuperE_OUT>action){
if(!isParallel()){
sourceStageSpliterator().forEachRemaining(action);
}
else{
super.forEachOrdered(action);
}
}
}

StatelessOp

无状态的中间管道，类路径java.util.stream.ReferencePipeline.StatelessOp

//java.util.stream.ReferencePipeline.StatelessOp
abstractstaticclassStatelessOp<E_IN,E_OUT>extendsReferencePipeline<E_IN,E_OUT>{

StatelessOp(AbstractPipelineupstream,StreamShapeinputShape,intopFlags){
super(upstream,opFlags);
assertupstream.getOutputShape()==inputShape;
}

@Override
finalbooleanopIsStateful(){
returnfalse;
}
}

StatefulOp

有状态的中间管道，类路径java.util.stream.ReferencePipeline.StatefulOp

//java.util.stream.ReferencePipeline.StatefulOp
abstractstaticclassStatefulOp<E_IN,E_OUT>extendsReferencePipeline<E_IN,E_OUT>{

StatefulOp(AbstractPipelineupstream,StreamShapeinputShape,intopFlags){
super(upstream,opFlags);
assertupstream.getOutputShape()==inputShape;
}

@Override
finalbooleanopIsStateful(){
returntrue;
}

@Override
abstractNodeopEvaluateParallel(PipelineHelperhelper,
Spliteratorspliterator,
IntFunctiongenerator);

TerminalOp

管道流的结束操作，类路径java.util.stream.TerminalOp

interfaceTerminalOp<E_IN,R>{

//获取此操作的输入类型的形状
defaultStreamShapeinputShape(){returnStreamShape.REFERENCE;}

//获取操作的流标志。终端操作可以设置StreamOpFlag定义的流标志的有限子集，并且这些标志与管道的先前组合的流和中间操作标志组合在一起。
defaultintgetOpFlags(){return0;}

//使用指定的PipelineHelper对操作执行并行评估，该操作描述上游中间操作。
defaultRevaluateParallel(PipelineHelperhelper,Spliteratorspliterator){
if(Tripwire.ENABLED)
Tripwire.trip(getClass(),"{0}triggeringTerminalOp.evaluateParallelserialdefault");
returnevaluateSequential(helper,spliterator);
}

//使用指定的PipelineHelper对操作执行顺序评估，该操作描述上游中间操作。
RevaluateSequential(PipelineHelperhelper,Spliteratorspliterator);
}

ReduceOp

类路径java.util.stream.ReduceOps.ReduceOp

privatestaticabstractclassReduceOp<T,R,SextendsAccumulatingSink<T,R,S>>implementsTerminalOp<T,R>{
privatefinalStreamShapeinputShape;

ReduceOp(StreamShapeshape){
inputShape=shape;
}

publicabstractSmakeSink();

@Override
publicStreamShapeinputShape(){
returninputShape;
}

//通过匿名子类实现makeSink()获取Sink
@Override
publicRevaluateSequential(PipelineHelperhelper,Spliteratorspliterator){
returnhelper.wrapAndCopyInto(makeSink(),spliterator).get();
}

@Override
publicRevaluateParallel(PipelineHelperhelper,Spliteratorspliterator){
returnnewReduceTask<>(this,helper,spliterator).invoke().get();
}
}

MatchOp

类路径java.util.stream.MatchOps.MatchOp

privatestaticfinalclassMatchOp<T>implementsTerminalOp<T,Boolean>{
privatefinalStreamShapeinputShape;
finalMatchKindmatchKind;
finalSupplier>sinkSupplier;

MatchOp(StreamShapeshape,MatchKindmatchKind,Supplier>sinkSupplier){
this.inputShape=shape;
this.matchKind=matchKind;
this.sinkSupplier=sinkSupplier;
}

@Override
publicintgetOpFlags(){
returnStreamOpFlag.IS_SHORT_CIRCUIT|StreamOpFlag.NOT_ORDERED;
}

@Override
publicStreamShapeinputShape(){
returninputShape;
}

//使用内置的sinkSupplier获取Sink
@Override
publicBooleanevaluateSequential(PipelineHelperhelper,Spliteratorspliterator){
returnhelper.wrapAndCopyInto(sinkSupplier.get(),spliterator).getAndClearState();
}

@Override
publicBooleanevaluateParallel(PipelineHelperhelper,Spliteratorspliterator){
returnnewMatchTask<>(this,helper,spliterator).invoke();
}
}

FindOp

类路径java.util.stream.FindOps.FindOp

privatestaticfinalclassFindOp<T,O>implementsTerminalOp<T,O>{
privatefinalStreamShapeshape;
finalbooleanmustFindFirst;
finalOemptyValue;
finalPredicatepresentPredicate;
finalSupplier>sinkSupplier;

FindOp(booleanmustFindFirst,
StreamShapeshape,
OemptyValue,
PredicatepresentPredicate,
Supplier>sinkSupplier){
this.mustFindFirst=mustFindFirst;
this.shape=shape;
this.emptyValue=emptyValue;
this.presentPredicate=presentPredicate;
this.sinkSupplier=sinkSupplier;
}

@Override
publicintgetOpFlags(){
returnStreamOpFlag.IS_SHORT_CIRCUIT|(mustFindFirst?0:StreamOpFlag.NOT_ORDERED);
}

@Override
publicStreamShapeinputShape(){
returnshape;
}

//通过内置sinkSupplier获取Sink
@Override
publicOevaluateSequential(PipelineHelperhelper,Spliteratorspliterator){
Oresult=helper.wrapAndCopyInto(sinkSupplier.get(),spliterator).get();
returnresult!=null?result:emptyValue;
}

@Override
publicOevaluateParallel(PipelineHelperhelper,Spliteratorspliterator){
returnnewFindTask<>(this,helper,spliterator).invoke();
}
}

ForEachOp

类路径java.util.stream.ForEachOps.ForEachOp

staticabstractclassForEachOp<T>implementsTerminalOp<T,Void>,TerminalSink<T,Void>{
privatefinalbooleanordered;

protectedForEachOp(booleanordered){
this.ordered=ordered;
}

@Override
publicintgetOpFlags(){
returnordered?0:StreamOpFlag.NOT_ORDERED;
}

//自己实现了Sink
@Override
publicVoidevaluateSequential(PipelineHelperhelper,Spliteratorspliterator){
returnhelper.wrapAndCopyInto(this,spliterator).get();
}

@Override
publicVoidevaluateParallel(PipelineHelperhelper,Spliteratorspliterator){
if(ordered)
newForEachOrderedTask<>(helper,spliterator,this).invoke();
else
newForEachTask<>(helper,spliterator,helper.wrapSink(this)).invoke();
returnnull;
}

@Override
publicVoidget(){
returnnull;
}

staticfinalclassOfRef<T>extendsForEachOp<T>{
finalConsumersuperT>consumer;

OfRef(ConsumersuperT>consumer,booleanordered){
super(ordered);
this.consumer=consumer;
}

@Override
publicvoidaccept(Tt){
consumer.accept(t);
}
}
...
}

Sink

类路径java.util.stream.Sink

interfaceSink<T>extendsConsumer<T>{
//开始遍历元素之前调用该方法，通知Sink做好准备。
defaultvoidbegin(longsize){}
//所有元素遍历完成之后调用，通知Sink没有更多的元素了。
defaultvoidend(){}
//是否可以结束操作，可以让短路操作尽早结束。
defaultbooleancancellationRequested(){
returnfalse;
}
//遍历元素时调用，接受一个待处理元素，并对元素进行处理。Stage把自己包含的操作和回调方法封装到该方法里，前一个Stage只需要调用当前Stage.accept(Tt)方法就行了。
voidaccept(Tt);
}

这里Sink的子类实现中分为两种：中间操作匿名实现ChainedReference和TerminalOp子类所提供的Sink。

ChainedReference

类路径java.util.stream.Sink.ChainedReference，这里是中间操作的默认模板父类

staticabstractclassChainedReference<T,E_OUT>implementsSink<T>{
protectedfinalSinksuperE_OUT>downstream;

publicChainedReference(SinksuperE_OUT>downstream){
this.downstream=Objects.requireNonNull(downstream);
}

@Override
publicvoidbegin(longsize){
downstream.begin(size);
}

@Override
publicvoidend(){
downstream.end();
}

@Override
publicbooleancancellationRequested(){
returndownstream.cancellationRequested();
}
}

在上述的中间操作管道流中都是通过匿名类继承ChainedReference实现onWrapSink(int, Sink)返回一个指定操作的Sink。

TerminalSink

这里为什么讲提供呢？这是因为不同的实现TerminalOp的子类中在实现java.util.stream.TerminalOp#evaluateSequential中都是通过helper.wrapAndCopyInto(TerminalOp子类实现提供的Sink, spliterator)中通过参数传递的方式提供的，不同的子类传递的方式不一样所以此处用了一个提供Sink

由ReduceOps中实现TerminalOp所提供的ReducingSink，它是由匿名类实现java.util.stream.ReduceOps.ReduceOp#makeSink来交付给helper.wrapAndCopyInto(makeSink(), spliterator)的。

publicstaticTerminalOpmakeRef(Useed,BiFunctionsuperT,U>reducer,BinaryOperatorcombiner){
Objects.requireNonNull(reducer);
Objects.requireNonNull(combiner);
classReducingSinkextendsBox<U>implementsAccumulatingSink<T,U,ReducingSink>{
@Override
publicvoidbegin(longsize){
state=seed;
}

@Override
publicvoidaccept(Tt){
state=reducer.apply(state,t);
}

@Override
publicvoidcombine(ReducingSinkother){
state=combiner.apply(state,other.state);
}
}
returnnewReduceOp(StreamShape.REFERENCE){
@Override
publicReducingSinkmakeSink(){
returnnewReducingSink();
}
};
}

由ForEachOps中实现TerminalOp所提供的是this，它的提供方式就是通过this交付给helper.wrapAndCopyInto(this, spliterator)。

//这里ForEachOp自己通过TerminalSink间接的实现了Sink
staticabstractclassForEachOp<T>implementsTerminalOp<T,Void>,TerminalSink<T,Void>{
@Override
publicVoidevaluateSequential(PipelineHelperhelper,Spliteratorspliterator){
returnhelper.wrapAndCopyInto(this,spliterator).get();
}
}

由MatchOps中实现TerminalOp所提供的sinkSupplier通过构造函数由外部赋值，通过Supplier接口的get()来交付给helper.wrapAndCopyInto(sinkSupplier.get(), spliterator)。

privatestaticfinalclassMatchOp<T>implementsTerminalOp<T,Boolean>{
finalSupplier>sinkSupplier;

@Override
publicBooleanevaluateSequential(PipelineHelperhelper,Spliteratorspliterator){
returnhelper.wrapAndCopyInto(sinkSupplier.get(),spliterator).getAndClearState();
}
}

由FindOps中实现TerminalOp所提供的与上述MatchOps是一致的

privatestaticfinalclassFindOp<T,O>implementsTerminalOp<T,O>{
finalSupplier>sinkSupplier;

@Override
publicOevaluateSequential(PipelineHelperhelper,Spliteratorspliterator){
Oresult=helper.wrapAndCopyInto(sinkSupplier.get(),spliterator).get();
returnresult!=null?result:emptyValue;
}
}

Collector

在Collector中有以下几个实现接口：

• Supplier：结果类型的提供器。
• BiConsumer：将元素放入结果的累加器。
• BinaryOperator：合并部分结果的组合器。
• Function：对结果类型转换为最终结果类型的转换器。
• Set：保存Collector特征的集合

并行流

前述都是基于串行流的讲解，其实并行流也是基于上述的helper.wrapAndCopyInto(op.sinkSupplier.get(), spliterator)这个方法上面做的一层基于ForkJoinTask多线程框架的封装。

ForkJoinTask

ForkJoin框架的思想就是分而治之，它将一个大任务切割为多个小任务这个过程称为fork，将每个任务的执行的结果进行汇总的过程称为join。ForkJoin框架相关的接口关系图如下（图是盗的）：

AbstractTask

类路径java.util.stream.AbstractTask，AbstractTask继承了在JUC中已经封装好的ForkJoinTask抽象子类java.util.concurrent.CountedCompleter。

此类基于CountedCompleter ，它是fork-join任务的一种形式，其中每个任务都有未完成子代的信号量计数，并且该任务隐式完成并在其最后一个子代完成时得到通知。 内部节点任务可能会覆盖CountedCompleter的onCompletion方法，以将子任务的结果合并到当前任务的结果中。

拆分和设置子任务链接是由内部节点的compute()完成的。 在叶节点的compute()时间，可以确保将为所有子代设置父代的子代相关字段（包括父代子代的同级链接）。

例如，执行减少任务的任务将覆盖doLeaf()以使用Spliterator对该叶节点的块执行减少Spliterator ，并覆盖onCompletion()以合并内部节点的子任务的结果：

@Override
protectedReduceTaskmakeChild(Spliteratorspliterator){
//返回一个ForkJoinTask任务
returnnewReduceTask<>(this,spliterator);
}

@Override
protectedSdoLeaf(){
//其他实现大同小异
returnhelper.wrapAndCopyInto(op.makeSink(),spliterator);
}

@Override
publicvoidonCompletion(CountedCompletercaller){
//非叶子节点进行结果组合
if(!isLeaf()){
SleftResult=leftChild.getLocalResult();
leftResult.combine(rightChild.getLocalResult());
setLocalResult(leftResult);
}
//GCspliterator,leftandrightchild
super.onCompletion(caller);
}

AbstractTask封装了分片任务的算法模板，通过是Spliterator的trySplit()方法来实现分片的细节，详细算法源码如下(类路径:java.util.stream.AbstractTask#compute)：

@Override
publicvoidcompute(){
//将当前这个spliterator作为右节点（此时为root节点）
Spliteratorrs=spliterator,ls;
//评估任务的大小
longsizeEstimate=rs.estimateSize();
//获取任务阈值
longsizeThreshold=getTargetSize(sizeEstimate);
booleanforkRight=false;
@SuppressWarnings("unchecked")Ktask=(K)this;
//细节不多赘述，下面我用图来讲解算法
/**
*根节点指定为：右边节点
*root
*split()
*leftright
*left.fork()
*split()
*lr
*rs=ls
*right.fork()
*split()
*lr
*l.fork()
*/
while(sizeEstimate>sizeThreshold&&(ls=rs.trySplit())!=null){
KleftChild,rightChild,taskToFork;
task.leftChild=leftChild=task.makeChild(ls);
task.rightChild=rightChild=task.makeChild(rs);
task.setPendingCount(1);
if(forkRight){
forkRight=false;
//左右节点切换进行fork和split
rs=ls;
task=leftChild;
taskToFork=rightChild;
}
else{
forkRight=true;
task=rightChild;
taskToFork=leftChild;
}
//fork任务加入队列中去
taskToFork.fork();
sizeEstimate=rs.estimateSize();
}
//将执行doLeaf底层就是单个串行流的操作
task.setLocalResult(task.doLeaf());
//将结果组合成一个最终结果
task.tryComplete();
}

AbstractTask执行与分片流程图如下：

到这里Stream流的相关知识介绍到这，这里附上一副总体图来加深下印象

审核编辑 ：李倩