日志分析并行分解設(shè)計(jì)與實(shí)現(xiàn)

任務(wù)就是要完成一個(gè)日志分析應(yīng)用。需求沒有很明確，只是要有這么一個(gè)東西能夠滿足分析收集后的日志，將分析后的原始數(shù)據(jù)入庫(kù)，作為后期分析和統(tǒng)計(jì)使用。在動(dòng)手做之前，我還是給這個(gè)應(yīng)用作了最基本的需求定義：靈活配置（輸入源，輸出目標(biāo)，分析器的實(shí)現(xiàn)等），高效（并行任務(wù)分解）。就這兩點(diǎn)能夠做到，那么將來需求如何變化都可以適應(yīng)。Tiger的Concurrent包是滿足后面那項(xiàng)最好的實(shí)現(xiàn)，里面的線程池，異步服務(wù)調(diào)用，并發(fā)控制都能夠極好的完成并行任務(wù)分解的工作。J2SE7的Concurrent包中將會(huì)增加fork-join風(fēng)格的并行分解庫(kù)，其實(shí)這個(gè)是更細(xì)粒度的任務(wù)分解，同時(shí)能夠在當(dāng)前多CPU的情況下提高執(zhí)行效率，充分利用CPU的一種實(shí)現(xiàn)。

背景:由于服務(wù)路由應(yīng)用訪問量十分大，即時(shí)的將訪問記錄入庫(kù)對(duì)于路由應(yīng)用本身以及數(shù)據(jù)庫(kù)來說無疑都會(huì)產(chǎn)生很大的壓力和影響。因此考慮首先將訪問信息通過log4j記錄在本地（當(dāng)然自己需要定制一下 Log4j的Appender和Filter），然后通過服務(wù)器的定時(shí)任務(wù)腳本來將日志集中到日志分析應(yīng)用所在的機(jī)器上（這里通過配置可以決定日志是根據(jù)什么時(shí)間間隔來產(chǎn)生新文件）。日志分析應(yīng)用就比較單純的讀取日志，分析日將即時(shí)統(tǒng)計(jì)信息存入到集中式緩存志，輸出分析結(jié)果（包括寫入數(shù)據(jù)庫(kù)或者是■1志，輸出分析結(jié)果（包括寫入數(shù)據(jù)庫(kù)或者是■1Memcached中）。網(wǎng)絡(luò)結(jié)構(gòu)圖如下：

