淺析Java中并發(fā)工具類的使用

更新時間：2022年12月07日 10:49:48 作者：初念初戀

在JDK的并發(fā)包里提供了幾個非常有用的并發(fā)工具類。CountDownLatch、CyclicBarrier和Semaphore工具類提供了一種并發(fā)流程控制的手段，Exchanger工具類提供了在線程間交換數(shù)據(jù)的一種方法。本文主要介紹了它們的使用，需要的可以參考一下

類	作用
CountDownLatch	線程等待直到計數(shù)器減為0時開始工作
CyclicBarrier	作用跟CountDownLatch類似，但是可以重復(fù)使用
Semaphore	限制線程的數(shù)量
Exchanger	兩個線程交換數(shù)據(jù)

CountDownLatch

概述

CountDownLatch可以使一個或多個線程等待其他線程各自執(zhí)行完畢后再執(zhí)行。

CountDownLatch定義了一個計數(shù)器，和一個阻塞隊列，當(dāng)計數(shù)器的值遞減為0之前，阻塞隊列里面的線程處于掛起狀態(tài)，當(dāng)計數(shù)器遞減到0時會喚醒阻塞隊列所有線程，這里的計數(shù)器是一個標(biāo)志，可以表示一個任務(wù)一個線程，也可以表示一個倒計時器。

案例

玩吃雞游戲的時候，正式開始游戲之前，肯定會加載一些前置場景，例如：“加載地圖”、“加載人物模型”、“加載背景音樂”等。

public class CountDownLatchDemo {
    // 定義前置任務(wù)線程
    static class PreTaskThread implements Runnable {

        private String task;
        private CountDownLatch countDownLatch;

