丨stream如何提高遍歷集合效率

例如，系統(tǒng)中的訂單表，常常使用用戶ID的Hash值來實(shí)現(xiàn)分表分庫，這樣是為了減例如，當(dāng)我們需要查詢出過濾條件下的所有訂單，并按照訂單的某個條件進(jìn)行排序，單個數(shù)據(jù)源查詢出來的數(shù)據(jù)是可以按照某個條件進(jìn)行排序的，但多個數(shù)據(jù)源查詢出來已經(jīng)排序好的數(shù)據(jù)，并不代表合并后是正確的排序，所以我們需要在應(yīng)用層對合并數(shù)據(jù)集合重新進(jìn)行排序。在Java8之前，我們通常是通過for循環(huán)或者Itor迭代來重新排序合并數(shù)據(jù)，又或者通過重新定義Collections.sorts的Comparator方法來實(shí)現(xiàn)，這兩種方式對于大數(shù)據(jù)量系Java8中添加了一個新的接口類Stream，他和我們之前接觸的字節(jié)流概念不太一樣，Java8集合中的Stream相當(dāng)于高級版的I tor，他可以通過Lambda表達(dá)式對集合進(jìn)行各種非常便利、高效的聚合操作（AggregateOperation），或者大批量數(shù)據(jù)操作(BulkDataOperation)。Stream的聚合操作與數(shù)據(jù)庫SQL的聚合操作sorted、filter、map等類似。我們在應(yīng)用層就可以高效地實(shí)現(xiàn)類似數(shù)據(jù)庫SQL的聚合操作了，而在數(shù)據(jù)操作方面，Stream不僅可接下來我們就用一個簡單的例子來體驗(yàn)下Stream的簡潔與強(qiáng)大這個Demo的需求是過濾分組一所中學(xué)里身高在160cm以上的男女同學(xué)，我們先用傳統(tǒng)代Map<String,List<Student>>stuMap=newHashMap<String,for(Studentstu:studentsList)if(stu.getHeight()>160){//如果身高大于if ())==null){//該還沒分List<Student>list=newArrayList<Student>();//新建該學(xué)生的list.add(stu);//將學(xué)789

stuMap.put(stu.get(),list);//將列表放到map}else{//該分類已存stuMap.get(stu.get()).add(stu);//該分類已存在，則直接放進(jìn)去即}我們再使用Java8中的StreamAPI進(jìn)行實(shí)現(xiàn)串行實(shí)代1Map<String,List<Student>>stuMap=stuList.stream().filter((Students)->并行實(shí)代1Map<String,List<Student>>stuMap=stuList.parallelStream().filter((Students)->通過上面兩個簡單的例子，我們可以發(fā)現(xiàn)，Stream合Lambda達(dá)式實(shí)現(xiàn)遍歷篩選功上面我們初步了解了Java8的StreamAPI，那Stream如何做到優(yōu)化迭代的呢？并行又是如何實(shí)現(xiàn)的？下面我們就透過Stream源碼剖析Stream的實(shí)現(xiàn)原理。Stream操作在了解Stream實(shí)現(xiàn)原理之前，我們先來了解下Stream操作分類，因?yàn)樗牟僮鞣诸悓tream中的操作分為兩大類：中間操作（Intermediateoperations）和終結(jié)操（Terminaloperations）。中間操作只對操作進(jìn)行了記錄，即只會返回一個流，不會進(jìn)行 不受之前元素的影響，后者是指該操作只有拿到所有元后才能繼續(xù)下去。終結(jié)操作又可以分為短路（Shor-circiing）與非短路（Ushor-irciing）操作，前者是指遇到某些符合條件的元素就可以得到最終結(jié)果，后者是指必須處理完所有元素才能得到最終結(jié)果。操作分類詳情如下圖所示：理管道（Pipeline），實(shí)現(xiàn)了Stream的高效。Stream源碼在了解Stream如何工作之前，我們先來了解下Stream包是由哪些主要結(jié)構(gòu)類組合的，各個類的職責(zé)是什么。參照下BaseStream和Stream為最頂端的接口類。BaseStream主要定義了流的基本接口方法， tor、isParallel等；Stream則定義了一些流的常用操作方法，例如，map、filter等。 是一個結(jié)構(gòu)類，他通過定義內(nèi)部類組裝了流。他定義了 essOp、StatefulOp三個內(nèi)部類，實(shí)現(xiàn)了BaseStream與Stream的接口方Sink接口是定義每個Stream操作之間關(guān)系的協(xié)議，他包含begin()、end()、 t()四個方法。ReferencePipeline最終會將整個Stream流操作組裝成一個調(diào)用鏈，而這條調(diào)用鏈上的各個Stream操作的上下關(guān)系就是通過SinkStream操作我們知道，一個Stream的各個操作是由處理管道組裝，并統(tǒng)一完成數(shù)據(jù)處理的。在中每次的中斷操作會以使用階段（Stage）命名管道結(jié)構(gòu)通常是由ReferencePipeline類實(shí)現(xiàn)的，前面講解Stream包結(jié)構(gòu)時，我提 essOp、StatefulOp三種內(nèi)部類Head類主要用來定義數(shù)據(jù)源操作，在我們初次調(diào)用names.stream()方法時，會初次加Head對象，此時為加載數(shù)據(jù)源操作；接著加載的是中間操作，分別為無狀態(tài)中間操 essOp象和有狀態(tài)StatefulOp象，此時的Stage沒有執(zhí)行，而是Pipeline生成了一個中間操作Stage鏈表；當(dāng)我們調(diào)用終結(jié)操作時，會生成一個最終的Stage，通過這個Stage觸發(fā)之前的中間操作，從最后一個Stage開始，遞歸產(chǎn)生一個Sink鏈。如下圖所示：下面我們再通過一個例子來感受下Stream操作分類是如何實(shí)現(xiàn)高效迭代大數(shù)據(jù)集合代1List<String>names=Arrays.asList("張三","李四","王老五","李三","劉老四2StringmaxLenStartWithZ=.filter(name->name.startsWith("張這個例子的需求是查找出一個長度最長，并且以張為姓氏的名字。從代碼角度來看，你可能會認(rèn)為是這樣的操作流程：首先遍歷一次集合，得到以“張”開頭的所有名字；然后遍歷一次filer得到的集合，將名字轉(zhuǎn)換成數(shù)字長度；最后再從長度集合中找到最長的那個名字并且返回。首先因?yàn)閚amesArrayList合names.stream法將會調(diào)用集合類基礎(chǔ)接口Collection的Stream方法：代defaultStream<E>stream()returnStreamSupport.stream(splitor(), 然后，Stream方法就會調(diào)用StreamSupport類的Stream方法，方法中初始化了一ReferencePipelineHead部類對象代publicstatic<T>Stream<T> tor<T> tor,booleanparallel)returnnewReferencePipeline.Head<>(spli 再調(diào)用filter和map方法，這兩個方法都是無狀態(tài)的中間操作，所以執(zhí)行filter和操作時，并沒有進(jìn)行任何的操作，而是分別創(chuàng)建了一個Stage來標(biāo)識用戶的每一次操而通常情況下Stream的操作又需要一個回調(diào)函數(shù)，所以一個完整的Stage是由數(shù)據(jù)來源、操作、回調(diào)函數(shù)組成的三元組來表示。如下圖所示，分別是ReferencePipeline的filter方法和map方法：代12publicfinalStream<P_OUT>filter(Predicate<?superP_OUT>predicate)34returnnewStaessOp<P_OUT,P_OUT>(this,5StreamOpFlag.NOT_SIZED)67Sink<P_OUT>opWrapSink(intflags,Sink<P_OUT>sink)8returnnewSink.ChainedReference<P_OUT,P_OUT>(sink)9publicvoidbegin(longsize)} publicvoidaccept(P_OUTu) if

代publicfinal<R>Stream<R>map(Function<?superP_OUT,?extendsR>mapper)returnnewStaessOp<P_OUT,R>(this,StreamOpFlag.NOT_SORTED|Sink<P_OUT>opWrapSink(intflags,Sink<R>sink)returnnewSink.ChainedReference<P_OUT,R>(sink)publicvoidaccept(P_OUTu)}}}new essOp將會調(diào)用父 Pipeline的構(gòu)造函數(shù)，這個構(gòu)造函數(shù)將前后Stage系起來，生成一個Stage表代 Pipeline<?,E_IN,?>previousStage,intopFlags)ifthrownewpreviousStage.linkedOrConsumed=previousStage.nextStage=this;//將當(dāng)前的stage的next指針指向之前的6this.previousStage=previousStage;//賦值當(dāng)前stage當(dāng)全局變量this.sourceOrOpFlags=opFlags&binedFlags this.sourceStage=ifsourceStage.sourceAnyStateful=