Concurrent概述:Concurrent概述:卷廷虹蘭q看Java的Doc很容易就理解了Concurrent,這里我只是大致的說一下幾個(gè)自己在應(yīng)用中使用的接口：BlockingQueue<E>:看看名字就知道了，阻塞式隊(duì)列，可以設(shè)置大小。適合于生產(chǎn)者和消費(fèi)者模式，生產(chǎn)者在隊(duì)列滿時(shí)阻塞，消費(fèi)者在隊(duì)列空時(shí)阻塞。在日志分析應(yīng)用開發(fā)中被用于分析任務(wù)（生產(chǎn)者）和輸出任務(wù)（消費(fèi)者）之間的分析結(jié)果存儲(chǔ)通道。Callable<V>:任何需要執(zhí)行的任務(wù)都可以定義成Callable，類似于線程的Runnable接口，可以被ServiceExecutor指派給內(nèi)部的線程異步執(zhí)行，并且返回對(duì)象或者拋出異常。在日志分析應(yīng)用開發(fā)中，非定時(shí)性的任務(wù)都定義成為此類型。ConcurrentMap<K,V>:這個(gè)以前常常使用，效率要遠(yuǎn)遠(yuǎn)Collections.synchronizedCollection和synchronizedo后面還會(huì)提到實(shí)踐中的幾個(gè)實(shí)用的技巧來防止在高并發(fā)的情況下出現(xiàn)問題。在日志分析應(yīng)用中，此類型的Map作為保存日志文件分析狀態(tài)的緩存（日志文件分為兩種狀態(tài)：分析中，分析結(jié)束。如果不存在于Map中就認(rèn)為尚未分析，那么將其納入Map然后啟動(dòng)分析處理線程工作，如果存在于Map中標(biāo)示為分析中，那么將不會(huì)再分析此文件，如果分析結(jié)束并且被輸出，將會(huì)標(biāo)示此文件分析結(jié)束，異步清理線程將會(huì)定時(shí)根據(jù)策略刪除或移動(dòng)文件）。ExecutorService:內(nèi)置線程池，異步執(zhí)行指派任務(wù)，并可以根據(jù)返回的Future來跟蹤執(zhí)行情況。在日志分析應(yīng)用開發(fā)中，被用于非定時(shí)性任務(wù)執(zhí)行。ScheduledExecutorService:內(nèi)置線程池，定時(shí)異步執(zhí)行指派任務(wù)，并可以根據(jù)返回的Future來跟蹤執(zhí)行情況。在日志分析應(yīng)用開發(fā)中，被用于定時(shí)性任務(wù)執(zhí)行。以上就是被使用到的接口，具體實(shí)現(xiàn)策略配置就不在此贅述了。整體結(jié)構(gòu)設(shè)計(jì):整體設(shè)計(jì)還是基于開始設(shè)定的兩個(gè)原則:靈活配置,高效性（任務(wù)分解,并行流水線執(zhí)1=-靈活配置,高效性（任務(wù)分解,并行流水線執(zhí)1=-行）。說到任務(wù)分解又會(huì)想起讀書時(shí)候的離散數(shù)學(xué)中關(guān)鍵路徑等等。任務(wù)分解還是要根據(jù)具體情況來分析和設(shè)計(jì)，不然并行不但不會(huì)提高效率，反而還降低了處理效率。就日志分析來看，主要的處理過程可以分成這么幾個(gè)任務(wù)：檢查日志來源目錄，鎖定需要分析的文件。（執(zhí)行需要時(shí)間很短，可通過定時(shí)間隔執(zhí)行）。分析已經(jīng)被鎖定的日志文件，產(chǎn)生分析結(jié)果。（執(zhí)行需要時(shí)間根據(jù)日志文件大小來決定，因此需要線程異步執(zhí)行，結(jié)果根據(jù)設(shè)定拆分成細(xì)粒度包，降低輸出線程等待時(shí)間）。檢查分析結(jié)果隊(duì)列。（執(zhí)行需要時(shí)間很短，當(dāng)前是配置了SingleThreadExecutor來執(zhí)行檢查阻塞隊(duì)列的工作，同時(shí)獲取到分析結(jié)果包以后立刻創(chuàng)建線程來執(zhí)行輸出任務(wù)）輸出分析結(jié)果，如果輸出成功，將分析過的日志文件在日志文件狀態(tài)緩存中的狀態(tài)更新為已分析。（執(zhí)行時(shí)間根據(jù)輸出情況來定，當(dāng)前實(shí)現(xiàn)的是批量輸出到數(shù)據(jù)庫(kù)中，根據(jù)配置來批量提交入庫(kù)，后續(xù)還會(huì)考慮實(shí)時(shí)統(tǒng)計(jì)到集中式Cache作為監(jiān)控使用）。清理分析日志文件。（執(zhí)行時(shí)間較短，設(shè)定了定時(shí)線程池執(zhí)行清理任務(wù)，根據(jù)策略配置來執(zhí)行清理和移動(dòng)文件任務(wù)，并且清除在日志文件狀態(tài)緩存中的信息）根據(jù)上面的分解可以看到，其實(shí)在單線程工作的過程中，容易造成阻塞而影響性能的主要是讀取，分析和寫出這三個(gè)過程的協(xié)調(diào)，一個(gè)一個(gè)讀取分析和寫出，性能一定低于讀取和分析并行工作，而分析完畢才寫出，性能一定低于分析部分，寫出部分。同時(shí)由于細(xì)分各個(gè)任務(wù)，因此任務(wù)與任務(wù)之間的耦合度降低，可以運(yùn)行期獲取具體的任務(wù)實(shí)現(xiàn)配置，達(dá)到靈活配置的目的。F面就具體的看看整個(gè)流程，以及其中的一些細(xì)節(jié)的說明，這里根據(jù)下圖中的序號(hào)來逐一描述:配置了ScheduleExecutor來檢查日志所屬目錄中的日志文件，Executor的線程池大小以及檢查時(shí)間間隔都根據(jù)配置來設(shè)定。Tip:定時(shí)任務(wù)可以設(shè)置delay時(shí)間，那么可以根據(jù)你的任務(wù)數(shù)量以及時(shí)間間隔來設(shè)定每一個(gè)任務(wù)的delay時(shí)間，均勻的將這些任務(wù)分布，提高效率。當(dāng)ReadSchedule被執(zhí)行時(shí)，將會(huì)去檢查AnalysisLogFileStateConcurrentCache（也就是上面提到的ConcurrentMap）中是否存在此文件，如果不存在證明尚未分析，需要將其置入Cache，如果已經(jīng)存在就去查詢其他文件。Tip:這里用了一點(diǎn)小技巧，通常我們對(duì)于此類操作應(yīng)該做兩部分工作，get然后再put，但是這樣可能就會(huì)在高并發(fā)的情況下出現(xiàn)問題，因?yàn)檫@兩個(gè)操作不是一個(gè)原子操作。ConcurrentMap提供了putIfAbsent操作，這個(gè)操作意思就是說如果需要put的key沒有存在于Map中，那么將會(huì)把key,value存入，并且返回null,如果已經(jīng)存在了key那么就返回key在map已(resources.putIfAbsent(filename,經(jīng)對(duì)應(yīng)的值過if經(jīng)對(duì)應(yīng)的值過ifConstants.FILE_STATUS_ANALYSISING)==null)就可以把兩個(gè)操作合并成為一個(gè)操作。日志讀取的工作線程完成鎖定文件以后，就將后續(xù)的工作交給LogAnalysisServiceExecutor來創(chuàng)建分析任務(wù)異步執(zhí)行分析操作，日志讀取工作線程任務(wù)就此完成。LogAnalysisSchedule是運(yùn)行期裝載具體的接口實(shí)現(xiàn)類（采用的就是類似于JAXP等框架使用的META-INF/services來讀取工廠類，載入接口實(shí)現(xiàn)）。AnalysisSchedule執(zhí)行的主要任務(wù)就是分析文件，并且根據(jù)配置將分析結(jié)果拆分并串行的置入到BlockQueue中，提供給輸出線程使用。Receiver主要工作就是守候著BlockQueue，當(dāng)有數(shù)據(jù)結(jié)果產(chǎn)生就創(chuàng)建WriteSchedule來異步執(zhí)行輸出。■1LogWriterServiceExecutor根據(jù)配置來決定內(nèi)置線程池大小，同時(shí)在Receiver獲取到數(shù)據(jù)包時(shí)產(chǎn)生WriteSchedule來異步執(zhí)行輸出工作?！?WriteSchedule和AnalysisSchedule一樣可以運(yùn)行期裝載接口實(shí)現(xiàn)類，這樣提供了靈活的輸出策略配置。Tips:在數(shù)據(jù)庫(kù)輸出的時(shí)候需要配置批量提交記錄最大數(shù)，分批提交提高性能，也防止過大結(jié)果集批量提交問題?！?=J最■1寫出完成以后需要更新鎖定文件的狀態(tài)，標(biāo)示成為巳經(jīng)分析成功。這里還遺留一點(diǎn)問題，在一個(gè)日志文件分包的過程中每一個(gè)包都回記錄隸屬于哪一個(gè)分析文件，文件的最后一個(gè)數(shù)據(jù)包將會(huì)被標(biāo)示。在輸出成功以后會(huì)去檢查哪些包是文件最后數(shù)據(jù)包，更新此文件為已分析成功，如果出現(xiàn)異常，那么將會(huì)把這些文件狀態(tài)清除，接受下一次的重新分析。這里一個(gè)文件部分包提交暫時(shí)沒有做到事務(wù)一致，如果出現(xiàn)部分成功可能會(huì)重復(fù)分析和記錄?！?=J最■1最后就是CleanSchedule被定時(shí)執(zhí)行，根據(jù)策略來刪除或者移動(dòng)已經(jīng)被分析過的文件。Tips:ScheduledExecutorService內(nèi)部可以配置線程池，當(dāng)執(zhí)行定時(shí)任務(wù)比較耗時(shí)，線程池中的線程都被占用的情況下，定時(shí)任務(wù)將不會(huì)準(zhǔn)確的按時(shí)執(zhí)行，因此設(shè)計(jì)過程中需要注意的是，定時(shí)任務(wù)一般是簡(jiǎn)短的工作任務(wù)，如果比較耗時(shí)，那么應(yīng)該結(jié)合ScheduledExecutorService 和ExecutorService，定時(shí)任務(wù)完成必要工作以后將耗時(shí)工作轉(zhuǎn)交給ExecutorService創(chuàng)建的即時(shí)執(zhí)行異步線程去處理，保證ScheduleExecutor正常工作。?entity

