Java 8 Stream流底層原理

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

函數(shù)式接口

初識(shí)lambda呢，函數(shù)式接口肯定是繞不過(guò)去的，函數(shù)式接口就是一個(gè)有且僅有一個(gè)抽象方法，但是可以有多個(gè)非抽象方法的接口。函數(shù)式接口可以被隱式轉(zhuǎn)換為lambda表達(dá)式。

@FunctionalInterface
publicinterfaceCloseable{

voidclose();
}

在java.util.function它包含了很多類，用來(lái)支持Java的函數(shù)式編程，該包中的函數(shù)式接口有：

操作

流程

Stream相關(guān)接口繼承圖：

Stream流水線組織結(jié)構(gòu)示意圖（圖是盜的）：

Collection

類路徑java.util.colltction

@Override
defaultSpliteratorspliterator(){
returnSpliterators.spliterator(this,0);
}
//常用Stream流轉(zhuǎn)換
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;

//此管道對(duì)象表示的中間操作的操作標(biāo)志。
protectedfinalintsourceOrOpFlags;

//管道中的下一個(gè)階段；如果這是最后一個(gè)階段，則為null。在鏈接到下一個(gè)管道時(shí)有效地結(jié)束。
privateAbstractPipelinenextStage;

//如果是順序的，則此管道對(duì)象與流源之間的中間操作數(shù)；如果是并行的，則為先前有狀態(tài)的中間操作數(shù)。在管道準(zhǔn)備進(jìn)行評(píng)估時(shí)有效。
privateintdepth;

//源和所有操作的組合源標(biāo)志和操作標(biāo)志，直到此流水線對(duì)象表示的操作為止（包括該流水線對(duì)象所代表的操作）。在管道準(zhǔn)備進(jìn)行評(píng)估時(shí)有效。
privateintcombinedFlags;

//源拆分器。僅對(duì)頭管道有效。如果管道使用非null值，那么在使用管道之前，sourceSupplier必須為null。在使用管道之后，如果非null，則將其設(shè)置為null。
privateSpliteratorsourceSpliterator;

//來(lái)源供應(yīng)商。僅對(duì)頭管道有效。如果非null，則在使用管道之前，sourceSpliterator必須為null。在使用管道之后，如果非null，則將其設(shè)置為null。
privateSupplier>sourceSupplier;

//如果已鏈接或使用此管道，則為True
privatebooleanlinkedOrConsumed;

//如果正在執(zhí)行任何有狀態(tài)操作，則為true；否則為true。僅對(duì)源階段有效。
privatebooleansourceAnyStateful;

privateRunnablesourceCloseAction;

//如果管道是并行的，則為true；否則，管道為順序的；否則為true。僅對(duì)源階段有效。
privatebooleanparallel;

ReferencePipeline

類路徑：java.util.stream.ReferencePipeline

filter

//java.util.stream.ReferencePipeline#filter
@Override
publicfinalStreamfilter(PredicatesuperP_OUT>predicate){
Objects.requireNonNull(predicate);
//返回一個(gè)匿名無(wú)狀態(tài)的管道
returnnewStatelessOp(this,StreamShape.REFERENCE,StreamOpFlag.NOT_SIZED){
//下游生產(chǎn)線所需要的回調(diào)接口
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
@Override
publicvoidbegin(longsize){
downstream.begin(-1);
}
//真正執(zhí)行操作的方法，依靠ChainedReference內(nèi)置ReferencePipeline引用下游的回調(diào)
@Override
publicvoidaccept(P_OUTu){
//只有滿足條件的元素才能被下游執(zhí)行
if(predicate.test(u))
downstream.accept(u);
}
};
}
};
}

map

//java.util.stream.ReferencePipeline#map
publicfinalStreammap(FunctionsuperP_OUT,?extendsR>mapper){
Objects.requireNonNull(mapper);
//返回一個(gè)匿名無(wú)狀態(tài)的管道
returnnewStatelessOp(this,StreamShape.REFERENCE,StreamOpFlag.NOT_SORTED|StreamOpFlag.NOT_DISTINCT){
//下游生產(chǎn)線所需要的回調(diào)接口
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
//真正執(zhí)行操作的方法，依靠ChainedReference內(nèi)置ReferencePipeline引用下游的回調(diào)
@Override
publicvoidaccept(P_OUTu){
//執(zhí)行轉(zhuǎn)換后提供給下游執(zhí)行
downstream.accept(mapper.apply(u));
}
};
}
};
}

