CountDownLatch 同步倒數(shù)計數(shù)器

CountDownLatch是一個同步倒數(shù)計數(shù)器适篙。CountDownLatch允許一個或多個線程等待其他線程完成操作。
CountDownLatch對象內(nèi)部存有一個整數(shù)作為計數(shù)器晾腔。調(diào)用countDown()方法就將計數(shù)器減1，當計數(shù)到達0時啊犬，則所有等待者會停止等待灼擂。計數(shù)器的操作是原子性的。
CountDownLatch類的常用API
構(gòu)造方法
*CountDownLatch(int count) * 構(gòu)造方法參數(shù)指定了計數(shù)的次數(shù)觉至。
方法
void await() 使當前線程在鎖存器倒計數(shù)至0之前一直等待剔应，除非線程被中斷。
boolean await(long timeout, TimeUnit unit) 使當前線程在鎖存器倒計數(shù)至0之前一直等待，除非線程被中斷或超出了指定的等待時間峻贮。
void countDown() 計數(shù)減1席怪。當計數(shù)為0，則釋放所有等待的線程纤控。
long getCount() 返回當前計數(shù)挂捻。
String toString() 返回標識此鎖存器及其狀態(tài)的字符串。
用給定的計數(shù)初始化 CountDownLatch實例船万。每調(diào)用一次countDown()方法刻撒，計數(shù)器減1。計數(shù)器大于0 時唬涧，await()方法會阻塞其他線程繼續(xù)執(zhí)行疫赎。 利用該特性，可以讓主線程等待子線程的結(jié)束碎节。

需要注意的是捧搞，一旦CountDownLatch的計數(shù)到0，則無法再將該計數(shù)無法被重置狮荔。

一種典型的場景就是火箭發(fā)射胎撇。在火箭發(fā)射前，為了保證萬無一失殖氏，往往還要進行各項設(shè)備晚树、儀器的檢查。只有等所有檢查完畢后雅采，引擎才能點火爵憎。這種場景就非常適合使用CountDownLatch。它可以使得點火線程婚瓜，等待所有檢查線程全部完工后宝鼓，再執(zhí)行。
例：有三個工人在為老板干活巴刻。老板有一個習慣愚铡，當三個工人把一天的活都干完了的時候，他就來檢查所有工人所干的活胡陪。如下代碼設(shè)計兩個類沥寥，Worker代表工人，Boss代表老板柠座。

import java.util.Random; import java.util.concurrent.CountDownLatch; 
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; 
import java.util.concurrent.TimeUnit; 
public class CountDownLatchDemo { 
public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        CountDownLatch latch = new CountDownLatch(3);    // 同步倒數(shù)計數(shù)器邑雅。
 Worker w1 = new Worker(latch, "張三");
        Worker w2 = new Worker(latch, "李四");
        Worker w3 = new Worker(latch, "王五");
        Boss boss = new Boss(latch);

