Java之CountDownLatch原理全面解析

更新時(shí)間：2022年10月28日 09:13:24 作者：阿拉的夢(mèng)想

這篇文章主要介紹了Java之CountDownLatch原理解析，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

CountDownLatch原理解析

1. demo展示

代碼邏輯展示了主線程中創(chuàng)建2個(gè)子線程分別去執(zhí)行任務(wù)，主線程等2個(gè)子線程執(zhí)行完畢后，再接著執(zhí)行下面的代碼；

常用場景：

分別計(jì)算，匯總結(jié)果。如，多個(gè)線程分別解析excel中的sheet,等待全部解析完畢后匯總結(jié)果；

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class CountDownLatchDemo {
    //定義一個(gè)倒計(jì)時(shí)閂鎖
    static CountDownLatch c = new CountDownLatch(2);

    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            try {
                TimeUnit.MICROSECONDS.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("我是線程1");
            //釋放一個(gè)
            c.countDown();
        }).start();

        new Thread(() -> {
            try {
                TimeUnit.MICROSECONDS.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("我是線程2");
            //釋放一個(gè)
            c.countDown();
        }).start();

        System.out.println("我是主線程，我要等那兩個(gè)線程執(zhí)行完畢...");
        //等待倒計(jì)時(shí)為0
        c.await();
        System.out.println("我是主線程,那兩個(gè)線程都執(zhí)行完了");
    }
}

輸出：

我是主線程，我要等那兩個(gè)線程執(zhí)行完畢...

我是線程2

我是線程1

我是主線程,那兩個(gè)線程都執(zhí)行完了

2. 原理解析

1.先看構(gòu)造函數(shù)new CountDownLatch(2)做了什么？

這是初始化了AQS子類，并將AQS的狀態(tài)state設(shè)置為傳入的2；

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

2.看c.countDown()做了什么？

它釋放了一個(gè)共享鎖狀態(tài)，也就是state減1；

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

3.再看c.await()做了什么？

await方法是CounDownLatch中定義的，它調(diào)用了其內(nèi)部類Sync（也是AQS的子類）的獲取共享鎖的方法acquireSharedInterruptibly；

acquireSharedInterruptibly方法中調(diào)用了CountDownLatch內(nèi)部類Sync中實(shí)現(xiàn)的獲取共享鎖的方法tryAcquireShared，返回值不小0就算獲取到了鎖，await方法就能返回了，如果返回值小于0將會(huì)進(jìn)入阻塞等待;

CountDownLatch內(nèi)部類Sync中tryAcquireShared的實(shí)現(xiàn)很簡單，只要state=0就返回1，否則返回-1；上面說了返回一個(gè)不小于0的數(shù)字，c.await()就相當(dāng)于獲取到了鎖，就可以返回了，主線程就可以繼續(xù)執(zhí)行了。

通過上面分析，每次c.countDown(),就會(huì)將state減1，state=0的時(shí)候主線程恢復(fù)執(zhí)行；

Java CountDownLatch學(xué)習(xí)總結(jié)

來源包

同為 java.util.concurrent 下的，即也是并發(fā)多線程相關(guān)下的類，直譯 “倒計(jì)時(shí)鎖存器”，一般用于多線程場景，單一的線程也可以，用于等待多個(gè)任務(wù)完成后再執(zhí)行其他操作；

提供方法

await()