SfirvicaAcpassLog4\i:—ndLozCl^anYork紂fool￥￥￥8:updJKjrtalwclooflcstjt^?entity

SfirvicaAcpassLog4\i:—ndLozCl^anYork紂fool￥￥￥8:updJKjrtalwclooflcstjt^Njme: L?9Anal'/zorAuthor: grugnohuVersion:1.0Ciedtcd2008+222:38;19Updates:230942245:58:134：”dwe歸xdtcsuftpwggot>~]]—~一—L—一―—1IBlockQueueo￡IIAnalyzizdLogIResult'I°lEO!TOC\o"1-5"\h\zI IIoiI II IIO；I II I 6： 9*t resultcdanah.izer/LogAnalyzerConfig-recordMaxCount:ini=100LogAnalyzerConfig-recordMaxCount:ini=100-readChecMntervalini=EO-readerThreadCount:int=10='.''jrrtQrThrejdC:-uni:int=8-cleanerThreadCount:int=2-cleanerChecklntemal:int=10K60-r^acurce□li□eSize:int=1000-3nslyssThreadCount:int=20-logFileDir:Stiing?-backupDir:StiingdqI?stGft.ftQrArijlyftf：boolsan=truq-dbcommitSi.ilaxCount:int=20i—■IT比口AnalyzerLog二LngFm&ry.gutL...helper:LogAnalyzerHeIper<T>十runp:'.'Old+startO：mid+stopQ:void+jTwin〔String[|):mid十getRecordMaz：CountQ十getRecordMaz：CountQ:ini+se1RemrdMounitint):void+qetRejdChedilnterval|):int+selReadChecknterval(int):void+qstRcjdc：rThra□dCuniQ:int+se1ReaderFhreadCount(int):void+getMVriterThreadCouniQ:int十selWrllerThreadCount(ln1):uoid+aetCleanerThieadCouniO:int+seiCleaneiThreadCounifinf):void+getCleanerChecklnterralO:ini+ aiCIgsngiCh4cklntarv3l(int):void+getResourceQueuesizeQ:int+se1ResourceQueueSize(int):void十getAnalysisThreadCountQ:In1+seiAnalysisThreadCounitinf):noid+getLogFileDirO:StringQ+selLogFileDir(String0):void+ifDQlQl4Atl：QrAn^lyfi￡Q:b*:>olQjn+se1DeIateAfter^nalysis^booIean):必id+getBackupDirQ:Siring十se1Backupl>Ir(3trlng):/old+aetDbi：：ommitMaxCouniO'int+seiDbcommitMaxCoanoint):voidi—■ITLogAnalj/zerHelperLnwgq「L「g二LnwFqm「ygTt_...readers:ScheduIedExecutorService□n EM^outar￡?rvio?consunier:ExecutorServiceijijriters:ExecutorServiceScheduIedEkecutorServiceijijriterResourceQueue:LinkedBlockingQueue<Anesult<T：=■：二resouroes:ConcurrentHashMap<Siring,lnteger>config:LogAnaiyzerConfig~rejdqrH□ndlqr:Schc：duIadFutufq^7^-i；[])~consumerHandler:Future<?>~cleanHandler:ScheduledFuture<?>([])+getConfiaO:LagAnalyzerConfig+startAnaysisQ:void+configWoikErThreadPoolQ：void+^topAnayfi^:void上面的類圖中主要描述的就是日志分析應(yīng)用的三個(gè)主類:類似于控制臺(tái)的LogAnalyzer,具體內(nèi)部資源管理類，配置類。（T表示采用泛型）Lorqs:Lop=SMictory.SWtL..?rasultAnjl5XkR?ult<T>CD)Lorqs:Lop=SMictory.SWtL..?rasultAnjl5XkR?ult<T>CD)/conflj.L&aAnj|yz<jfC0nflQ> C0ncuii4ntHj^iM^pc$1iing.lnt?0ai>gSTgO.ESuIgpgwrO:void?夕83*V2l網(wǎng)MtRcwlttAnji-fSttRasyiVT^:void?-?t?；onnc(L^aAftjivosjconiitf):void，?tRewurcei：Conc<jner*tHxriMjp*smnD.Inte-sep):voidmjggNduieLogerLopFdctorF.gtL.sHe，ResouK>!QueueLir^rdBlocfcnpQueue<Andlpi!fte5uHL&qflarLor—Logr?<￡^r/L&qflarLor—Logr?<￡^r/?ItftiRcecouicaOuouwLir^adBlodjngOuduftcAnaljdiRwulkT??iMOuicacConcuiiantHacftMap^Slikig.lrilfc^i?afNor;:E>：KWo%m心sMg.Lo護(hù)心yzgonfigmF■!?:Sbin^configiLo0An4l^>z?rCordi9xeffJitlarftsaourceQu?u<Linh?d0lad?in9Qu*u?<*?n4lyataR??ult<void9?iTyp?(|SchadulaT-^pax*iR?xfile^SliinQ^void?un0:voidp*o:?xT|^fnrty,Ar^'yuM<T,?)?p；d4arMlyguoldwtCoMi^to-jAnjt|i6iC uoldC1?wSciwdul? /C/c力merSoheaJeC1?wSciwdul? /C/c力merSoheaJeTOC\o"1-5"\h\zLoggeiLoa-LonF4dorya?1L... ；i?irt?iRaaoufc?Ouau?Linh?<18lad?ir>0Qw?u?<An4lvxaft??uU*Tj*'?laiouicM：CortojrrardH?>hlil4p<^inn9，lnU0er> *osnlip:Lo^An4^xetCordiQ :,^vlT^pe^):SdieduUT|p? ???ffti；Urf^??ourc?Ou?u?^Ljnk?dBlod?in90u?u?^An4lyiia^??jlktfT>?>):roid￡?1RftC0uittfi(Concuii6ntHafihMapcSIdng.InWQAr?),成iur<?>old ::“id ,'$SConrt^Li 。福。).?，況d :D*C3ugrte ，」leLogorLoo-Loafa?on/.5CD.“EOMto心go:IMonn&tL^OJO，piocoZnjryfl^ervlt*Prfonit0!Lo^>(D:voidQMM^mw兒og?Q?iMohttertopPao5*Wonrt0rt_opD3C<IM0nito?Lo^pJ0)LogsLog=LoqF""qSL..、,wtrt4rR??urccQuouo:LmKQdSloBmqQuQuX&mlystffiQEKT，，&、r??urc?:Conwncr?lHxriMjp<SO4aaf^'owtttcc:EzewtofSeMs\wniiB：LoaAnarfCeiCoMig??拘ICoMiQMaMW尸eitoMg)fs^eVype0:Sc^cdultTypesef^nwrRewjtwQueueCU^dBiotWh^Qijeu^^^j^ijRe^jn^T^^TvoidV0ldsopeQwld，MhitV5(ExesS訪eMce):*。億?dft?oljr?5(Conc?rTentM35hM3^<$hihc.tMegr>)"id^W^ii1<riR<uouiCfcQuau4(Urik<nlBk^rioQu