java併發編程學習10--同步器--倒計時門栓

【同步器

java.util.concurrent包包含幾個能幫助人們管理相互合做的線程集的類。這些機制具備爲線程直間的共用集結點模式提供的‘預製功能’。若是有一個相互合做的線程知足這些行爲模式之一，那麼應該直接使用提供的類庫而不是顯示的使用鎖與條件的集合。

【倒計時門栓

一個倒計時門栓（CountDownlatch）讓一個線程集直到計數變爲0.倒計時門栓是一次性的，一旦計數爲0就不能再重用了。一個有用的特例是計數值爲1的門栓。實現一個只能經過一次的門。線程在門外等待直到另外一個線程將計數值變爲0。舉例來說，假設一個線程集須要一些初始數據來完成工做。工做線程被啓動並在，門外等候，另外一個線程準備數據，當數據準備好時，調用countDown()，全部的工做線程就能夠繼續工做了。而後再使用一個門栓檢查何時工做線程所有運行完成。每一個工做線程在結束前將門栓計數器減一，門栓的計數變爲0就代表工做完成。

【經常使用方法

• public void countDown():遞減鎖存器的計數，若是計數到達零，則釋放全部等待的線程。若是當前計數大於零，則將計數減小。若是新的計數爲零，出於線程調度目的，將從新啓用全部的等待線程。若是當前計數等於零，則不發生任何操做。
• public boolean await():使當前線程在鎖存器倒計數至零以前一直等待，若是當前計數爲零，則此方法馬上返回 true 值。

• public boolean await(long timeout,TimeUnit unit) throws InterruptedException:使當前線程在鎖存器倒計數至零以前一直等待，除非線程被中斷或超出了指定的等待時間。若是當前計數爲零，則此方法馬上返回 true 值。若是當前計數大於零，則出於線程調度目的，將禁用當前線程，且在發生如下三種狀況之一前，該線程將一直處於休眠狀態：

  • 因爲調用 countDown() 方法，計數到達零；或者其餘某個線程中斷當前線程；或者已超出指定的等待時間。若是計數到達零，則該方法返回 true 值。
  • 若是當前線程：在進入此方法時已經設置了該線程的中斷狀態；或者在等待時被中斷，則拋出 InterruptedException，而且清除當前線程的已中斷狀態。
  • 若是超出了指定的等待時間，則返回值爲 false。若是該時間小於等於零，則此方法根本不會等待。

【例子

模擬一個應用程序：在正式開始工做前須要初始化數據，初始化數據使用三個線程，正式執行須要五個線程：

• 初始化線程

public class InitThread implements Runnable{

    private CountDownLatch downLatch;
    private String name;

    public InitThread(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 ie){
        }
        System.out.println(this.name + "初始化數據完成");
        //計數器減一
        this.downLatch.countDown();
    }
    private void doWork(){
        System.out.println(this.name + "正在初始化數據... ...");
    }

}

• 初始化線程監視器

/**
 * 檢測初始化數據監視器,由於須要判斷是否初始化線程所有執行完畢，這裏用callable返回結果。runnable不能返回值因此沒法判斷。
 */
public class InitMonitor implements Callable<String>{

    private ExecutorService executor;
    private CountDownLatch initLatch;
    private List<Runnable> initThreads;

    public InitMonitor(ExecutorService executor){
        this.executor = executor;
        //初始化線程：3個
        initLatch = new CountDownLatch(3);
        initThreads = Arrays.asList(new InitThread(initLatch,"InitOne"),
                                    new InitThread(initLatch,"InitTwo"),
                                    new InitThread(initLatch,"InitThree"));
    }

    public String call() {
        System.out.println("=========初始化START==========");
        initThreads.stream().forEach(initThread -> executor.submit(initThread));
        try {
            initLatch.await();
        } catch (InterruptedException e) {
        }
        System.out.println("***********初始化END*************");
        return "INIT_SUCCESS";
    }
}

• 工做線程

public class ExecuteThread implements Runnable{

    private CountDownLatch downLatch;
    private String name;

    public ExecuteThread(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 ie){
        }
        System.out.println(this.name + "執行完成");
        //計數器減一
        this.downLatch.countDown();
    }
    private void doWork(){
        System.out.println(this.name + "正在執行... ...");
    }
}

• 工做線程監視器

public class ExecuteMonitor implements Callable<String>{

    private ExecutorService executor;
    private CountDownLatch executeLatch;
    private List<Runnable> executeThreads;

    public ExecuteMonitor(ExecutorService executor){
        this.executor = executor;
        //執行線程:5個
        executeLatch = new CountDownLatch(5);
        executeThreads = Arrays.asList(new ExecuteThread(executeLatch,"ExecuteOne"),
                                        new ExecuteThread(executeLatch,"ExecuteTwo"),
                                        new ExecuteThread(executeLatch,"ExecuteThree"),
                                        new ExecuteThread(executeLatch,"ExecuteFour"),
                                        new ExecuteThread(executeLatch,"ExecuteFive"));
    }

    public String call() {
        System.out.println("========執行START========");
        executeThreads.stream().forEach(executeThread -> executor.submit(executeThread));
        try {
            executeLatch.await();
        } catch (InterruptedException e) {
        }
        System.out.println("*********執行END*********");
        return "EXECUTE_SUCCESS";
    }
}

• 應用程序

public class Application implements Runnable{

    private ExecutorService executor;
    private InitMonitor initMonitor;
    private ExecuteMonitor executeMonitor;

    public Application(ExecutorService executor){
        this.executor = executor;
        initMonitor = new InitMonitor(executor);
        executeMonitor = new ExecuteMonitor(executor);
    }
    @Override
    public void run() {
        System.out.println("===============應用程序執行開始====================》》》");
        FutureTask<String> initTask = new FutureTask<String>(initMonitor);
        executor.submit(initTask);
        try {
            //若是初始化成功開始執行工做線程，在調用get()時，若是沒有執行完成會自動阻塞，因此這裏不須要使用isDone檢測。
            if("INIT_SUCCESS".equals(initTask.get())){
                FutureTask<String> executeTask = new FutureTask<String>(executeMonitor);
                executor.submit(executeTask);
                if("EXECUTE_SUCCESS".equals(executeTask.get())){
                    executor.shutdown();
                    System.out.println("===============應用程序執行完畢====================");
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}

• 客戶端

客戶端必定儘可能簡介，全部細節所有屏蔽，這裏只留下一個能夠自定義線程池給用戶自行選擇

public class Test {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        Application application = new Application(executor);
        application.run();
    }
}