Java多(duo)線(xian)程：CountDownLatch、CyclicBarrier 和 Semaphore
場景(jing)描述：
多線程設計過程中(zhong)，經常會遇到(dao)需要等待其(qi)它線程結束以后再做其(qi)他事情的情況(kuang)。
有幾(ji)種方案(an)：
1.在主線(xian)(xian)程(cheng)中設置一(yi)自(zi)定義(yi)全(quan)局計(ji)數(shu)標(biao)志，在工作線(xian)(xian)程(cheng)完(wan)(wan)成(cheng)時，計(ji)數(shu)減(jian)1。主線(xian)(xian)程(cheng)偵測該標(biao)志是(shi)否為0，一(yi)旦為0，表示(shi)所有工作線(xian)(xian)程(cheng)已經完(wan)(wan)成(cheng)。
2.使(shi)用Java標準的(de)類CountDownLatch來完成這項工作，原理(li)是一(yi)樣的(de)，計(ji)數。
CountDownLatch
一個(ge)(ge)同步輔助(zhu)類，在(zai)完成(cheng)一組正在(zai)其他線程中執行的操(cao)作(zuo)之(zhi)前，它允許一個(ge)(ge)或多個(ge)(ge)線程一直等(deng)待。
其機制是：
當多(duo)(duo)個(ge)(ge)（具體數量等于(yu)初始化CountDownLatch時的(de)(de)count參數的(de)(de)值(zhi)）線(xian)程(cheng)都達到了預期狀(zhuang)態(tai)或完成預期工作時觸發(fa)事(shi)件，其他線(xian)程(cheng)可(ke)以等待(dai)(dai)這(zhe)個(ge)(ge)事(shi)件來(lai)觸發(fa)自己的(de)(de)后(hou)續工作。這(zhe)里需(xu)要注意的(de)(de)是(shi)(shi)，等待(dai)(dai)的(de)(de)線(xian)程(cheng)可(ke)以是(shi)(shi)多(duo)(duo)個(ge)(ge)，即CountDownLatch是(shi)(shi)可(ke)以喚醒多(duo)(duo)個(ge)(ge)等待(dai)(dai)的(de)(de)線(xian)程(cheng)的(de)(de)。達到自己預期狀(zhuang)態(tai)的(de)(de)線(xian)程(cheng)會(hui)調用(yong)CountDownLatch的(de)(de)countDown方法，而等待(dai)(dai)的(de)(de)線(xian)程(cheng)會(hui)調用(yong)CountDownLatch的(de)(de)await方法。
CountDownLatch 很適合用來將一個任務分(fen)為n個獨立(li)的部分(fen)，等這些部分(fen)都完成后繼續接下來的任務，CountDownLatch 只(zhi)能出發一次(ci)，計數值(zhi)不能被(bei)重(zhong)置。
流程圖
如上圖所示，當(dang)7個線(xian)程都(dou)完成latch.countDown調用后，最下面(mian)那條線(xian)程會從latch.await返回(hui)，繼續執行(xing)后面(mian)的代(dai)碼
函數列表
- CountDownLatch(int count) ：構造一個用給定計數初始(shi)化的 CountDownLatch。
- void await()：使當前線程(cheng)在鎖存(cun)器倒計數至(zhi)零之前一(yi)直等待，除非線程(cheng)被中斷。
- boolean await(long timeout, TimeUnit unit) 使當前(qian)線程(cheng)在(zai)鎖存(cun)器(qi)倒(dao)計數(shu)至零之前(qian)一(yi)直等待，除非線程(cheng)被中斷或(huo)超出了指定的(de)等待時間。
- void countDown() 遞(di)減鎖(suo)存(cun)器的計數(shu)，如果(guo)計數(shu)到(dao)達零(ling)，則釋(shi)放所有等待的線程。
實現原理
請參(can)考：
實(shi)例
我(wo)們(men)來看一(yi)個具體的例子。假設我(wo)們(men)使用一(yi)臺多核(he)的機器對(dui)一(yi)組數據(ju)進(jin)行排序(xu)，那(nei)么我(wo)們(men)可(ke)以把這(zhe)組數據(ju)分到(dao)不同線程(cheng)中進(jin)行排序(xu)，然(ran)后合(he)并；可(ke)以利(li)用線程(cheng)池來管理多線程(cheng)；可(ke)以將CountDownLatch用作各個分組數據(ju)都排好(hao)序(xu)的通知。下面是代碼片段：
先看主線程
int count = 10; final CountDownLatch latch = new CountDownLatch(count); int[] datas = new int[10204]; int step = datas.length / count; for (int i=0; i < count; i++) { int start = i * step; int end = (i+1) * step; if (i == count - 1) end = datas.length; threadpool.execute(new MyRunnable(latch, datas, start, end)); } latch.await(); //合并數(shu)據
我們再看(kan)一下具體任務的(de)代碼，即MyRunnable的(de)run方法的(de)實(shi)現：
public void run() { //數據排(pai)序 latch.countDown(); }
CyclicBarrier
可以協同多個(ge)線程(cheng)，讓(rang)多個(ge)線程(cheng)在這個(ge)屏(ping)障前等待(dai)，直到(dao)所有線程(cheng)都達到(dao)了(le)這個(ge)屏(ping)障時(shi)，再一(yi)起繼續執行后面的(de)動作。
CyclicBarrier適用于多個線程(cheng)有固定的多步需(xu)要執行(xing)(xing)，線程(cheng)間互相等待，當都(dou)執行(xing)(xing)完(wan)了(le)，再一(yi)起(qi)執行(xing)(xing)下一(yi)步。因為(wei)該 barrier 在(zai)釋放等待線程(cheng)后可以重用，所(suo)以稱它為(wei)循環 的 barrier。
流程圖(tu)
上(shang)圖中的7個(ge)(ge)線(xian)(xian)程(cheng)(cheng)各(ge)有一(yi)個(ge)(ge)barrier.await,那么(me)任何一(yi)個(ge)(ge)線(xian)(xian)程(cheng)(cheng)在執行到barrier.await時就會(hui)(hui)進入阻(zu)塞(sai)等待狀態，直(zhi)到7個(ge)(ge)線(xian)(xian)程(cheng)(cheng)都到了(le)barrier.await時才會(hui)(hui)同時從await返回，繼續后(hou)面的工作。此外如(ru)果(guo)在構造CyclicBarrier時設置了(le)一(yi)個(ge)(ge)Runnable實現，那么(me)最后(hou)一(yi)個(ge)(ge)到barrier.await的線(xian)(xian)程(cheng)(cheng)會(hui)(hui)執行這個(ge)(ge)Runnable的run方法，以完成一(yi)些(xie)預設的工作。
(01) CountDownLatch的作用(yong)是允(yun)(yun)許(xu)1或N個線(xian)(xian)程等待其他線(xian)(xian)程完成執行；而CyclicBarrier則是允(yun)(yun)許(xu)N個線(xian)(xian)程相互等待。
(02) CountDownLatch的計(ji)數器無法被(bei)(bei)重置；CyclicBarrier的計(ji)數器可以(yi)被(bei)(bei)重置后使用，因此(ci)它被(bei)(bei)稱為(wei)是循環的barrier。
CountDownLatch 適用于一(yi)組(zu)線程(cheng)(cheng)和(he)另(ling)一(yi)個(ge)(ge)主線程(cheng)(cheng)之(zhi)間的(de)工作(zuo)協作(zuo)。一(yi)個(ge)(ge)主線程(cheng)(cheng)等待一(yi)組(zu)工作(zuo)線程(cheng)(cheng)的(de)任務完畢才繼(ji)續它的(de)執(zhi)行是使用 CountDownLatch 的(de)主要場景；CyclicBarrier 用于一(yi)組(zu)或(huo)幾組(zu)線程(cheng)(cheng)，比如一(yi)組(zu)線程(cheng)(cheng)需要在(zai)一(yi)個(ge)(ge)時(shi)間點上達成一(yi)致(zhi)，例如同時(shi)開(kai)始一(yi)個(ge)(ge)工作(zuo)。另(ling)外，CyclicBarrier 的(de)循環(huan)特性和(he)構(gou)造函數(shu)所接(jie)受的(de) Runnable 參數(shu)也是 CountDownLatch 所不(bu)具備的(de)。
|
|
CountDownLatch
|
CyclicBarrier
|
|
適用場景
|
主線(xian)程(cheng)等(deng)待其他工(gong)作線(xian)程(cheng)結束
|
多個線程相(xiang)互(hu)等(deng)待(dai)，直(zhi)到所有線程都達到一個障礙點(dian)Barrier
|
|
主要(yao)方(fang)法
|
CountDownLatch(int count) 主(zhu)線(xian)程調用：初始化(hua)計數
await() 主線程調用 ： 阻塞，直到等(deng)待計數為(wei)0時解除阻塞
countDown() 工作線(xian)程調用(yong) ： 計數減(jian)1
|
CyclicBarrier(int parties , Runnnable barrierAction) : 初始(shi)化參與者數量(liang)和障礙(ai)點執行Action，action可選，由主線程初始(shi)化
await() : 由工作(zuo)線程調用，每(mei)被(bei)調用一次，計數便會(hui)減少(shao)1，并阻(zu)塞住當前線程 ， 直到(dao)所有線程都達到(dao)障(zhang)礙(ai)點
|
|
等(deng)待結束
|
各(ge)線程之間不再相互影響， 可以繼續做(zuo)自己的事情， 不再執行下一個工(gong)作(zuo)目(mu)標。
|
在(zai)障礙點到達后， 允許(xu)所有(you)線(xian)程(cheng)繼(ji)續執行，到達下(xia)一個目(mu)標后，可以恢復使(shi)用(yong)CyclicBarrier， barrier 在(zai)釋放等(deng)待線(xian)程(cheng)后可以重(zhong)用(yong)
|
|
異(yi)常(chang)
|
|
如果其中(zhong)一個線程由于中(zhong)斷、錯誤、或(huo)者超時(shi)導致永久離開障礙點，其他線程也將拋出異常。
|
實例
int count = 10; final CyclicBarrier barrier = new CyclicBarrier(count + 1); int[] datas = new int[10204]; int step = datas.length / count; for (int i=0; i < count; i++) { int start = i * step; int end = (i+1) * step; if (i == count - 1) end = datas.length; threadpool.execute(new MyRunnable(barrier, datas, start, end)); } barrier.await(); //合并數據(ju)
可以看到CyclicBarrier對象傳入的(de)參數(shu)值比CountDownLatch大1，原(yuan)因是(shi)構造(zao)CountDownLatch的(de)參數(shu)是(shi)調用(yong)countDown的(de)數(shu)量，而CyclicBarrier的(de)數(shu)量是(shi)await的(de)數(shu)量
public void run() { //數據排(pai)序 try { barrier.await(); }catch (...) }
Semaphore
Semaphore 信(xin)號量對象管理的信(xin)號就(jiu)像令牌(pai)，構造時傳入個數(shu)(shu)，總數(shu)(shu)就(jiu)是控制并發的數(shu)(shu)量。我們需(xu)要控制(zhi)并(bing)發的(de)代碼，執(zhi)行前先獲取(qu)信(xin)(xin)號(hao)(通過(guo)acquire獲取(qu)信(xin)(xin)號(hao)許可(ke))，執(zhi)行后(hou)(hou)歸還信(xin)(xin)號(hao)(通過(guo)release歸還信(xin)(xin)號(hao)許可(ke))。每次acquire成功返回后(hou)(hou)，Semaphore可(ke)用(yong)(yong)的(de)信(xin)(xin)號(hao)量就會(hui)減少一(yi)個，如果(guo)沒有可(ke)用(yong)(yong)的(de)信(xin)(xin)號(hao)，acquire調用(yong)(yong)就會(hui)阻塞，等待有release調用(yong)(yong)釋放信(xin)(xin)號(hao)后(hou)(hou)，acquire才會(hui)得(de)到信(xin)(xin)號(hao)并(bing)返回。
如果(guo)Semaphore管理的(de)信號(hao)量為(wei)1個(ge)，那(nei)么就退化到(dao)(dao)互斥鎖了；如果(guo)多于一個(ge)信號(hao)量，則主要用于控(kong)制(zhi)并(bing)(bing)發(fa)數。與(yu)通(tong)過控(kong)制(zhi)線程數來(lai)控(kong)制(zhi)并(bing)(bing)發(fa)數的(de)方式相比，通(tong)過Semaphore來(lai)控(kong)制(zhi)并(bing)(bing)發(fa)數可以(yi)控(kong)制(zhi)得更加細粒度，因(yin)為(wei)真正被控(kong)制(zhi)最大(da)并(bing)(bing)發(fa)的(de)代碼放到(dao)(dao)acquire和release之間(jian)就行了。
Semaphore類位于java.util.concurrent包下，它提供了(le)2個構造器：
public Semaphore(int permits) { //參(can)數(shu)permits表(biao)示(shi)(shi)許(xu)可數(shu)目(mu)，即(ji)同(tong)時可以(yi)允許(xu)多少(shao)線(xian)程進行(xing)訪問
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) { //這個(ge)多了(le)一個(ge)參(can)數(shu)fair表(biao)示(shi)(shi)是否是公(gong)平的，即(ji)等待(dai)時間越久的越先(xian)獲取許(xu)可
sync = (fair)? new FairSync(permits) : new NonfairSync(permits);
}
實例
例如(ru)我們(men)需要控制遠程(cheng)方(fang)法(fa)的(de)并(bing)發(fa)量，超過并(bing)發(fa)量的(de)方(fang)法(fa)就等待有其(qi)他(ta)方(fang)法(fa)執(zhi)行返(fan)回后(hou)再執(zhi)行，那么(me)其(qi)代碼如(ru)下：
semaphore.acquire(); try { //調(diao)用遠(yuan)程通信的方(fang)法(fa) } finally { semaphore.release(); }
參考資料(liao)：