導(dǎo)致當(dāng)前線程等待，直到鎖存器倒數(shù)到零，除非該線程是｛@linkplain Thread35;interrupt interrupted｝即被打斷狀態(tài)。
如果當(dāng)前計(jì)數(shù)為零，則此方法立即返回。
如果當(dāng)前計(jì)數(shù)大于零，則當(dāng)前線程將出于線程調(diào)度目的被禁用，并處于休眠狀態(tài)，直到發(fā)生以下兩種情況之一：
由于調(diào)用{@link#countDown}方法，計(jì)數(shù)達(dá)到零；或者其他線程{@linkplain thread#中斷}當(dāng)前線程。

如果當(dāng)前線程：

在進(jìn)入此方法時(shí)設(shè)置了其中斷狀態(tài)；或者
在等待時(shí){@linkplain Thread#interrupt interrupted}，
則拋出{@link InterruptedException}，并清除當(dāng)前線程的中斷狀態(tài)。

簡單說就是當(dāng)使用了這個(gè)方法后當(dāng)前這一個(gè)線程將進(jìn)入等待狀態(tài)，直到計(jì)數(shù)器被減到0或者當(dāng)前線程被中斷，計(jì)數(shù)器被減到0后，所有等待的線程將被喚醒繼續(xù)向下執(zhí)行

await(long timeout, TimeUnit unit)

同上，但是指定了等待的超時(shí)時(shí)間，即線程除了上方兩種被喚醒的情況下，等待到超時(shí)時(shí)間后也會(huì)被喚醒

countDown()：當(dāng)前計(jì)數(shù)器減一，如果如果減到 0 則喚醒所有等待在這個(gè) CountDownLatch 上的線程。
getCount()：獲取當(dāng)前計(jì)數(shù)的數(shù)值

業(yè)務(wù)書寫示例

即將需要一會(huì)兒處理的業(yè)務(wù) list 設(shè)置為計(jì)數(shù)器的大小，

然后對(duì)里面的業(yè)務(wù)數(shù)據(jù)執(zhí)行異步操作，處理業(yè)務(wù)過程中不論是否有異常都需要對(duì)計(jì)數(shù)器減一，最終使用 await 等待所有任務(wù)執(zhí)行完成，執(zhí)行完成后，將進(jìn)入后續(xù)處理

? ? ? ? ? ? final CountDownLatch latch = new CountDownLatch(lists.size());
? ? ? ? ? ??
? ? ? ? ? ? for (List<JSONObject> item: lists) {
? ? ? ? ? ? ? ? executor.submit(new Runnable() {
? ? ? ? ? ? ? ? ? ? @Override
? ? ? ? ? ? ? ? ? ? public void run() {
? ? ? ? ? ? ? ? ? ? ? ? // ....... 業(yè)務(wù)處理
? ? ? ? ? ? ? ? ? ? ? ? } catch (Exception e) {
? ? ? ? ? ? ? ? ? ? ? ? ? ? // 異常處理
? ? ? ? ? ? ? ? ? ? ? ? } finally {
? ? ? ? ? ? ? ? ? ? ? ? ? ? latch.countDown();
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? });
? ? ? ? ? ? }

? ? ? ? ? ? try {
? ? ? ? ? ? ? ? latch.await();
? ? ? ? ? ? } catch (InterruptedException e) {
? ? ? ? ? ? ? ? log.error("線程被中斷", e);
? ? ? ? ? ? }

? ? ? ?// lists 處理完成后的其他業(yè)務(wù)操作

一般代碼示例

public static void main(String[] args) throws InterruptedException {
? ? ? ? final CountDownLatch downLatch = new CountDownLatch(3);

? ? ? ? Await wait111 = new Await("wait111", downLatch);
? ? ? ? Await wait222 = new Await("wait222", downLatch);
? ? ? ? CountDownStart countDownStart = new CountDownStart(downLatch);

? ? ? ? wait111.start();
? ? ? ? wait222.start();
? ? ? ? Thread.sleep(1000);
? ? ? ? countDownStart.run();

? ? }


class Await extends Thread{

? ? private CountDownLatch countDownLatch;
? ? private String name;

? ? public Await(String name, CountDownLatch countDownLatch){
? ? ? ? this.name = name;
? ? ? ? this.countDownLatch = countDownLatch;
? ? }


? ? @Override
? ? public void run() {
? ? ? ? System.out.println(name + " start.....");
? ? ? ? System.out.println(name + " run.....");
? ? ? ? try {
? ? ? ? ? ? countDownLatch.await();
? ? ? ? } catch (InterruptedException e) {
? ? ? ? ? ? e.printStackTrace();
? ? ? ? }
? ? ? ? System.out.println(name + " continue.....run");
? ? }
}

class CountDownStart extends Thread{

? ? private CountDownLatch countDownLatch;

? ? public CountDownStart(CountDownLatch countDownLatch){
? ? ? ? this.countDownLatch = countDownLatch;
? ? }

? ? @Override
? ? public void run() {
? ? ? ? countDownLatch.countDown();
? ? ? ? countDownLatch.countDown();
? ? ? ? countDownLatch.countDown();
? ? ? ? System.out.println("start countDown");
? ? }
}

運(yùn)行結(jié)果：

wait222 start.....
wait222 run.....
wait111 start.....
wait111 run.....
start countDown
wait111 continue.....run
wait222 continue.....run

但是當(dāng)我把線程等待去除后：

? ? public static void main(String[] args) throws InterruptedException {
? ? ? ? final CountDownLatch downLatch = new CountDownLatch(3);

? ? ? ? Await wait111 = new Await("wait111", downLatch);
? ? ? ? Await wait222 = new Await("wait222", downLatch);
? ? ? ? CountDownStart countDownStart = new CountDownStart(downLatch);

? ? ? ? wait111.start();
? ? ? ? wait222.start();
// ? ? ? ?Thread.sleep(1000);
? ? ? ? countDownStart.run();

? ? }

結(jié)果：