        public PreTaskThread(String task, CountDownLatch countDownLatch) {
            this.task = task;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                Random random = new Random();
                Thread.sleep(random.nextInt(1000));
                System.out.println(task + " - 任務(wù)完成");
                countDownLatch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        // 假設(shè)有三個模塊需要加載
        CountDownLatch countDownLatch = new CountDownLatch(3);

        // 主任務(wù)
        new Thread(() -> {
            try {
                System.out.println("等待數(shù)據(jù)加載...");
                System.out.println(String.format("還有%d個前置任務(wù)", countDownLatch.getCount()));
                countDownLatch.await();
                System.out.println("數(shù)據(jù)加載完成，正式開始游戲！");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 前置任務(wù)
        new Thread(new PreTaskThread("加載地圖數(shù)據(jù)", countDownLatch)).start();
        new Thread(new PreTaskThread("加載人物模型", countDownLatch)).start();
        new Thread(new PreTaskThread("加載背景音樂", countDownLatch)).start();
    }
}

輸出：

等待數(shù)據(jù)加載...
還有3個前置任務(wù)
加載地圖數(shù)據(jù) - 任務(wù)完成
加載人物模型 - 任務(wù)完成
加載背景音樂 - 任務(wù)完成
數(shù)據(jù)加載完成，正式開始游戲！

原理

CountDownLatch的方法很簡單，如下：

// 構(gòu)造方法：
public CountDownLatch(int count)

public void await() // 等待
public boolean await(long timeout, TimeUnit unit) // 超時等待
public void countDown() // count - 1
public long getCount() // 獲取當(dāng)前還有多少count

CountDownLatch構(gòu)造器中的計數(shù)值（count）實際上就是閉鎖需要等待的線程數(shù)量。這個值只能被設(shè)置一次，而且CountDownLatch沒有提供任何機制去重新設(shè)置這個計數(shù)值。

與CountDownLatch的第一次交互是主線程等待其他線程。主線程必須在啟動其他線程后立即調(diào)用CountDownLatch.await()方法。這樣主線程的操作就會在這個方法上阻塞，直到其他線程完成各自的任務(wù)。

其他N 個線程必須引用閉鎖對象，因為他們需要通知CountDownLatch對象，他們已經(jīng)完成了各自的任務(wù)。這種通知機制是通過CountDownLatch.countDown()方法來完成的；每調(diào)用一次這個方法，在構(gòu)造函數(shù)中初始化的count值就減1。所以當(dāng)N個線程都調(diào) 用了這個方法，count的值等于0，然后主線程就能通過await()方法，恢復(fù)執(zhí)行自己的任務(wù)。

源碼分析

CountDownLatch有一個內(nèi)部類叫做Sync，它繼承了AbstractQueuedSynchronizer類，其中維護(hù)了一個整數(shù)state，并且保證了修改state的可見性和原子性，源碼如下：

private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }

        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c - 1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

創(chuàng)建CountDownLatch實例時，也會創(chuàng)建一個Sync的實例，同時把計數(shù)器的值傳給Sync實例，源碼如下：

public CountDownLatch(int count) {
  if (count < 0) throw new IllegalArgumentException("count < 0");
  this.sync = new Sync(count);
}

在countDown方法中，只調(diào)用了Sync實例的releaseShared方法，源碼如下：

public void countDown() {
    sync.releaseShared(1);
}

其中的releaseShared方法，先對計數(shù)器進(jìn)行減1操作，如果減1后的計數(shù)器為0，喚醒被await方法阻塞的所有線程，源碼如下：

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) { //對計數(shù)器進(jìn)行減一操作
        doReleaseShared();//如果計數(shù)器為0，喚醒被await方法阻塞的所有線程
        return true;
    }
    return false;
}

其中的tryReleaseShared方法，先獲取當(dāng)前計數(shù)器的值，如果計數(shù)器為0時，就直接返回；如果不為0時，使用CAS方法對計數(shù)器進(jìn)行減1操作，源碼如下：

protected boolean tryReleaseShared(int releases) {
    for (;;) {//死循環(huán)，如果CAS操作失敗就會不斷繼續(xù)嘗試。
        int c = getState();//獲取當(dāng)前計數(shù)器的值。
        if (c == 0)// 計數(shù)器為0時，就直接返回。
            return false;
        int nextc = c-1;
        if (compareAndSetState(c, nextc))// 使用CAS方法對計數(shù)器進(jìn)行減1操作
            return nextc == 0;//如果操作成功，返回計數(shù)器是否為0
    }
}

在await方法中，只調(diào)用了Sync實例的acquireSharedInterruptibly方法，源碼如下：

public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

其中acquireSharedInterruptibly方法，判斷計數(shù)器是否為0，如果不為0則阻塞當(dāng)前線程，源碼如下：

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)//判斷計數(shù)器是否為0
        doAcquireSharedInterruptibly(arg);//如果不為0則阻塞當(dāng)前線程
}

其中tryAcquireShared方法，是AbstractQueuedSynchronizer中的一個模板方法，其具體實現(xiàn)在Sync類中，其主要是判斷計數(shù)器是否為零，如果為零則返回1，如果不為零則返回-1，源碼如下：

protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

CyclicBarrier

概述

CyclicBarrier 翻譯為中文是循環(huán)（Cyclic）柵欄（Barrier）的意思，它的大概含義是實現(xiàn)一個可循環(huán)利用的屏障。

CyclicBarrier 作用是讓一組線程相互等待，當(dāng)達(dá)到一個共同點時，所有之前等待的線程再繼續(xù)執(zhí)行，且 CyclicBarrier 功能可重復(fù)使用，使用reset()方法重置屏障。

案例

同樣用玩游戲的例子。如果玩一個游戲有多個“關(guān)卡”，那使用CountDownLatch顯然不太合適，那需要為每個關(guān)卡都創(chuàng)建一個實例。那我們可以使用CyclicBarrier來實現(xiàn)每個關(guān)卡的數(shù)據(jù)加載等待功能。

public class CyclicBarrierDemo {
    static class PreTaskThread implements Runnable {

        private String task;
        private CyclicBarrier cyclicBarrier;

        public PreTaskThread(String task, CyclicBarrier cyclicBarrier) {
            this.task = task;
            this.cyclicBarrier = cyclicBarrier;
        }

