日志分析并行分解設計與實現(xiàn)

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

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

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