flatMap

//java.util.stream.ReferencePipeline#flatMap
@Override
publicfinalStreamflatMap(FunctionsuperP_OUT,?extendsStream>mapper){
Objects.requireNonNull(mapper);
//返回一個(gè)匿名無(wú)狀態(tài)的管道
returnnewStatelessOp(this,StreamShape.REFERENCE,
StreamOpFlag.NOT_SORTED|StreamOpFlag.NOT_DISTINCT|StreamOpFlag.NOT_SIZED){
//下游生產(chǎn)線所需要的回調(diào)接口
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
@Override
publicvoidbegin(longsize){
downstream.begin(-1);
}
//真正執(zhí)行操作的方法，依靠ChainedReference內(nèi)置ReferencePipeline引用下游的回調(diào)
@Override
publicvoidaccept(P_OUTu){
try(Streamresult=mapper.apply(u)){
//劃分為多個(gè)流執(zhí)行下游（分流）
if(result!=null)
result.sequential().forEach(downstream);
}
}
};
}
};
}

peek

//java.util.stream.ReferencePipeline#peek
@Override
publicfinalStreampeek(ConsumersuperP_OUT>action){
Objects.requireNonNull(action);
//返回一個(gè)匿名無(wú)狀態(tài)的管道
returnnewStatelessOp(this,StreamShape.REFERENCE,0){
//下游生產(chǎn)線所需要的回調(diào)接口
@Override
SinkopWrapSink(intflags,Sinksink){
returnnewSink.ChainedReference(sink){
//真正執(zhí)行操作的方法，依靠ChainedReference內(nèi)置ReferencePipeline引用下游的回調(diào)
@Override
publicvoidaccept(P_OUTu){
//先執(zhí)行自定義方法，在執(zhí)行下游方法
action.accept(u);
downstream.accept(u);
}
};
}
};
}

sorted

@Override
publicfinalStreamsorted(){
//不提供Comparator，會(huì)使用元素自實(shí)現(xiàn)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);
//根據(jù)不同的flag進(jìn)行不同排序
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){
//返回一個(gè)匿名有狀態(tài)的管道
returnnewReferencePipeline.StatefulOp(upstream,StreamShape.REFERENCE,StreamOpFlag.IS_DISTINCT|StreamOpFlag.NOT_SIZED){

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

if(StreamOpFlag.DISTINCT.isKnown(flags)){
//已經(jīng)是去重過(guò)了
returnsink;
}elseif(StreamOpFlag.SORTED.isKnown(flags)){
//有序流
returnnewSink.ChainedReference(sink){
booleanseenNull;
//這個(gè)為先執(zhí)行的前序元素
TlastSeen;

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

@Override
publicvoidend(){
seenNull=false;
lastSeen=null;
downstream.end();
}
//這里通過(guò)有序的特性，前序元素與后序元素比較，如果相等則跳過(guò)執(zhí)行后序的元素
@Override
publicvoidaccept(Tt){
if(t==null){
//這里控制元素為null只有一個(gè)
if(!seenNull){
seenNull=true;
downstream.accept(lastSeen=null);
}
}elseif(lastSeen==null||!t.equals(lastSeen)){
//這里將前序元素賦值給lastSeen
downstream.accept(lastSeen=t);
}
}
};
}else{
//底層通過(guò)Set進(jìn)行去重，所以該元素需要重寫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(skip0)
thrownewIllegalArgumentException("Skipmustbenon-negative:"+skip);
//返回一個(gè)匿名有狀態(tài)的管道
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);
}
//自己實(shí)現(xiàn)真正操作的方法
@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));
}
//具有特定轉(zhuǎn)換的使用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;
}

//只是簡(jiǎn)單的消費(fèi)
@Override
publicvoidaccept(Tt){
consumer.accept(t);
}
}

Head

流的數(shù)據(jù)元的頭，類路徑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

無(wú)狀態(tài)的中間管道，類路徑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

有狀態(tài)的中間管道，類路徑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

管道流的結(jié)束操作，類路徑java.util.stream.TerminalOp

