丨如何使用設(shè)計(jì)模式優(yōu)化并發(fā)編程

publicclassQueryIdActionpublicvoidexecute(Contextcontext)trytrylongid=}(InterruptedException{}}}//執(zhí)行方publicclassExecutionTaskimplementsRunnableprivateQueryNameActionqueryNameAction=newprivateQueryIdActionqueryIdAction=newpublicvoidrun()finalContextcontext=newSystem.out.println("ThenamequerySystem.out.println("TheidquerySystem.out.println("TheNameis"+context.getName()"and}}publicstaticvoidmain(String[]args)72

IntStream.range(1,5).forEach(i->newThread(newContextTest().new}執(zhí)行結(jié)果代ThenamequeryThenamequeryThenamequeryThenamequeryTheidqueryTheidqueryTheidqueryTheidqueryTheNameisThread-1andidTheNameisThread-2andidTheNameisThread-3andidTheNameisThread-0andid然而這種方式太笨拙了，每次調(diào)用方法時(shí)，都需要傳入Context為參數(shù)，而且影響一些除了以上這些方法，其實(shí)我們還可以使用ThreadLocal實(shí)現(xiàn)上下文。ThreadLocal是線程本地變量，可以實(shí)現(xiàn)多線程的數(shù)據(jù)。ThreadLocal為每一個(gè)使用該變量的線程都提供一份獨(dú)立的副本，線程間的數(shù)據(jù)是的，每一個(gè)線程只能各自內(nèi)部的副本變量。ThreadLocal中有三個(gè)常用的方法：set、get、initialValue，我們可以通過(guò)以下一個(gè)簡(jiǎn)單的例子來(lái)看看ThreadLocal的使用：代privatevoidtestThreadLocal()Threadt=newThread()ThreadLocal<String>mStringThreadLocal=new4publicvoidrun() 14接下來(lái)，我們使用ThreadLocal來(lái)重新實(shí)現(xiàn)最開始的上下文設(shè)計(jì)。你會(huì)發(fā)現(xiàn)，我們?cè)趦蓚€(gè)方法中并沒(méi)有通過(guò)變量來(lái)傳下文，只是通過(guò)ThreadLocal獲取了當(dāng)前線程的上下文信代publicclassContextTest//上下文publicstaticclassContextprivateStringprivatelong6publiclonggetId()return publicvoidsetId(longid) this.id= publicStringgetName() return publicvoidsetName(Stringname) = //上下文到ThreadLocal publicfinalstaticclassActionContext privatestaticfinalThreadLocal<Context>threadLocal=new protectedContextinitialValue() returnnew publicstaticActionContextgetActionContext() return publicContextgetContext() return //獲取ActionContext單 publicstaticclassContextHolder privatefinalstaticActionContextactionContext=new //設(shè)置上下文名publicclassQueryNameActionpublicvoidexecute()tryStringname= }catch(InterruptedExceptione) //設(shè)置上下文publicclassQueryIdActionpublicvoidexecute()trylongid=}catch(InterruptedExceptione) //執(zhí)行方publicclassExecutionTaskimplementsRunnableprivateQueryNameActionqueryNameAction=newprivateQueryIdActionqueryIdAction=new publicvoidrun() queryNameAction.execute();// System.out.println("Thenamequery queryIdAction.execute();//設(shè)置線程 System.out.println("Theidquery System.out.println("TheNameis"+ publicstaticvoidmain(String[]args)IntStream.range(1,5).forEach(i->newThread(newContextTest().new 93運(yùn)行結(jié)果代ThenamequeryThenamequeryThenamequeryThenamequeryTheidqueryTheidqueryTheidqueryTheidqueryTheNameisThread-2andidTheNameisThread-0andidTheNameisThread-1andidTheNameisThread-3andidThread-Per-Message設(shè)計(jì)模式翻譯過(guò)來(lái)的意思就是每個(gè)消息一個(gè)線程的意思。例如，我們?cè)谔幚鞸ocket通信的時(shí)候，通常是一個(gè)線程處理以及I/O讀寫，如果I/O讀這個(gè)時(shí)候Thread-Per-Message模式就可以很好地解決這個(gè)問(wèn)題，一個(gè)線程I/O事件，每當(dāng)?shù)揭粋€(gè)I/O，則交給另一個(gè)處理線程執(zhí)行I/O操作。下面，我們還是通代//IO處publicclassServerHandlerimplementsprivateSocket4publicServerHandler(Socketsocket)this.socket= 8publicvoidrun()BufferedReaderin=PrintWriterout=Stringmsg=tryin=newBufferedReader(newout=newwhile((msg=in.readLine())!=null&&msg.length()!=0){//當(dāng)連接成功后在System.out.println("serverreceived:"+ }catch(Exceptione)}finallytry}catch(IOExceptione) try}catch(Exceptione) try}catch(IOExceptione) 41代//SocketpublicclassServer3privatestaticintDEFAULT_PORT=privatestaticServerSocket6publicstaticvoidstart()throwsIOException publicstaticvoidstart(intport)throwsIOExceptionif(server!=null) try //啟動(dòng)服 server=new //通過(guò)無(wú)線循環(huán)客戶端連 while(true) Socketsocket= //當(dāng)有新的客戶端接入時(shí)，會(huì)執(zhí)行下 longstart= newThread(new longend= }finally

