java并發(fā)編程學(xué)習(xí)11--同步器--柵欄

desdik 發(fā)布于2019-08-15 13:26 / 3207人閱讀

摘要：考慮大量線程運(yùn)行在一次計算的不同部分的情形。一旦所有的線程都到達(dá)了這個柵欄，柵欄就撤銷，線程可以繼續(xù)運(yùn)行。那些已經(jīng)在等待的線程立即中止的調(diào)用。如果在執(zhí)行屏障操作過程中發(fā)生異常，則該異常將傳播到當(dāng)前線程中，并將置于損壞狀態(tài)。

【同步器

java.util.concurrent包包含幾個能幫助人們管理相互合作的線程集的類。這些機(jī)制具有為線程直間的共用集結(jié)點模式提供的‘預(yù)制功能’。如果有一個相互合作的線程滿足這些行為模式之一，那么應(yīng)該直接使用提供的類庫而不是顯示的使用鎖與條件的集合。

【柵欄

CyclicBarrier類實現(xiàn)了一個集結(jié)點（rendezvous）稱為柵欄（barrier）。考慮大量線程運(yùn)行在一次計算的不同部分的情形。當(dāng)所有部分都準(zhǔn)備好時，需要把結(jié)果組合到一起。當(dāng)一個線程完成了它那部分的任務(wù)后，我們讓他運(yùn)行到柵欄處。一旦所有的線程都到達(dá)了這個柵欄，柵欄就撤銷，線程可以繼續(xù)運(yùn)行。如果任何一個線程在柵欄上等待時離開，那么柵欄就被破壞掉（線程離開可能時等待超時）。這種情況下，其他所有線程await方法上拋出BrokenBarrierException異常。那些已經(jīng)在等待的線程立即中止await的調(diào)用。可以提供一個可選的柵欄動作，當(dāng)所有線程到達(dá)柵欄時，就會執(zhí)行這個動作。

Runnable barrierAction = ...
CyclicBarrier barrire = new CyclicBarrier(nthreads,barrierAction);

柵欄被稱為時循環(huán)的，以為可以在所有等待的線程被釋放后重用（這里與倒計時門閂不同，倒計時門閂只能使用一次）。

【常用方法

public int await() throws InterruptedException,BrokenBarrierException：
在所有參與者都已經(jīng)在此 barrier 上調(diào)用 await方法之前，將一直等待。如果當(dāng)前線程不是將到達(dá)的最后一個線程，出于調(diào)度目的，將禁用它，且在發(fā)生以下情況之一前，該線程將一直處于休眠狀態(tài)：

  - 最后一個線程到達(dá)；或者
  - 其他某個線程中斷當(dāng)前線程；或者
  - 其他某個線程中斷另一個等待線程；或者
  - 其他某個線程在等待 barrier 時超時；或者
  - 其他某個線程在此 barrier 上調(diào)用 reset()。

如果當(dāng)前線程在進(jìn)入此方法時已經(jīng)設(shè)置了該線程的中斷狀態(tài)；或者在等待時被中斷則InterruptedException，并且清除當(dāng)前線程的已中斷狀態(tài)。

如果在線程處于等待狀態(tài)時 barrier 被 reset()，或者在調(diào)用 await 時 barrier 被損壞，抑或任意一個線程正處于等待狀態(tài)，則拋出 BrokenBarrierException 異常。

如果任何線程在等待時被中斷，則其他所有等待線程都將拋出 BrokenBarrierException 異常，并將 barrier 置于損壞狀態(tài)。

如果當(dāng)前線程是最后一個將要到達(dá)的線程，并且構(gòu)造方法中提供了一個非空的屏障操作，則在允許其他線程繼續(xù)運(yùn)行之前，當(dāng)前線程將運(yùn)行該操作。

如果在執(zhí)行屏障操作過程中發(fā)生異常，則該異常將傳播到當(dāng)前線程中，并將 barrier 置于損壞狀態(tài)。
返回：到達(dá)的當(dāng)前線程的索引，其中，索引 getParties() - 1 指示將到達(dá)的第一個線程，零指示最后一個到達(dá)的線程.

拋出：

InterruptedException - 如果當(dāng)前線程在等待時被中斷

BrokenBarrierException - 如果另一個線程在當(dāng)前線程等待時被中斷或超時，或者重置了 barrier，或者在調(diào)用 await 時 barrier 被損壞，抑或由于異常而導(dǎo)致屏障操作（如果存在）失敗。

【例子

1.應(yīng)用程序啟動前，所有子線程準(zhǔn)備工作完成，不需要返回值

public class Application{

    public static void main(String[] args) throws IOException, InterruptedException {
        //需要三個子線程準(zhǔn)備
        CyclicBarrier barrier = new CyclicBarrier(3);
        //如果初始化barrier的線程數(shù)，與實際子線程數(shù)不同，barrier將會一直等待
        ExecutorService executor = Executors.newFixedThreadPool(3);
        executor.submit(new Thread(new InitThread(barrier, "子線程1")));
        executor.submit(new Thread(new InitThread(barrier, "子線程2")));
        executor.submit(new Thread(new InitThread(barrier, "子線程3")));

        executor.shutdown();
    }
}

class InitThread implements Runnable {
    // 一個同步輔助類，它允許一組線程互相等待，直到到達(dá)某個公共屏障點 (common barrier point)
    private CyclicBarrier barrier;
    private String name;

    public InitThread(CyclicBarrier barrier, String name) {
        this.barrier = barrier;
        this.name = name;
    }
    @Override
    public void run() {
        try {
            Thread.sleep(1000 * (new Random()).nextInt(8));
            System.out.println(name + " 初始化完成...");
            barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println(name + " 開始運(yùn)行！");
    }
}

2.應(yīng)用程序啟動前，所有子線程準(zhǔn)備工作完成，需要返回值匯總

public class ApplicationNeedReturn {

    public static void main(String[] args) throws IOException, InterruptedException, ExecutionException {
        //需要三個子線程準(zhǔn)備
        CyclicBarrier barrier = new CyclicBarrier(3);
        //如果初始化barrier的線程數(shù)，與實際子線程數(shù)不同，barrier將會一直等待
        ExecutorService executor = Executors.newFixedThreadPool(3);
        FutureTask time1 = new FutureTask(new InitThread2(barrier, "子線程1"));
        FutureTask time2 = new FutureTask(new InitThread2(barrier, "子線程2"));
        FutureTask time3 = new FutureTask(new InitThread2(barrier, "子線程3"));
        executor.submit(time1);
        executor.submit(time2);
        executor.submit(time3);
        long total = time1.get() + time2.get() + time3.get();
        System.out.println("準(zhǔn)備完成，耗時：" + total);
        executor.shutdown();
    }
}

class InitThread2 implements Callable {

    private CyclicBarrier barrier;
    private String name;
    public InitThread2(CyclicBarrier barrier, String name) {
        this.barrier = barrier;
        this.name = name;
    }

    @Override
    public Long call() {
        long time = 1000 * (new Random()).nextInt(8);
        try {
            Thread.sleep(time);
            System.out.println(name + " 初始化完成... : " + time) ;
            barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        return time;
    }
}