interfaceTerminalOp<E_IN,R>{

//獲取此操作的輸入類型的形狀
defaultStreamShapeinputShape(){returnStreamShape.REFERENCE;}

//獲取操作的流標(biāo)志。終端操作可以設(shè)置StreamOpFlag定義的流標(biāo)志的有限子集，并且這些標(biāo)志與管道的先前組合的流和中間操作標(biāo)志組合在一起。
defaultintgetOpFlags(){return0;}

//使用指定的PipelineHelper對(duì)操作執(zhí)行并行評(píng)估，該操作描述上游中間操作。
defaultRevaluateParallel(PipelineHelperhelper,Spliteratorspliterator){
if(Tripwire.ENABLED)
Tripwire.trip(getClass(),"{0}triggeringTerminalOp.evaluateParallelserialdefault");
returnevaluateSequential(helper,spliterator);
}

//使用指定的PipelineHelper對(duì)操作執(zhí)行順序評(píng)估，該操作描述上游中間操作。
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;
}

//通過(guò)匿名子類實(shí)現(xiàn)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;
}

//使用內(nèi)置的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;
}

//通過(guò)內(nèi)置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;
}

//自己實(shí)現(xiàn)了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>{
//開始遍歷元素之前調(diào)用該方法，通知Sink做好準(zhǔn)備。
defaultvoidbegin(longsize){}
//所有元素遍歷完成之后調(diào)用，通知Sink沒(méi)有更多的元素了。
defaultvoidend(){}
//是否可以結(jié)束操作，可以讓短路操作盡早結(jié)束。
defaultbooleancancellationRequested(){
returnfalse;
}
//遍歷元素時(shí)調(diào)用，接受一個(gè)待處理元素，并對(duì)元素進(jìn)行處理。Stage把自己包含的操作和回調(diào)方法封裝到該方法里，前一個(gè)Stage只需要調(diào)用當(dāng)前Stage.accept(Tt)方法就行了。
voidaccept(Tt);
}

這里Sink的子類實(shí)現(xiàn)中分為兩種：中間操作匿名實(shí)現(xiàn)ChainedReference和TerminalOp子類所提供的Sink。

ChainedReference

類路徑java.util.stream.Sink.ChainedReference，這里是中間操作的默認(rèn)模板父類

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

在上述的中間操作管道流中都是通過(guò)匿名類繼承ChainedReference實(shí)現(xiàn)onWrapSink(int, Sink)返回一個(gè)指定操作的Sink。

TerminalSink

這里為什么講提供呢？這是因?yàn)椴煌膶?shí)現(xiàn)TerminalOp的子類中在實(shí)現(xiàn)java.util.stream.TerminalOp#evaluateSequential中都是通過(guò)helper.wrapAndCopyInto(TerminalOp子類實(shí)現(xiàn)提供的Sink, spliterator)中通過(guò)參數(shù)傳遞的方式提供的，不同的子類傳遞的方式不一樣所以此處用了一個(gè)提供Sink

由ReduceOps中實(shí)現(xiàn)TerminalOp所提供的ReducingSink，它是由匿名類實(shí)現(xiàn)java.util.stream.ReduceOps.ReduceOp#makeSink來(lái)交付給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中實(shí)現(xiàn)TerminalOp所提供的是this，它的提供方式就是通過(guò)this交付給helper.wrapAndCopyInto(this, spliterator)。

//這里ForEachOp自己通過(guò)TerminalSink間接的實(shí)現(xiàn)了Sink
staticabstractclassForEachOp<T>implementsTerminalOp<T,Void>,TerminalSink<T,Void>{
@Override
publicVoidevaluateSequential(PipelineHelperhelper,Spliteratorspliterator){
returnhelper.wrapAndCopyInto(this,spliterator).get();
}
}

由MatchOps中實(shí)現(xiàn)TerminalOp所提供的sinkSupplier通過(guò)構(gòu)造函數(shù)由外部賦值，通過(guò)Supplier接口的get()來(lái)交付給helper.wrapAndCopyInto(sinkSupplier.get(), spliterator)。

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

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

由FindOps中實(shí)現(xiàn)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中有以下幾個(gè)實(shí)現(xiàn)接口：

• Supplier：結(jié)果類型的提供器。
• BiConsumer：將元素放入結(jié)果的累加器。
• BinaryOperator：合并部分結(jié)果的組合器。
• Function：對(duì)結(jié)果類型轉(zhuǎn)換為最終結(jié)果類型的轉(zhuǎn)換器。
• Set：保存Collector特征的集合

并行流

前述都是基于串行流的講解，其實(shí)并行流也是基于上述的helper.wrapAndCopyInto(op.sinkSupplier.get(), spliterator)這個(gè)方法上面做的一層基于ForkJoinTask多線程框架的封裝。