        executor.execute(w3); // 工人工作。
 executor.execute(w2);
        executor.execute(w1);
        executor.execute(boss); // 老板工作愚隧。
 executor.shutdown();
    }
}
 class Worker implements Runnable { 
private CountDownLatch downLatch; 
private String name; 
public Worker(CountDownLatch downLatch, String name) { 
this.downLatch = downLatch; 
this.name = name;
 } 
public void run() { 
        this.doWork();    // 工人工作蒂阱。
        try {
            TimeUnit.SECONDS.sleep(new Random().nextInt(10));  // 工作時長锻全。
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(this.name + "活干完了！"); 
        this.downLatch.countDown();  // 計數(shù)減1录煤。
 } private void doWork() {
        System.out.println(this.name + "正在干活!");
    }
} 
class Boss implements Runnable { 
private CountDownLatch downLatch; 
public Boss(CountDownLatch downLatch) { 
this.downLatch = downLatch;
    } public void run() {
        System.out.println("老板正在等所有的工人干完活......"); try { this.downLatch.await();    // 當計數(shù)不為0時鳄厌，線程永遠阻塞。為0則繼續(xù)執(zhí)行妈踊。
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("工人活都干完了了嚎，老板開始檢查了！");
    }
}

CountDownLatch類與join方法

CountDownLatch實例本質(zhì)與Thread的join方法相同廊营。但join方法僅可以支持當前線程等待一個線程的結(jié)束歪泳，若需要等待多個線程，則需要逐個線程的調(diào)用join方法露筒，非常麻煩呐伞。CountDwonLatch可以很方便的實現(xiàn)一個線程等待多個線程。

CyclicBarrier 循環(huán)屏障

CyclicBarrier用于讓一組線程運行并互相等待慎式，直到共同到達一個公共屏障點 (common barrier point伶氢，又被稱為同步點)，被屏障攔截的所有線程就會繼續(xù)執(zhí)行瘪吏。

CyclicBarrier與CountDownLatch的功能非常類似癣防。但一個CyclicBarrier實例在釋放等待線程后可以繼續(xù)使用。讓下一批線程在屏障點等待掌眠。但CountDownLatch實例只能被使用一次蕾盯。所以CyclicBarrier被稱為*循環(huán) *的 barrier。

典型的比如公司的人員利用集體郊游蓝丙，先各自從家出發(fā)到公司集合级遭，再同時出發(fā)游玩，在指定地點集合渺尘。CyclicBarrier表示大家彼此在某處等待装畅，集合好后才開始出發(fā)，分散活動后又在指定地點集合碰面沧烈。

CyclicBarrier類API

構(gòu)造器

CyclicBarrier(int parties) 創(chuàng)建CyclicBarrier對象，parties 表示屏障攔截的線程數(shù)量像云。

CyclicBarrier(int parties, Runnable barrierAction) 創(chuàng)建 CyclicBarrier對象锌雀，該構(gòu)造方法提供了一個Runnable 參數(shù)，在一組線程中的最后一個線程到達之后迅诬，執(zhí)行Runnable中的程序腋逆，再之后釋放正在等待的線程。Runnable在屏障點上只運行一次侈贷。

方法

int await() 通知CyclicBarrier實例惩歉，當前線程已經(jīng)到達屏障點，然后當前線程將被阻塞。

int await(long timeout, TimeUnit unit) 指定當前線程被阻塞的時間撑蚌。

int getNumberWaiting() 返回當前在屏障處等待的線程數(shù)上遥。

int getParties() 返回CyclicBarrier的需要攔截的線程數(shù)。

boolean isBroken() 查詢此屏障是否處于損壞狀態(tài)争涌。

void reset() 將屏障重置為其初始狀態(tài)粉楚。

例1：各省數(shù)據(jù)獨立，分庫存?zhèn)恋妗榱颂岣哂嬎阈阅苣Ｈ恚y(tǒng)計時采用每個省開一個線程先計算單省結(jié)果，最后匯總饮潦。

import java.util.concurrent.BrokenBarrierException; 
import java.util.concurrent.CyclicBarrier; 
public class Total { 
public static void main(String[] args) {
        TotalService totalService = new TotalServiceImpl();
        CyclicBarrier barrier = new CyclicBarrier(5, new TotalTask(totalService)); // 實際系統(tǒng)是查出所有省編碼code的列表燃异，然后循環(huán)，每個code生成一個線程继蜡。
        new BillTask(new BillServiceImpl(), barrier, "北京").start(); 
        new BillTask(new BillServiceImpl(), barrier, "上海").start(); 
        new BillTask(new BillServiceImpl(), barrier, "廣西").start(); 
        new BillTask(new BillServiceImpl(), barrier, "四川").start(); 
        new BillTask(new BillServiceImpl(), barrier, "黑龍江").start();
    }

} /** * 主任務(wù)：匯總?cè)蝿?wù) */
class TotalTask implements Runnable { 
    private TotalService totalService;

    TotalTask(TotalService totalService) { 
    this.totalService = totalService;
    } public void run() { 
    // 讀取內(nèi)存中各省的數(shù)據(jù)匯總回俐，過程略。
     totalService.count();
        System.out.println("開始全國匯總");
    }

} /** * 子任務(wù)：計費任務(wù) */
class BillTask extends Thread { 
    private BillService billService;     // 計費服務(wù)
    private CyclicBarrier barrier; 
    private String code;    // 代碼壹瘟，按省代碼分類鲫剿，各省數(shù)據(jù)庫獨立。
BillTask(BillService billService, CyclicBarrier barrier, String code) { 
    this.billService = billService; 
    this.barrier = barrier; 
    this.code = code;
    } 
public void run() {
        System.out.println("開始計算--" + code + "省--數(shù)據(jù)稻轨！");
        billService.bill(code); // 把bill方法結(jié)果存入內(nèi)存灵莲，如ConcurrentHashMap,vector等,代碼略
        System.out.println(code + "省已經(jīng)計算完成,并通知匯總Service！"); 
        try { // 通知barrier已經(jīng)完成
        barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
} 
interface BillService { 
      public void bill(String code);
} 
interface TotalService { 
      public void count();
} 
class BillServiceImpl implements BillService{

    @Override 
    public void bill(String code) {}
} 
class TotalServiceImpl implements TotalService{

    @Override 
    public void count(){}
}

例2：賽跑時殴俱，等待所有人都準備好時政冻，才起跑。

public class CyclicBarrierTest { public static void main(String[] args) throws IOException, InterruptedException {
        CyclicBarrier barrier = new CyclicBarrier(3);
        ExecutorService executor = Executors.newFixedThreadPool(3);

        executor.submit(new Thread(new Runner(barrier, "1號選手")));
        executor.submit(new Thread(new Runner(barrier, "2號選手")));
        executor.submit(new Thread(new Runner(barrier, "3號選手")));

        executor.shutdown();
    }
} class Runner implements Runnable { // 一個同步輔助類线欲，它允許一組線程互相等待明场，直到到達某個公共屏障點 (common barrier point)
    private CyclicBarrier barrier; private String name; 
    public Runner(CyclicBarrier barrier, String name) { 
    super(); 
    this.barrier = barrier; 
    this.name = name;
    }

    @Override 
    public void run() { 
    try {
            Thread.sleep(1000 * (new Random()).nextInt(8));
            System.out.println(name + " 準備好了..."); // barrier的await方法，在所有參與者都已經(jīng)在此           barrier 上調(diào)用 await 方法之前李丰，將一直等待苦锨。
       barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println(name + " 起跑！");
    }

}

例3：JDK6中的示例用法：下面是一個在并行分解設(shè)計中使用 barrier 的例子趴泌。給出示意代碼結(jié)構(gòu)舟舒，不可運行。

class Solver { 
final int N; final float[][] data; 
final CyclicBarrier barrier; 
class Worker implements Runnable { 
int myRow; 
public Worker(int row) {
            myRow = row;
        } public void run() { while (!done()) {
                processRow(myRow); try {
                    barrier.await();
                } catch (InterruptedException ex) { return;
                } catch (BrokenBarrierException ex) { return;
                }
            }
        }

    } public Solver(float[][] matrix) {
        data = matrix;
        N = matrix.length;

        barrier = new CyclicBarrier(N, new Runnable() { public void run() {
                mergeRows(...);
            }
        }); for (int i = 0; i < N; ++i){ new Thread(new Worker(i)).start();
        }
        waitUntilDone();
    }
}

在這個例子中嗜憔，每個 worker 線程處理矩陣的一行秃励，在處理完所有的行之前，該線程將一直在屏障處等待吉捶。處理完所有的行之后夺鲜，將執(zhí)行所提供的 Runnable 屏障操作皆尔，并合并這些行。如果合并者確定已經(jīng)找到了一個解決方案币励，那么 done() 將返回 true慷蠕，所有的 worker 線程都將終止。

如果屏障操作在執(zhí)行時不依賴于正掛起的線程榄审，則線程組中的任何線程在獲得釋放時都能執(zhí)行該操作砌们。為方便此操作，每次調(diào)用 await() 都將返回能到達屏障處的線程的索引搁进。然后浪感，可以選擇哪個線程應(yīng)該執(zhí)行屏障操作，例如：

對于失敗的同步嘗試，CyclicBarrier 使用了一種要么全部要么全不 (all-or-none) 的破壞模式：如果因為中斷、失敗或者超時等原因鸡典，導致線程過早地離開了屏障點喇聊，那么在該屏障點等待的其他所有線程也將通過 BrokenBarrierException（如果它們幾乎同時被中斷晕粪，則用 InterruptedException）以反常的方式離開。

Semaphore信號量

Semaphore用于控制并發(fā)線程數(shù)。Semaphore實例可以控制當前訪問自身的線程個數(shù)。使用Semaphore可以控制同時訪問資源的線程個數(shù)捐名。例如，實現(xiàn)一個文件允許的并發(fā)訪問數(shù)闹击。

Semaphore維護了一個許可集镶蹋。“許可”即線程進入臨界區(qū)的許可赏半。一個臨界區(qū)可以有多個許可贺归。獲取許可的線程即可進入。通過 acquire() 獲取一個許可断箫，如果線程沒有獲取到就等待拂酣，而 release() 表示釋放一個許可≈僖澹可以把Semaphore看成是一種共享鎖婶熬。Semaphore允許同一時間多個線程同時訪問臨界區(qū)。

生活的理解：Semaphore實現(xiàn)的功能就類似廁所有5個坑埃撵，假如有十個人要上廁所尸诽，那么同時能有多少個人去上廁所呢？同時只能有5個人能夠占用盯另，當5個人中的任何一個人讓開后，其中在等待的另外5個人中又有一個可以占用了洲赵。另外等待的5個人中可以是隨機獲得優(yōu)先機會鸳惯，也可以是按照先來后到的順序獲得機會商蕴，這取決于構(gòu)造Semaphore對象時傳入的參數(shù)選項。

Semaphore對象也可以實現(xiàn)互斥鎖的功能芝发，并且可以是由一個線程獲得了"鎖"绪商，再由另一個線程釋放"鎖"，這可應(yīng)用于死鎖恢復的一些場合辅鲸。

在一些企業(yè)系統(tǒng)中格郁，開發(fā)人員經(jīng)常需要限制未處理的特定資源請求（線程/操作）數(shù)量，事實上独悴，限制有時候能夠提高系統(tǒng)的吞吐量例书，因為它們減少了對特定資源的爭用。盡管完全可以手動編寫限制代碼刻炒，但使用 Semaphore類可以更輕松地完成此任務(wù)决采，它將幫您執(zhí)行限制。

常用API

public void acquire() // 獲取許可坟奥。
public void acquireUninterruptibly()
public boolean tryAcquire()
public boolean tryAcquire(long timeout, TimeUnit unit)
public void release() // 釋放許可树瞭。該方法一般調(diào)用于finally塊中。

例：10 個線程都在運行爱谁，可以對運行SemaphoreApp的Java進程執(zhí)行jstack來驗證晒喷，只有3個線程是活躍的。在一個信號計數(shù)器釋放之前访敌，其他7個線程都處于空閑狀態(tài)凉敲。

import java.util.Random;
import java.util.concurrent.Semaphore; 
public class SemaphoreApp { 
public static void main(String[] args) { // 匿名Runnable實例。定義線程運行程序捐顷。
        Runnable limitedCall = new Runnable() { final Random rand = new Random(); final Semaphore available = new Semaphore(3);     // 最多可以發(fā)出3個"許可"
            int count = 0; public void run() { int time = rand.nextInt(15); int num = count++; try {
                    available.acquire(); // 當前線程獲取"許可"荡陷。若沒有獲取許可，則等待于此迅涮。
                    System.out.println("Executing " + "long-running action for " + time + " seconds... #" + num); 
                    Thread.sleep(time * 1000); 
                    System.out.println("Done with #" + num + "!"); 
                } catch (InterruptedException intEx) { 
                    intEx.printStackTrace(); 
                } finally {
                    available.release(); // 當前線程釋放"許可"
 }
            } 
        }; for (int i = 0; i < 10; i++) { new Thread(limitedCall).start(); 
        }
}

例：停車示例废赞。停車場只有10個車位，現(xiàn)在有30輛車去停車叮姑。當車位滿時出來一輛車才能有一輛車進入停車唉地。

import java.util.concurrent.Semaphore; public class Car implements Runnable { private final Semaphore parkingSlot; private int carNo; public Car(Semaphore parkingSlot, int carNo) { this.parkingSlot = parkingSlot; this.carNo = carNo;
    } public void run() { try {
            parkingSlot.acquire(); // 車嘗試獲取"車位"
 parking();
            sleep(300);
            leaving();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            parkingSlot.release(); // 釋放"車位"
 }
    } private void parking() {
        System.out.println(String.format("%d號車泊車", carNo));
    } private void leaving() {
        System.out.println(String.format("%d號車離開車位", carNo));
    } private static void sleep(long millis) { try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

} // --------------------------------------------------------------------------
import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; public class ParkingCars { private static final int NUMBER_OF_CARS = 30; private static final int NUMBER_OF_PARKING_SLOT = 10; public static void main(String[] args) {
        Semaphore parkingSlot = new Semaphore(NUMBER_OF_PARKING_SLOT, true);    // "車位",采用FIFO,設(shè)置true。
 ExecutorService service = Executors.newCachedThreadPool();    // 創(chuàng)建線程池传透。模擬30輛車"停車"耘沼。
        for (int carNo = 1; carNo <= NUMBER_OF_CARS; carNo++) {
            service.execute(new Car(parkingSlot, carNo));
        }
        sleep(3000);
        service.shutdown(); // 關(guān)閉線程池。 // 輸出剩余可以用的資源數(shù)朱盐。
        System.out.println(parkingSlot.availablePermits() + " 個停車位可以用!");
    } private static void sleep(long millis) { try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

Exchanger 交換器

Exchanger用于實現(xiàn)線程間的數(shù)據(jù)交換群嗤。Exchanger提供一個同步點，在同步點上兵琳，兩個線程使用exchange方法交換彼此數(shù)據(jù)狂秘。如果第一個線程先執(zhí)行exchange方法骇径，則它會等待第二個線程執(zhí)行exchange方法。當兩個線程同時到達同步點時者春，這兩個線程即可以交換數(shù)據(jù)破衔。交換完畢后，各自進行以后的程序流程钱烟。當兩個線程通過Exchanger交換數(shù)據(jù)的時候晰筛，這個交換對于兩個線程來說是線程安全的。

exchange()方法將本線程的數(shù)據(jù)作為參數(shù)拴袭，傳遞給伙伴線程读第，并且該方法返回伙伴線程提供的數(shù)據(jù)。

當在運行不對稱的活動時Exchanger很有用稻扬，比如當一個線程填充了buffer卦方，另一個線程從buffer中消費數(shù)據(jù)時，這兩個線程可以用Exchanger來交換數(shù)據(jù)泰佳。

Exchanger<V>類的API

構(gòu)造器

Exchanger() 創(chuàng)建一個新的 Exchanger盼砍。

方法

V exchange(V x) 等待另一個線程到達此交換點（除非當前線程被中斷），然后將給定的對象傳送給該線程逝她，并接收該線程的對象浇坐。

V exchange(V x, long timeout, TimeUnit unit) 等待另一個線程到達此交換點（除非當前線程被中斷，或者超出了指定的等待時間）黔宛，然后將給定的對象傳送給該線程近刘，同時接收該線程的對象。

例：以下這個程序demo要做的事情就是生產(chǎn)者在交換前生產(chǎn)5個"生產(chǎn)者"臀晃，然后再與消費者交換5個數(shù)據(jù)觉渴，然后再生產(chǎn)5個"交換后生產(chǎn)者"，而消費者要在交換前消費5個"消費者"徽惋，然后再與生產(chǎn)者交換5個數(shù)據(jù)案淋，然后再消費5個"交換后消費者"。importjava.util.ArrayList;

import java.util.Iterator; import java.util.List; import java.util.concurrent.Exchanger; /** * 兩個線程間的數(shù)據(jù)交換 */
public class ExchangerDemo { private static final Exchanger<List<String>> ex = new Exchanger<List<String>>(); private static void sleep(long millis){ try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    } /** * 內(nèi)部類险绘，數(shù)據(jù)生成者 */
    class DataProducer implements Runnable { private List<String> list = new ArrayList<String>(); public void run() {
            System.out.println("生產(chǎn)者開始生產(chǎn)數(shù)據(jù)"); for (int i = 1; i <= 5; i++) {
                System.out.println("生產(chǎn)了第" + i + "個數(shù)據(jù)踢京，耗時1秒");
                list.add("生產(chǎn)者" + i);
                sleep(1000);
            }
            System.out.println("生產(chǎn)數(shù)據(jù)結(jié)束");
            System.out.println("開始與消費者交換數(shù)據(jù)"); try { //將數(shù)據(jù)準備用于交換，并返回消費者的數(shù)據(jù)
                list = (List<String>) ex.exchange(list);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("結(jié)束與消費者交換數(shù)據(jù)");
            System.out.println("生產(chǎn)者與消費者交換數(shù)據(jù)后宦棺，再生產(chǎn)數(shù)據(jù)"); for (int i = 6; i < 10; i++) {
                System.out.println("交換后生產(chǎn)了第" + i + "個數(shù)據(jù)瓣距，耗時1秒");
                list.add("交換后生產(chǎn)者" + i);
                sleep(1000);
            }

            System.out.println("遍歷生產(chǎn)者交換后的數(shù)據(jù)"); for (Iterator<String> iterator = list.iterator(); iterator.hasNext();) {
                System.out.println(iterator.next());
            }
        }

    } /** * 內(nèi)部類，數(shù)據(jù)消費者 */
    class DataConsumer implements Runnable { private List<String> list = new ArrayList<String>(); public void run() {
            System.out.println("消費者開始消費數(shù)據(jù)"); for (int i = 1; i <= 5; i++) {
                System.out.println("消費了第" + i + "個數(shù)據(jù)"); // 消費者產(chǎn)生數(shù)據(jù)代咸，后面交換的時候給生產(chǎn)者
                list.add("消費者" + i);
             }

            System.out.println("消費數(shù)據(jù)結(jié)束");
            System.out.println("開始與生產(chǎn)者交換數(shù)據(jù)"); try { // 進行數(shù)據(jù)交換蹈丸，返回生產(chǎn)者的數(shù)據(jù)
                list = (List<String>) ex.exchange(list);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("消費者與生產(chǎn)者交換數(shù)據(jù)后，再消費數(shù)據(jù)"); for (int i = 6; i < 10; i++) {
                System.out.println("交換后消費了第" + i + "個數(shù)據(jù)");
                list.add("交換后消費者" + i);
                sleep(1000);
            }
            sleep(1000);
            System.out.println("開始遍歷消費者交換后的數(shù)據(jù)"); for (Iterator<String> iterator = list.iterator(); iterator.hasNext();) {
                System.out.println(iterator.next());
            }
        }
    } // 主方法
    public static void main(String args[]) {
        ExchangerDemo et = new ExchangerDemo(); new Thread(et.new DataProducer()).start(); new Thread(et.new DataConsumer()).start();
    }
}