        @Override
        public void run() {
            // 假設(shè)總共三個關(guān)卡
            for (int i = 1; i < 4; i++) {
                try {
                    Random random = new Random();
                    Thread.sleep(random.nextInt(1000));
                    System.out.println(String.format("關(guān)卡%d的任務(wù)%s完成", i, task));
                    cyclicBarrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
            System.out.println("本關(guān)卡所有前置任務(wù)完成，開始游戲...");
        });

        new Thread(new PreTaskThread("加載地圖數(shù)據(jù)", cyclicBarrier)).start();
        new Thread(new PreTaskThread("加載人物模型", cyclicBarrier)).start();
        new Thread(new PreTaskThread("加載背景音樂", cyclicBarrier)).start();
    }
}

輸出:

關(guān)卡1的任務(wù)加載背景音樂完成
關(guān)卡1的任務(wù)加載地圖數(shù)據(jù)完成
關(guān)卡1的任務(wù)加載人物模型完成
本關(guān)卡所有前置任務(wù)完成，開始游戲...
關(guān)卡2的任務(wù)加載人物模型完成
關(guān)卡2的任務(wù)加載背景音樂完成
關(guān)卡2的任務(wù)加載地圖數(shù)據(jù)完成
本關(guān)卡所有前置任務(wù)完成，開始游戲...
關(guān)卡3的任務(wù)加載背景音樂完成
關(guān)卡3的任務(wù)加載地圖數(shù)據(jù)完成
關(guān)卡3的任務(wù)加載人物模型完成
本關(guān)卡所有前置任務(wù)完成，開始游戲...

與CountDownLatch有一些不同。CyclicBarrier沒有分為await()和countDown()，而是只有單獨的一個await()方法。

一旦調(diào)用await()方法的線程數(shù)量等于構(gòu)造方法中傳入的任務(wù)總量，就代表達(dá)到屏障了。CyclicBarrier允許我們在達(dá)到屏障的時候可以執(zhí)行一個任務(wù)，可以在構(gòu)造方法傳入一個Runnable類型的對象。

源碼分析

構(gòu)造函數(shù)：

public CyclicBarrier(int parties, Runnable barrierAction) {
  if (parties <= 0) throw new IllegalArgumentException();
  this.parties = parties;
  this.count = parties;
  this.barrierCommand = barrierAction;
}

public CyclicBarrier(int parties) {
  this(parties, null);
}

默認(rèn)barrierAction是null，這個參數(shù)是Runnable參數(shù)，當(dāng)最后線程達(dá)到的時候執(zhí)行的任務(wù)，上述案例就是在達(dá)到屏障時，輸出“本關(guān)卡所有前置任務(wù)完成，開始游戲...”。parties 是參與的線程數(shù)。

接著看下await方法，有兩個重載，區(qū)別是是否有等待超時，源碼如下：

 public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

public int await(long timeout, TimeUnit unit)
    throws InterruptedException,
           BrokenBarrierException,
           TimeoutException {
    return dowait(true, unit.toNanos(timeout));
}

重點看下dowait()，核心邏輯就是這個方法，源碼如下：

private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {       
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 每次使用屏障都會生成一個實例
            final Generation g = generation;

            // 如果被破壞了就拋異常
            if (g.broken)
                throw new BrokenBarrierException();

            // 線程中斷檢測
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }

            // 剩余的等待線程數(shù)
            int index = --count;
            // 最后線程到達(dá)時 
            if (index == 0) {  // tripped
                // 標(biāo)記任務(wù)是否被執(zhí)行（就是傳進(jìn)入的runable參數(shù)）
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    // 執(zhí)行任務(wù)
                    if (command != null)
                        command.run();
                    ranAction = true;
                    // 完成后 進(jìn)行下一組 初始化 generation 初始化 count 并喚醒所有等待的線程 
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }

            // index 不為0時 進(jìn)入自旋
            for (;;) {
                try {
                    // 先判斷超時 沒超時就繼續(xù)等著
                    if (!timed)
                        trip.await();
                        // 如果超出指定時間 調(diào)用 awaitNanos 超時了釋放鎖
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                        // 中斷異常捕獲
                } catch (InterruptedException ie) {
                    // 判斷是否被破壞
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        // 否則的話中斷當(dāng)前線程
                        Thread.currentThread().interrupt();
                    }
                }