ForkJoinTask

ForkJoin框架的思想就是分而治之，它將一個(gè)大任務(wù)切割為多個(gè)小任務(wù)這個(gè)過(guò)程稱為fork，將每個(gè)任務(wù)的執(zhí)行的結(jié)果進(jìn)行匯總的過(guò)程稱為join。ForkJoin框架相關(guān)的接口關(guān)系圖如下（圖是盜的）：

AbstractTask

類路徑java.util.stream.AbstractTask，AbstractTask繼承了在JUC中已經(jīng)封裝好的ForkJoinTask抽象子類java.util.concurrent.CountedCompleter。

此類基于CountedCompleter ，它是fork-join任務(wù)的一種形式，其中每個(gè)任務(wù)都有未完成子代的信號(hào)量計(jì)數(shù)，并且該任務(wù)隱式完成并在其最后一個(gè)子代完成時(shí)得到通知。 內(nèi)部節(jié)點(diǎn)任務(wù)可能會(huì)覆蓋CountedCompleter的onCompletion方法，以將子任務(wù)的結(jié)果合并到當(dāng)前任務(wù)的結(jié)果中。

拆分和設(shè)置子任務(wù)鏈接是由內(nèi)部節(jié)點(diǎn)的compute()完成的。 在葉節(jié)點(diǎn)的compute()時(shí)間，可以確保將為所有子代設(shè)置父代的子代相關(guān)字段（包括父代子代的同級(jí)鏈接）。

例如，執(zhí)行減少任務(wù)的任務(wù)將覆蓋doLeaf()以使用Spliterator對(duì)該葉節(jié)點(diǎn)的塊執(zhí)行減少Spliterator ，并覆蓋onCompletion()以合并內(nèi)部節(jié)點(diǎn)的子任務(wù)的結(jié)果：

@Override
protectedReduceTaskmakeChild(Spliteratorspliterator){
//返回一個(gè)ForkJoinTask任務(wù)
returnnewReduceTask<>(this,spliterator);
}

@Override
protectedSdoLeaf(){
//其他實(shí)現(xiàn)大同小異
returnhelper.wrapAndCopyInto(op.makeSink(),spliterator);
}

@Override
publicvoidonCompletion(CountedCompletercaller){
//非葉子節(jié)點(diǎn)進(jìn)行結(jié)果組合
if(!isLeaf()){
SleftResult=leftChild.getLocalResult();
leftResult.combine(rightChild.getLocalResult());
setLocalResult(leftResult);
}
//GCspliterator,leftandrightchild
super.onCompletion(caller);
}

AbstractTask封裝了分片任務(wù)的算法模板，通過(guò)是Spliterator的trySplit()方法來(lái)實(shí)現(xiàn)分片的細(xì)節(jié)，詳細(xì)算法源碼如下(類路徑:java.util.stream.AbstractTask#compute)：

@Override
publicvoidcompute(){
//將當(dāng)前這個(gè)spliterator作為右節(jié)點(diǎn)（此時(shí)為root節(jié)點(diǎn)）
Spliteratorrs=spliterator,ls;
//評(píng)估任務(wù)的大小
longsizeEstimate=rs.estimateSize();
//獲取任務(wù)閾值
longsizeThreshold=getTargetSize(sizeEstimate);
booleanforkRight=false;
@SuppressWarnings("unchecked")Ktask=(K)this;
//細(xì)節(jié)不多贅述，下面我用圖來(lái)講解算法
/**
*根節(jié)點(diǎn)指定為：右邊節(jié)點(diǎn)
*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;
//左右節(jié)點(diǎn)切換進(jìn)行fork和split
rs=ls;
task=leftChild;
taskToFork=rightChild;
}
else{
forkRight=true;
task=rightChild;
taskToFork=leftChild;
}
//fork任務(wù)加入隊(duì)列中去
taskToFork.fork();
sizeEstimate=rs.estimateSize();
}
//將執(zhí)行doLeaf底層就是單個(gè)串行流的操作
task.setLocalResult(task.doLeaf());
//將結(jié)果組合成一個(gè)最終結(jié)果
task.tryComplete();
}

AbstractTask執(zhí)行與分片流程圖如下：

到這里Stream流的相關(guān)知識(shí)介紹到這，這里附上一副總體圖來(lái)加深下印象

審核編輯 ：李倩