System.out.println("Spendtimeis"+(end-

if(server!=null)System.out.println服務(wù)器已關(guān)閉。");}} publicstaticvoidmain(String[]args)throws//運(yùn)行服務(wù)newThread(new{publicvoidrun()try}catch(IOExceptione)}}}}以上，我們是完成了一個(gè)使用Thread-Per-Message計(jì)模式實(shí)現(xiàn)的Socket務(wù)端的代使用這種設(shè)計(jì)模式，如果遇到大的高并發(fā)，就會(huì)出現(xiàn)嚴(yán)重的性能問(wèn)題。如果針對(duì)每個(gè)I/O請(qǐng)求都創(chuàng)建一個(gè)線程來(lái)處理，在有大量請(qǐng)求同時(shí)進(jìn)來(lái)時(shí)，就會(huì)創(chuàng)建大量線程，而此時(shí)JVM這里的Worker是工人的意思，代表在WorkerThread設(shè)計(jì)模式中，會(huì)有一些工人（線來(lái)。除了工人角色，WorkerThread設(shè)計(jì)模式中還包括了流水線和產(chǎn)品。這種設(shè)計(jì)模式相比Thread-Per-Message計(jì)模式，可以減少頻繁創(chuàng)建、銷毀線程所帶來(lái)我們可以假設(shè)一個(gè)場(chǎng)景來(lái)看下該模式的實(shí)現(xiàn)，通過(guò)WorkerThread計(jì)模式來(lái)完成一個(gè)物假設(shè)一個(gè)物流倉(cāng)庫(kù)的物流分揀流水線上有8個(gè)機(jī)器人，它們不斷從流水線上獲取包裹并對(duì)代//publicclassPackageprivateStringprivateString5publicStringgetName()return 9publicvoidsetName(Stringname)= publicStringgetAddress()return publicvoidsetAddress(Stringaddress)this.address= publicvoidexecute()System.out.println(Thread.currentThread().getName()+"executed 25//3finalstaticint3finalstaticintMAX_PACKAGE_NUM=45finalPackage[]6finalWorker[]7int8int9intPackageChannel(intworkers)this.packageQueue=newthis.head=this.tail=this.count=this.workerPool=new}voidinit()for(inti=0;i<workerPool.length;i++){workerPool[i]=newWorker("Worker-"+i,}}*switchtostartallofworkertovoidstartWorker()}synchronizedvoidput(Packagepackagereq)while(count>=packageQueue.length)try}catch(InterruptedExceptione)}}this.packageQueue[tail]=this.tail=(tail+1)%}synchronizedPackagetake()while(count<=0)try

代 }catch(InterruptedExceptione)

}Packagerequest=this.head=(this.head+1)%this.count--return}}代//publicclassWorkerextendsprivatestaticfinalRandomrandom=newprivatefinalPackageChannel5 publicWorker(Stringname,PackageChannelchannel)78this.channel=9}publicvoidrun()while(true)try}catch(InterruptedExceptione)}}}}代publicclassTestpublicstaticvoidmain(String[]args)//新建8finalPackageChannelchannel=new//開始工78915

//為流水線添加for(inti=0;i<100;i++)Packagepackagereq=newPackage();}}我們可以看到，這里有8個(gè)工人在不斷地分揀倉(cāng)庫(kù)中已經(jīng)包裝好的商讀寫分離的業(yè)務(wù)場(chǎng)景中，則經(jīng)常會(huì)用到ThreadLocal實(shí)現(xiàn)動(dòng)態(tài)切換數(shù)據(jù)源操作。但在使用ThreadLocal時(shí)，我們需要注意內(nèi)存泄漏問(wèn)題，在之前的第25講中，我們已經(jīng)討論過(guò)這個(gè)當(dāng)主線程處理每次請(qǐng)求都非常耗時(shí)時(shí)，就可能出現(xiàn)阻塞問(wèn)題，這時(shí)候我們可以考慮將主線程業(yè)務(wù)分工到新的業(yè)務(wù)線程中，從而提高系統(tǒng)的并行處理能力。而Threa-Per-Message設(shè)計(jì)模式