                // 被破壞拋異常
                if (g.broken)
                    throw new BrokenBarrierException();

                // 正常調(diào)用 就返回 
                if (g != generation)
                    return index;

                // 超時了而被喚醒的情況 調(diào)用 breakBarrier()
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

總結(jié)下dowait()方法的邏輯：

線程調(diào)用后，會檢查barrier的狀態(tài)、線程狀態(tài)，異常狀態(tài)會中斷。
在初始化CyclicBarrier時，設(shè)置的資源值count，會進(jìn)行--count。
當(dāng)10個線程中前9個線程，執(zhí)行dowait()后，由于count!=0，因此會進(jìn)行for(;;)，在內(nèi)部會執(zhí)行Condition的trip.await()方法，進(jìn)行阻塞。
阻塞結(jié)束的條件有：超時、被喚醒、線程中斷。
當(dāng)?shù)?0個線程執(zhí)行dowait()后，由于count==0，會先檢查并執(zhí)行command的內(nèi)容。
最后執(zhí)行nextGeneration()，在內(nèi)部調(diào)用trip.signalAll()喚醒所有trip.await()的線程。

如果被破壞了怎么恢復(fù)呢？來看下reset()方法，源碼如下：

public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        breakBarrier();   // break the current generation
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}

源碼很簡單，break之后重新生成新的實例，對應(yīng)的會重新初始化count，在dowait里index==0也調(diào)用了nextGeneration，所以說它是可以循環(huán)利用的。

與CountDonwLatch的區(qū)別

CountDownLatch減計數(shù)，CyclicBarrier加計數(shù)。

CountDownLatch是一次性的，CyclicBarrier可以重用。

CountDownLatch和CyclicBarrier都有讓多個線程等待同步然后再開始下一步動作的意思，但是CountDownLatch的下一步的動作實施者是主線程，具有不可重復(fù)性；而CyclicBarrier的下一步動作實施者還是“其他線程”本身，具有往復(fù)多次實施動作的特點。

Semaphore

概述

Semaphore 一般譯作信號量，它也是一種線程同步工具，主要用于多個線程對共享資源進(jìn)行并行操作的一種工具類。它代表了一種許可的概念，是否允許多線程對同一資源進(jìn)行操作的許可，使用 Semaphore 可以控制并發(fā)訪問資源的線程個數(shù)。

使用場景

Semaphore 的使用場景主要用于流量控制。

比如數(shù)據(jù)庫連接，同時使用的數(shù)據(jù)庫連接會有數(shù)量限制，數(shù)據(jù)庫連接不能超過一定的數(shù)量，當(dāng)連接到達(dá)了限制數(shù)量后，后面的線程只能排隊等前面的線程釋放數(shù)據(jù)庫連接后才能獲得數(shù)據(jù)庫連接。

比如停車場的場景中，一個停車場有有限數(shù)量的車位，同時能夠容納多少臺車，車位滿了之后只有等里面的車離開停車場外面的車才可以進(jìn)入。

案例

模擬一下停車場的業(yè)務(wù)場景：

在進(jìn)入停車場之前會有一個提示牌，上面顯示著停車位還有多少，當(dāng)車位為 0 時，不能進(jìn)入停車場，當(dāng)車位不為 0 時，才會允許車輛進(jìn)入停車場。所以停車場有幾個關(guān)鍵因素：停車場車位的總?cè)萘?，?dāng)一輛車進(jìn)入時，停車場車位的總?cè)萘?- 1，當(dāng)一輛車離開時，總?cè)萘?+ 1，停車場車位不足時，車輛只能在停車場外等待。

public class SemaphoreDemo {

    private static Semaphore semaphore = new Semaphore(10);