this.depth=previousStage.depth+因?yàn)樵趧?chuàng)建每一個Stage時，都會包含一個opWrapSink()方法，該方把一個操作的具體實(shí)現(xiàn)封裝在Sink類中，Sink采用（處理->轉(zhuǎn)發(fā)）的模式來疊加操作。當(dāng)執(zhí)行max法時，會ReferencePipelinemax法，此時由于max法是終結(jié)操作，所以會創(chuàng)建一個TerminalOp操作，同時創(chuàng)建一個ReducingSink，并且將操作封裝在Sink類中。代publicfinalOptional<P_OUT>max(Comparator<?superP_OUT>comparator)return 最后，調(diào) Pipeline的wrapSink方法，該方調(diào)用opWrapSink生成一Sink表，Sink表中的每一個Sink裝了一個操作的具體實(shí)現(xiàn)代final<P_IN>Sink<P_IN>wrapSink(Sink<E_OUT>sink)5for( Pipeline Pipeline.this;sink= binedFlags, return(Sink<P_IN>) Sink表生成完成后，Stream始執(zhí)行，通過splitor代集合，執(zhí)行Sink鏈表代final<P_IN>voidcopyInto(Sink<P_IN>wrappedSink, tor<P_IN>splitor)4if(!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) elsecopyIntoWithCancel(wrappedSink,spli Java8中的Spli tor的forEachRemaining會迭代集合，每迭代一次，都會執(zhí)行一次filter操作，如果filter操作通過，就會觸發(fā)map操作，然后將結(jié)果放入到臨時數(shù)組object中，再進(jìn)行下一次的迭代。完成中間操作后，就會觸發(fā)終結(jié)操作max。這就是串行處理方式了，那么Stream的另一種處理數(shù)據(jù)的方式又是怎么操作的Stream并行Stream理數(shù)據(jù)的方式有兩種，串行處理和并行處理。要實(shí)現(xiàn)并行處理，我們只需要在例子的代碼中新增一個Parallel()方法，代碼如下所示：代1List<String>names=Arrays.asList("","","","","劉老四2StringmaxLenStartWithZ=.filter(name->name.startsWith("張Stream的并行處理在執(zhí)行終結(jié)操作之前，跟串行處理的實(shí)現(xiàn)是一樣的。而在調(diào)用終結(jié)方法之后，實(shí)現(xiàn)的方式就有點(diǎn)不太一樣，會調(diào)用TerminalOp的evaluateParallel方法進(jìn)行行處理代final<R>Revaluate(TerminalOp<E_OUT,R>terminalOp)assertgetOutputShape()==ifthrownewlinkedOrConsumed=6return?terminalOp.evaluateParallel(this,sourceSpli:terminalOp.evaluateSequential(this,sourceSpli 這里的并行處理指的是，Stream結(jié)合了ForkJoin框架，對Stream處理進(jìn)行了分片 tor中的estimateSize方估算出分片的數(shù)據(jù)量ForkJoin框架和估算算法，在這里我就不具體講解了，如果感，你可以深入源碼分析通過預(yù)估的數(shù)據(jù)量獲取最小處理單元的閥值，如果當(dāng)前分片大小大于最小處理單元的閥值，就繼續(xù)切分集合。每個分片將會生成一個Sink鏈表，當(dāng)所有的分片操作完成后，F(xiàn)orkJoin框架將會合并分片任何結(jié)果集。合理使用看到這里，你應(yīng)該對StreamAPI是如何優(yōu)化集合遍歷有個清晰的認(rèn)知了。StreamAPI用起來簡潔，還能并行處理，那是不是使用StreamAPI，系統(tǒng)性能就更好呢？通過一組測對常規(guī)的迭代、Stream串行迭代以及Stream并行迭代進(jìn)行性能測試對比，迭代循多核CPU服務(wù)器配置環(huán)境下，對比長度100的int數(shù)組的性能多核CPU服務(wù)器配置環(huán)境下，對比長度1.00E+8的int數(shù)組的性能多核CPU服務(wù)器配置環(huán)境下，對比長度1.00E+8對象數(shù)組過濾分組的性能；單核CPU服務(wù)器配置環(huán)境下，對比長度1.00E+8對象數(shù)組過濾分組的性能。 常規(guī)的迭代<Stream并行迭代<Stream串行迭代Stream并行迭代<常規(guī)的迭代<Stream串行迭代Stream并行迭代<常規(guī)的迭代<Stream串行迭代常規(guī)的迭代<Stream串行迭代<Stream并行迭代而更好；在單核CPU服務(wù)器配置環(huán)境中，也是常規(guī)迭代方式更有優(yōu)勢；而在大數(shù)據(jù)循環(huán)迭代中，如果服務(wù)器是多核CPU的情況下，Stream的并行迭代優(yōu)勢明顯。所以我們在平時處理大數(shù)據(jù)的集合時，應(yīng)該盡量考慮將應(yīng)用部署在多核CPU環(huán)境下，并且使用Stream的用事實(shí)說話，我們看到其實(shí)使用Stream必可以使系統(tǒng)性能更佳，還是要結(jié)合應(yīng)用場景進(jìn)行選擇，也就是合理地使用Stream?？v觀Stream設(shè)計實(shí)現(xiàn)，非常值得我們學(xué)習(xí)。從大的設(shè)計方向上來說，Stream從小的分類方向上來說，Stream將遍歷元素的操作和對元素的計算分為中間操作和終結(jié)操作，而中間操作又根據(jù)元間狀態(tài)有無干擾分為有狀態(tài)和無狀態(tài)操作，實(shí)現(xiàn)了鏈結(jié)構(gòu)中的在串行處理操作中，Stream在執(zhí)行每一步中間操作時，并不會做實(shí)際的數(shù)據(jù)操作處理，而是將這些中間操作串聯(lián)起來，最終由終結(jié)操作觸發(fā)，生成一個數(shù)據(jù)處理鏈表，通過Java8