CountDownLatch和Atomic原子操作類源碼解析

更新時(shí)間：2022年03月11日 15:42:00 作者：Q.E.D

這篇文章主要為大家介紹了CountDownLatch和Atomic原子操作類的源碼解析以及理解應(yīng)用，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步

引導(dǎo)語(yǔ)

本小節(jié)和大家一起來(lái)看看 CountDownLatch 和 Atomic 打頭的原子操作類，CountDownLatch 的源碼非常少，看起來(lái)比較簡(jiǎn)單，但 CountDownLatch 的實(shí)際應(yīng)用卻不是很容易；Atomic 原子操作類就比較好理解和應(yīng)用，接下來(lái)我們分別來(lái)看一下。

1、CountDownLatch

CountDownLatch 中文有的叫做計(jì)數(shù)器，也有翻譯為計(jì)數(shù)鎖，其最大的作用不是為了加鎖，而是通過(guò)計(jì)數(shù)達(dá)到等待的功能，主要有兩種形式的等待：

讓一組線程在全部啟動(dòng)完成之后，再一起執(zhí)行（先啟動(dòng)的線程需要阻塞等待后啟動(dòng)的線程，直到一組線程全部都啟動(dòng)完成后，再一起執(zhí)行）；
主線程等待另外一組線程都執(zhí)行完成之后，再繼續(xù)執(zhí)行。

我們會(huì)舉一個(gè)示例來(lái)演示這兩種情況，但在這之前，我們先來(lái)看看 CountDownLatch 的底層源碼實(shí)現(xiàn)，這樣就會(huì)清晰一點(diǎn)，不然一開(kāi)始就來(lái)看示例，估計(jì)很難理解。

CountDownLatch 有兩個(gè)比較重要的 API，分別是 await 和 countDown，管理著線程能否獲得鎖和鎖的釋放（也可以稱為對(duì) state 的計(jì)數(shù)增加和減少）。

1.1、await

await 我們可以叫做等待，也可以叫做加鎖，有兩種不同入?yún)⒌姆椒?，源碼如下：

public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
// 帶有超時(shí)時(shí)間的，最終都會(huì)轉(zhuǎn)化成毫秒
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

兩個(gè)方法底層使用的都是 sync，sync 是一個(gè)同步器，是 CountDownLatch 的內(nèi)部類實(shí)現(xiàn)的，如下：

private static final class Sync extends AbstractQueuedSynchronizer {}

可以看出來(lái) Sync 繼承了 AbstractQueuedSynchronizer，具備了同步器的通用功能。

無(wú)參 await 底層使用的是 acquireSharedInterruptibly 方法，有參的使用的是 tryAcquireSharedNanos 方法，這兩個(gè)方法都是 AQS 的方法，底層實(shí)現(xiàn)很相似，主要分成兩步：

1.使用子類的 tryAcquireShared 方法嘗試獲得鎖，如果獲取了鎖直接返回，獲取不到鎖走 2；

2.獲取不到鎖，用 Node 封裝一下當(dāng)前線程，追加到同步隊(duì)列的尾部，等待在合適的時(shí)機(jī)去獲得鎖。

第二步是 AQS 已經(jīng)實(shí)現(xiàn)了，第一步 tryAcquireShared 方法是交給 Sync 實(shí)現(xiàn)的，源碼如下：

// 如果當(dāng)前同步器的狀態(tài)是 0 的話，表示可獲得鎖
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

獲得鎖的代碼也很簡(jiǎn)單，直接根據(jù)同步器的 state 字段來(lái)進(jìn)行判斷，但還是有兩點(diǎn)需要注意一下：

獲得鎖時(shí)，state 的值不會(huì)發(fā)生變化，像 ReentrantLock 在獲得鎖時(shí)，會(huì)把 state + 1，但 CountDownLatch 不會(huì)；

CountDownLatch 的 state 并不是 AQS 的默認(rèn)值 0，而是可以賦值的，是在 CountDownLatch 初始化的時(shí)候賦值的，

代碼如下：

// 初始化,count 代表 state 的初始化值
public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    // new Sync 底層代碼是 state = count;
    this.sync = new Sync(count);
}

這里的初始化的 count 和一般的鎖意義不太一樣，count 表示我們希望等待的線程數(shù)，在兩種不同的等待場(chǎng)景中，count 有不同的含義：