    public static void main(String[] args) {
        for (int i = 0; i < 100; i++) {
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("歡迎 " + Thread.currentThread().getName() + " 來到停車場");
                    // 判斷是否允許停車
                    if (semaphore.availablePermits() == 0) {
                        System.out.println("車位不足，請耐心等待");
                    }
                    try {
                        // 嘗試獲取
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + " 進(jìn)入停車場");
                        Thread.sleep(new Random().nextInt(10000));// 模擬車輛在停車場停留的時間
                        System.out.println(Thread.currentThread().getName() + " 駛出停車場");
                        semaphore.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, i + "號車");
            thread.start();
        }
    }
}

Semaphore 的初始容量，也就是只有 10 個車位，我們用這 10 個車位來控制 100 輛車的流量，所以結(jié)果和我們預(yù)想的很相似，即大部分車都在等待狀態(tài)。但是同時仍允許一些車駛?cè)胪＼噲?，駛?cè)胪＼噲龅能囕v，就會 semaphore.acquire 占用一個車位，駛出停車場時，就會 semaphore.release 讓出一個車位，讓后面的車再次駛?cè)搿?/p>

原理

Semaphore內(nèi)部有一個繼承了AQS的同步器Sync，重寫了tryAcquireShared方法。在這個方法里，會去嘗試獲取資源。

如果獲取失?。ㄏ胍馁Y源數(shù)量小于目前已有的資源數(shù)量），就會返回一個負(fù)數(shù)（代表嘗試獲取資源失敗）。然后當(dāng)前線程就會進(jìn)入AQS的等待隊列。

Exchanger

概述

Exchanger類用于兩個線程交換數(shù)據(jù)。它支持泛型，也就是說你可以在兩個線程之間傳送任何數(shù)據(jù)。一個線程在完成一定的事務(wù)后想與另一個線程交換數(shù)據(jù)，則第一個先拿出數(shù)據(jù)的線程會一直等待第二個線程，直到第二個線程拿著數(shù)據(jù)到來時才能彼此交換對應(yīng)數(shù)據(jù)。

案例

案例1：A同學(xué)和B同學(xué)交換各自收藏的大片。

public class ExchangerDemo {

    public static void main(String[] args) throws InterruptedException {
        Exchanger<String> stringExchanger = new Exchanger<>();

        Thread studentA = new Thread(() -> {
            try {
                String dataA = "A同學(xué)收藏多年的大片";
                String dataB = stringExchanger.exchange(dataA);
                System.out.println("A同學(xué)得到了" + dataB);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        System.out.println("這個時候A同學(xué)是阻塞的，在等待B同學(xué)的大片");
        Thread.sleep(1000);

        Thread studentB = new Thread(() -> {
            try {
                String dataB = "B同學(xué)收藏多年的大片";
                String dataA = stringExchanger.exchange(dataB);
                System.out.println("B同學(xué)得到了" + dataA);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        studentA.start();
        studentB.start();
    }
}

輸出：

這個時候A同學(xué)是阻塞的，在等待B同學(xué)的大片
A同學(xué)得到了B同學(xué)收藏多年的大片
B同學(xué)得到了A同學(xué)收藏多年的大片

可以看到，當(dāng)一個線程調(diào)用exchange方法后，它是處于阻塞狀態(tài)的，只有當(dāng)另一個線程也調(diào)用了exchange方法，它才會繼續(xù)向下執(zhí)行。

Exchanger類還有一個有超時參數(shù)的方法，如果在指定時間內(nèi)沒有另一個線程調(diào)用exchange，就會拋出一個超時異常。

public V exchange(V x, long timeout, TimeUnit unit)

案例2：A同學(xué)被放鴿子，交易失敗。

public class ExchangerDemo {

    public static void main(String[] args) {
        Exchanger<String> stringExchanger = new Exchanger<>();
        Thread studentA = new Thread(() -> {
            String dataB = null;
            try {
                String dataA = "A同學(xué)收藏多年的大片";
                dataB = stringExchanger.exchange(dataA,5, TimeUnit.SECONDS);
                System.out.println("A同學(xué)得到了" + dataB);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                System.out.println("等待超時-TimeoutException");
            }
            System.out.println("A同學(xué)得到了："+dataB);
        });

        studentA.start();
    }
}

輸出：