讓一組線程在全部啟動(dòng)完成之后，再一起執(zhí)行的等待場(chǎng)景下， count 代表一組線程的個(gè)數(shù)；

主線程等待另外一組線程都執(zhí)行完成之后，再繼續(xù)執(zhí)行的等待場(chǎng)景下，count 代表一組線程的個(gè)數(shù)。

所以我們可以把 count 看做我們希望等待的一組線程的個(gè)數(shù)，可能我們是等待一組線程全部啟動(dòng)完成，可能我們是等待一組線程全部執(zhí)行完成。

1.2、countDown

countDown 中文翻譯為倒計(jì)時(shí)，每調(diào)用一次，都會(huì)使 state 減一，底層調(diào)用的方法如下：

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

releaseShared 是 AQS 定義的方法，方法主要分成兩步：

1.嘗試釋放鎖（tryReleaseShared），鎖釋放失敗直接返回，釋放成功走2

2.釋放當(dāng)前節(jié)點(diǎn)的后置等待節(jié)點(diǎn)。

第二步 AQS 已經(jīng)實(shí)現(xiàn)了，第一步是 Sync 實(shí)現(xiàn)的，我們一起來(lái)看下 tryReleaseShared 方法的實(shí)現(xiàn)源碼：

// 對(duì) state 進(jìn)行遞減，直到 state 變成 0；
// state 遞減為 0 時(shí)，返回 true，其余返回 false
protected boolean tryReleaseShared(int releases) {
    // 自旋保證 CAS 一定可以成功
    for (;;) {
        int c = getState();
        // state 已經(jīng)是 0 了，直接返回 false
        if (c == 0)
            return false;
        // 對(duì) state 進(jìn)行遞減
        int nextc = c-1;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

從源碼中可以看到，只有到 count 遞減到 0 時(shí)，countDown 才會(huì)返回 true。

1.3、示例

看完 CountDownLatch 兩個(gè)重要 API 后，我們來(lái)實(shí)現(xiàn)文章開(kāi)頭說(shuō)的兩個(gè)功能：

讓一組線程在全部啟動(dòng)完成之后，再一起執(zhí)行；

主線程等待另外一組線程都執(zhí)行完成之后，再繼續(xù)執(zhí)行。

代碼在 CountDownLatchDemo 類中，大家可以調(diào)試看看，源碼如下：

public class CountDownLatchDemo {
  // 線程任務(wù)
  class Worker implements Runnable {
    // 定義計(jì)數(shù)鎖用來(lái)實(shí)現(xiàn)功能 1
    private final CountDownLatch startSignal;
    // 定義計(jì)數(shù)鎖用來(lái)實(shí)現(xiàn)功能 2
    private final CountDownLatch doneSignal;
    Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
      this.startSignal = startSignal;
      this.doneSignal = doneSignal;
    }
		// 子線程做的事情
    public void run() {
      try {
        System.out.println(Thread.currentThread().getName()+" begin");
        // await 時(shí)有兩點(diǎn)需要注意：await 時(shí) state 不會(huì)發(fā)生變化，2：startSignal 的state初始化是 1，所以所有子線程都是獲取不到鎖的，都需要到同步隊(duì)列中去等待，達(dá)到先啟動(dòng)的子線程等待后面啟動(dòng)的子線程的結(jié)果
        startSignal.await();
        doWork();
        // countDown 每次會(huì)使 state 減一，doneSignal 初始化為 9，countDown 前 8 次執(zhí)行都會(huì)返回 false (releaseShared 方法)，執(zhí)行第 9 次時(shí)，state 遞減為 0，會(huì) countDown 成功，表示所有子線程都執(zhí)行完了，會(huì)釋放 await 在 doneSignal 上的主線程
        doneSignal.countDown();
        System.out.println(Thread.currentThread().getName()+" end");
      } catch (InterruptedException ex) {
      } // return;
    }
    void doWork() throws InterruptedException {
      System.out.println(Thread.currentThread().getName()+"sleep 5s …………");
      Thread.sleep(5000l);
    }
  }
  @Test
  public void test() throws InterruptedException {
    // state 初始化為 1 很關(guān)鍵，子線程是不斷的 await，await 時(shí) state 是不會(huì)變化的，并且發(fā)現(xiàn) state 都是 1，所有線程都獲取不到鎖
    // 造成所有線程都到同步隊(duì)列中去等待，當(dāng)主線程執(zhí)行 countDown 時(shí)，就會(huì)一起把等待的線程給釋放掉
    CountDownLatch startSignal = new CountDownLatch(1);
    // state 初始化成 9，表示有 9 個(gè)子線程執(zhí)行完成之后，會(huì)喚醒主線程
    CountDownLatch doneSignal = new CountDownLatch(9);
    for (int i = 0; i < 9; ++i) // create and start threads
    {
      new Thread(new Worker(startSignal, doneSignal)).start();
    }
    System.out.println("main thread begin");
    // 這行代碼喚醒 9 個(gè)子線程，開(kāi)始執(zhí)行(因?yàn)?startSignal 鎖的狀態(tài)是 1，所以調(diào)用一次 countDown 方法就可以釋放9個(gè)等待的子線程)
    startSignal.countDown();
    // 這行代碼使主線程陷入沉睡，等待 9 個(gè)子線程執(zhí)行完成之后才會(huì)繼續(xù)執(zhí)行(就是等待子線程執(zhí)行 doneSignal.countDown())
    doneSignal.await();           
    System.out.println("main thread end");
  }
}

執(zhí)行結(jié)果：

Thread-0 begin
Thread-1 begin
Thread-2 begin
Thread-3 begin
Thread-4 begin
Thread-5 begin
Thread-6 begin
Thread-7 begin
Thread-8 begin
main thread begin
Thread-0sleep 5s …………
Thread-1sleep 5s …………
Thread-4sleep 5s …………
Thread-3sleep 5s …………
Thread-2sleep 5s …………
Thread-8sleep 5s …………
Thread-7sleep 5s …………
Thread-6sleep 5s …………
Thread-5sleep 5s …………
Thread-0 end
Thread-1 end
Thread-4 end
Thread-3 end
Thread-2 end
Thread-8 end
Thread-7 end
Thread-6 end
Thread-5 end
main thread end

從執(zhí)行結(jié)果中，可以看出已經(jīng)實(shí)現(xiàn)了以上兩個(gè)功能，實(shí)現(xiàn)比較繞，大家可以根據(jù)注釋，debug 看一看。

2、Atomic 原子操作類

Atomic 打頭的原子操作類有很多，涉及到 Java 常用的數(shù)字類型的，基本都有相應(yīng)的 Atomic 原子操作類，如下圖所示：

圖片描述

Atomic 打頭的原子操作類，在高并發(fā)場(chǎng)景下，都是線程安全的，我們可以放心使用。

我們以 AtomicInteger 為例子，來(lái)看下主要的底層實(shí)現(xiàn)：

private volatile int value;
// 初始化
public AtomicInteger(int initialValue) {
    value = initialValue;
}
// 得到當(dāng)前值
public final int get() {
    return value;
}
// 自增 1，并返回自增之前的值    
public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}
// 自減 1，并返回自增之前的值    
public final int getAndDecrement() {
    return unsafe.getAndAddInt(this, valueOffset, -1);
}

從源碼中，我們可以看到，線程安全的操作方法，底層都是使用 unsafe 方法實(shí)現(xiàn)，以上幾個(gè) unsafe 方法不是使用 Java 實(shí)現(xiàn)的，都是線程安全的。

AtomicInteger 是對(duì) int 類型的值進(jìn)行自增自減，那如果 Atomic 的對(duì)象是個(gè)自定義類怎么辦呢，Java 也提供了自定義對(duì)象的原子操作類，叫做 AtomicReference。AtomicReference 類可操作的對(duì)象是個(gè)泛型，所以支持自定義類，其底層是沒(méi)有自增方法的，操作的方法可以作為函數(shù)入?yún)鬟f，源碼如下：

// 對(duì) x 執(zhí)行 accumulatorFunction 操作
// accumulatorFunction 是個(gè)函數(shù)，可以自定義想做的事情
// 返回老值
public final V getAndAccumulate(V x,
                                BinaryOperator<V> accumulatorFunction) {
    // prev 是老值，next 是新值
    V prev, next;
    // 自旋 + CAS 保證一定可以替換老值
    do {
        prev = get();
        // 執(zhí)行自定義操作
        next = accumulatorFunction.apply(prev, x);
    } while (!compareAndSet(prev, next));
    return prev;
}