2. 程式人生 > >高併發第十單:J.U.C AQS 元件:CountDownLatch. CyclicBarrier .Semaphore

高併發第十單:J.U.C AQS 元件:CountDownLatch. CyclicBarrier .Semaphore

阿新 • • 發佈：2018-12-11

AQS全名：AbstractQueuedSynchronizer，是併發容器J.U.C（java.lang.concurrent）下locks包內的一個類。它實現了一個FIFO(FirstIn、FisrtOut先進先出)的佇列。底層實現的資料結構是一個雙向列表。

Sync queue：同步佇列，是一個雙向列表。包括head節點和tail節點。head節點主要用作後續的排程。

Condition queue：非必須，單向列表。當程式中存在cindition的時候才會存在此列表。

AQS全名：AbstractQueuedSynchronizer，是併發容器J.U.C（java.lang.concurrent）下locks包內的一個類。它實現了一個FIFO(FirstIn、FisrtOut先進先出)的佇列。底層實現的資料結構是一個雙向列表。

　它維護了一個volatile int state（代表共享資源）和一個FIFO執行緒等待佇列（多執行緒爭用資源被阻塞時會進入此佇列）。這裡volatile是核心關鍵詞，具體volatile的語義，在此不述。state的訪問方式有三種:

  • getState()
  • setState()
  • compareAndSetState()

這裡就不詳細去說AQS了.因為開頭的那個文章已經說得很清楚了.介紹以下他的實現類吧

1.CountDownLatch(計數器)

CountDownLatch是在java1.5被引入的，它都存在於java.util.concurrent包下。CountDownLatch這個類能夠使一個執行緒等待其他執行緒完成各自的工作後再執行。例如，應用程式的主執行緒希望在負責啟動框架服務的執行緒已經啟動所有的框架服務之後再執行。CountDownLatch是通過一個計數器來實現的，計數器的初始值為執行緒的數量。每當一個執行緒完成了自己的任務後，計數器的值就會減1。當計數器值到達0時，它表示所有的執行緒已經完成了任務，然後在閉鎖上等待的執行緒就可以恢復執行任務。

主要在實時系統中的使用場景

  1. 實現最大的並行性：有時我們想同時啟動多個執行緒，實現最大程度的並行性。例如，我們想測試一個單例類。如果我們建立一個初始計數為1的CountDownLatch，並讓所有執行緒都在這個鎖上等待，那麼我們可以很輕鬆地完成測試。我們只需呼叫 一次countDown()方法就可以讓所有的等待執行緒同時恢復執行。
  2. 開始執行前等待n個執行緒完成各自任務：例如應用程式啟動類要確保在處理使用者請求前，所有N個外部系統已經啟動和運行了。
  3. 死鎖檢測：一個非常方便的使用場景是，你可以使用n個執行緒訪問共享資源，在每次測試階段的執行緒數目是不同的，並嘗試產生死鎖。
  4. 有一個任務想要往下執行，但必須要等到其他的任務執行完畢後才可以繼續往下執行 (最常用的)

構造方法：     

構造一個用給定計數初始化的 CountDownLatch(int count)     

普通方法：  

void await()            使當前執行緒在鎖存器倒計數至零之前一直等待，除非執行緒被中斷。

boolean await(long timeout, TimeUnit unit)            可以設定等待的時間，如果超過此時間，計數器還未清零，則不繼續等待

void countDown()            遞減鎖存器的計數，如果計數到達零，則釋放所有等待的執行緒 

 long getCount()            返回當前計數

看個例子

public class CountDownLatchDemo {

    private static final int THREAD_COUNT_NUM = 6;
    private static CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT_NUM);

    public static void main(String[] args) throws InterruptedException {

        for (int i = 0; i < 6; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "第" + index + "個任務完成！"
                            + Thread.currentThread().getName());
                    // 模擬完成一個任務,隨機模擬不同的尋找時間
                    Thread.sleep(3000);
                    System.out.println(Thread.currentThread().getName() + "完成:" + System.currentTimeMillis());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 每完成一個任務,需要等待的任務數減1
                // countDownLatch.countDown();

            }, "我是執行緒:" + i + ":").start();
            ;
        }
        // 等待檢查，即上述7個執行緒執行完畢之後，執行await後邊的程式碼
        // countDownLatch.await();
        System.out.println("所有任務完成!" + System.currentTimeMillis());

    }
}結果:

所有任務完成!1537617901498我是執行緒:3:第3個任務完成！我是執行緒:3:我是執行緒:0:第0個任務完成！我是執行緒:0:我是執行緒:1:第1個任務完成！我是執行緒:1:我是執行緒:2:第2個任務完成！我是執行緒:2:我是執行緒:4:第4個任務完成！我是執行緒:4:我是執行緒:5:第5個任務完成！我是執行緒:5:

我是執行緒:2:完成:1537617904499我是執行緒:1:完成:1537617904499我是執行緒:0:完成:1537617904499我是執行緒:3:完成:1537617904499我是執行緒:4:完成:1537617904499我是執行緒:5:完成:1537617904499

加上:

public class CountDownLatchDemo {

    private static final int THREAD_COUNT_NUM = 6;
    private static CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT_NUM);

    public static void main(String[] args) throws InterruptedException {

        for (int i = 0; i < 6; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "第" + index + "個任務完成！"
                            + Thread.currentThread().getName());
                    // 模擬完成第i個任務,
                    Thread.sleep(3000);
                    System.out.println(Thread.currentThread().getName() + "完成:" + System.currentTimeMillis());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 每完成一個任務,需要等待的任務數減1
                 countDownLatch.countDown();

            }, "我是執行緒:" + i + ":").start();
            ;
        }
        // 等待檢查，即上述7個執行緒執行完畢之後，執行await後邊的程式碼
         countDownLatch.await();
        System.out.println("所有任務完成!" + System.currentTimeMillis());

    }
}結果:

我是執行緒:2:第2個任務完成！我是執行緒:2:我是執行緒:1:第1個任務完成！我是執行緒:1:我是執行緒:0:第0個任務完成！我是執行緒:0:我是執行緒:3:第3個任務完成！我是執行緒:3:我是執行緒:4:第4個任務完成！我是執行緒:4:我是執行緒:5:第5個任務完成！我是執行緒:5:我是執行緒:1:完成:1537617977385我是執行緒:3:完成:1537617977385我是執行緒:4:完成:1537617977385我是執行緒:0:完成:1537617977385我是執行緒:2:完成:1537617977385我是執行緒:5:完成:1537617977387所有任務完成!1537617977388

順序有可能發生變化.但是  所有任務完成!1537617977388 時間肯定在他們之後,速度快最多一樣.肯定不會比他們小

這裡有個例子 執行緒數 大於 鎖住數時 會發生什麼呢. 

public class CountDownLatchDemo {

    private static final int THREAD_COUNT_NUM = 8;
    private static CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT_NUM);

    public static void main(String[] args) throws InterruptedException {

        for (int i = 0; i < 100; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "第" + index + "個任務完成！"
                            + Thread.currentThread().getName());
                    // 模擬收集第i個龍珠,隨機模擬不同的尋找時間
                    Thread.sleep(3000);
                    System.out.println(Thread.currentThread().getName() + "完成:" + System.currentTimeMillis());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 每收集到一顆龍珠,需要等待的顆數減1
                 countDownLatch.countDown();

            }, "我是執行緒:" + i + ":").start();
            ;
        }
        // 等待檢查，即上述7個執行緒執行完畢之後，執行await後邊的程式碼
         countDownLatch.await();
        System.out.println("所有任務完成!" + System.currentTimeMillis());

    }
}

結果
...............
所有任務完成!1537618986091
............
我是執行緒:39:完成:1537618986097
我是執行緒:37:完成:1537618986097
我是執行緒:38:完成:1537618986097
我是執行緒:35:完成:1537618986096
View Code

所以更加證明了.

CountDownLatch(THREAD_COUNT_NUM); 最多鎖住 THREAD_COUNT_NUM 個的執行緒,其他的執行緒就按原來的順序運行了

這個就直接證明了  在await()處,讓所有的任務完成了 才能繼續主執行緒

優點：

CountDownLatch的優點毋庸置疑，對使用者而言，你只需要傳入一個int型變數控制任務數量即可，至於同步佇列的出隊入隊維護，state變數值的維護對使用者都是透明的，使用方便。

缺點：

CountDownLatch設定了state後就不能更改，也不能迴圈使用。

2.CyclicBarrier 

既然說了 CountDownLatch設定了state後就不能更改，也不能迴圈使用。那就來個可以迴圈使用的

舉個例子:有四個遊戲玩家玩遊戲，遊戲有三個關卡，每個關卡必須要所有玩家都到達後才能允許通過。其實這個場景裡的玩家中如果有玩家A先到了關卡1，他必須等到其他所有玩家都到達關卡1時才能通過，也就是說執行緒之間需要相互等待。這和CountDownLatch的應用場景有區別，CountDownLatch裡的執行緒是到了執行的目標後繼續幹自己的其他事情，而這裡的執行緒需要等待其他執行緒後才能繼續完成下面的工作。

案例一: 一起等待

public class CyclicBarrierDemo {

    private static CyclicBarrier barrier = new CyclicBarrier(5);

    public static void main(String[] args) {

        for (int i = 0; i < 10; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println(String.format("我是第%s啟動了", index));
                    barrier.await();
                    System.out.println(String.format("我是第%s完成了", index));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }

            }, "我是第" + index + "個執行緒:").start();
        }

    }
}

結果:
我是第6啟動了
.......
我是第0啟動了
.......
我是第4完成了了

全部啟動,然後一起等待,再繼續完成任務

//案例二 最多等待時間

    private static void test2() {
        for (int i = 0; i < 10; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    System.out.println(String.format("我是第%s啟動了", index));
                    // 最多阻塞時間
                    barrier.await(2000, TimeUnit.MILLISECONDS);
                    System.out.println(String.format("我是第%s完成了", index));
                } catch (InterruptedException | BrokenBarrierException | TimeoutException e) {
                    e.printStackTrace();
                    barrier.reset();
                }

            }).start();
        }

    }

// 還有一個額外的方法是  構造是可以多構造一個Runnable，在計數器的值到達設定值後（但在釋放所有執行緒之前），該Runnable執行一次，注，Runnable在每個屏障點只執行一個 

private static CyclicBarrier barrier = new CyclicBarrier(1,()->{
        System.out.println("優先執行我");
    });

for (int i = 0; i < 2; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println(String.format("我是第%s啟動了", index));
                    barrier.await();
                    System.out.println(String.format("我是第%s完成了", index));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }

            }, "我是第" + index + "個執行緒:").start();
        }


結果是:
我是第1啟動了
我是第0啟動了
優先執行自己
優先執行自己
我是第1完成了
我是第0完成了
CyclicBarrier 和 CountDownLatch的比較:
  • CountDownLatch: 一個執行緒(或者多個)， 等待另外N個執行緒完成某個事情之後才能執行。--> 反正 你執行完  就ok.不能隨意放開
  • CyclicBarrier: N個執行緒相互等待，任何一個執行緒完成之前，所有的執行緒都必須等待。--> 可以到到某個條件.我放開就行了
  • CountDownLatch的計數器只能使用一次。而CyclicBarrier的計數器可以使用reset() 方法重置。所以CyclicBarrier能處理更為複雜的業務場景，比如如果計算髮生錯誤，可以重置計數器，並讓執行緒們重新執行一次。
  • CountDownLatch：減計數方式，CyclicBarrier：加計數方式

3. Semaphore

訊號量(Semaphore)，有時被稱為訊號燈，是在多執行緒環境下使用的一種設施, 它負責協調各個執行緒, 以保證它們能夠正確、合理的使用公共資源。

比喻:

　　Semaphore是一件可以容納N人的房間，如果人不滿就可以進去，如果人滿了，就要等待有人出來。對於N=1的情況，稱為binary semaphore。一般的用法是，用於限制對於某一資源的同時訪問 

官方一點就是:

用於保證同一時間併發訪問執行緒的數目。

訊號量在作業系統中是很重要的概念，Java併發庫裡的Semaphore就可以很輕鬆的完成類似作業系統訊號量的控制。

Semaphore可以很容易控制系統中某個資源被同時訪問的執行緒個數。 在資料結構中我們學過連結串列，連結串列正常是可以儲存無限個節點的，而Semaphore可以實現有限大小的列表。

使用場景：僅能提供有限訪問的資源。比如資料庫連線

上例子:

// 方式 一 直接獲取

// 給出10個資源 ,最多保證10個併發
private static final Semaphore SEMAPHORE = new Semaphore(10);
....... 
for (int i = 0; i < 100; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    SEMAPHORE.acquire();// 獲取一個許可
                    System.out.println(String.format("我是執行緒:%s", index));// 需要併發控制的內容
                    Thread.sleep(3000);
                    SEMAPHORE.release(); // 釋放一個許可
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }).start();
        }

//結果:
很明顯的能看到 10個10的執行

// 方式 二 嘗試獲取許可，獲取不到不執行  很多時候相當於只執行設定的併發量一次

private static final Semaphore SEMAPHORE = new Semaphore(10);

for (int i = 0; i < 100; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    // 嘗試獲取許可，獲取不到不執行
                    if(SEMAPHORE.tryAcquire() {
                        System.out.println(String.format("我是執行緒:%s", index));// 需要併發控制的內容
                        Thread.sleep(3000);
                        SEMAPHORE.release(); // 釋放一個許可
                    }
                    
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }).start();

 // 當時三  有個最長申請時間 

private static final Semaphore SEMAPHORE = new Semaphore(10);
for (int i = 0; i < 100; i++) {
            int index = i;
            new Thread(() -> {
                try {
                    //  嘗試獲取許可，獲取不到不執行 最長申請時間
                    if(SEMAPHORE.tryAcquire(5000,TimeUnit.MILLISECONDS)) {
                        System.out.println(String.format("我是執行緒:%s", index));// 需要併發控制的內容
                        Thread.sleep(3000);
                        SEMAPHORE.release(); // 釋放一個許可
                    }
                    
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }).start();
        }

注意:

1 . 其中  構造方法可以加公平鎖   :private static final Semaphore SEMAPHORE = new Semaphore(100,true);

2. SEMAPHORE.tryAcquire()  => 可以增加獲取條件量 SEMAPHORE.tryAcquire(10);釋放 SEMAPHORE.release(10);

相關推薦

高併發第十單:J.U.C AQS 元件:CountDownLatch. CyclicBarrier .Semaphore

AQS全名：AbstractQueuedSynchronizer，是併發容器J.U.C（java.lang.concurrent）下locks包內的一個類。它實現了一個FIFO(FirstIn、FisrtOut先進先出)的佇列。底層實現的資料結構是一個雙向列表。 Sync queue：同步佇列，是一個雙

高併發第八彈:J.U.C起航(java.util.concurrent)

java.util.concurrent是JDK自帶的一個併發的包主要分為以下5部分： 併發工具類(tools) 顯示鎖(locks) 原子變數類(aotmic) 併發集合(collections) Executor執行緒執行器 我們今天就說說 併發集合,除開 Queue,放線上程池的時候講 先介紹以下

高併發第十三彈:J.U.C 佇列 SynchronousQueue.ArrayBlockingQueue.LinkedBlockingQueue.LinkedTransferQueue

因為下一節會說執行緒池,要用執行緒池 那麼執行緒池有個很重要的引數 就是Queue的選擇 常用的佇列其實就兩種: 　　先進先出（FIFO）：先插入的佇列的元素也最先出佇列，類似於排隊的功能。從某種程度上來說這種佇列也體現了一種公平性。　　後進先出（LIFO）：後插入佇列的元素最先出佇列，這種佇列優先處理最近發

慕課網實戰·高併發探索（十二）：併發容器J.U.C -- AQS元件 鎖：ReentrantLock、ReentrantReadWriteLock、StempedLock

特別感謝：慕課網jimin老師的《Java併發程式設計與高併發解決方案》課程，以下知識點多數來自老師的課程內容。 jimin老師課程地址：Java併發程式設計與高併發解決方案 ReentrantLock java中有兩類鎖，一類是Synchron

高併發第十一彈:J.U.C -AQS(AbstractQueuedSynchronizer) 元件:Lock,ReentrantLock,ReentrantReadWriteLock,StampedLock

既然說到J.U.C 的AQS(AbstractQueuedSynchronizer)   不說 Lock 是不可能的.不過實話來說,一般 JKD8 以後我一般都不用Lock了.畢竟sychronized 的效率已經很高了.Lock在我的實際開發中的需求很少,但還是需要了解一下的. JAVA的兩種鎖 Reent

高併發第十二彈:併發容器J.U.C -- Executor元件FutureTask、ForkJoin

從本章開始就要說 Executor 的東西了.本次講的是一個很常用的FutureTask,和一個不是那麼常用的ForkJoin,我們現在就來介紹吧 引言 　　大部分時候建立執行緒的2種方式，一種是直接繼承Thread，另外一種就是實現Runnable介面。但是這兩種方式都有一個缺陷就是：在執行完任務之後無法獲

高並發第八彈:J.U.C起航(java.util.concurrent)

思想 依賴 分享圖片 變量 ... 查找 沖突 dex http java.util.concurrent是JDK自帶的一個並發的包主要分為以下5部分： 並發工具類(tools) 顯示鎖(locks) 原子變量類(aotmic) 並發集合(collections) Exe

JAVA學習筆記（併發程式設計 - 柒）- J.U.C元件2

J.U.C-FutureTask 在Java中一般通過繼承Thread類或者實現Runnable介面這兩種方式來建立執行緒，但是這兩種方式都有個缺陷，就是不能在執行完成後獲取執行的結果，因此Java 1.5之後提供了Callable和Future介面，通過它們就可以在任務執行完畢之後得到

JAVA學習筆記（併發程式設計 - 陸）- J.U.C之AQS及其相關元件詳解

文章目錄 J.U.C之AQS-介紹 關於AQS裡的state狀態： 關於自定義資源共享方式： 關於同步器設計： 如何使用： 具體實現的思路： 設計思想： 基於AQS的同步元件： AQS小結：

高併發第十四彈：執行緒池的介紹及使用

單執行緒就不說了因為簡單，並且 在實際的生產環境中一般必須來說 執行緒資源都是由執行緒池提供執行緒資源的。 執行緒池的好處 重用存在的執行緒，減少物件建立、消亡的開銷，效能好 可有效控制最大併發執行緒數，提高系統資源利用率，同時可以避免過多資源競爭，避免阻塞。 提供定時執行、定期執行、單執行緒、併發數控制等

J.U.C - AQS

java.util.concurrent（J.U.C）大大提高了併發效能，AQS 被認為是 J.U.C 的核心。 CountdownLatch 用來控制一個執行緒等待多個執行緒。 維護了一個計數器 cnt，每次呼叫 countDown() 方法會讓計數器的值減 1，減到 0 的時候，那些因為呼叫 awa

慕課網實戰·高併發探索（九）：併發容器 J.U.C

特別感謝：慕課網jimin老師的《Java併發程式設計與高併發解決方案》課程，以下知識點多數來自老師的課程內容。 jimin老師課程地址：Java併發程式設計與高併發解決方案 概述 Java併發容器JUC是三個單詞的縮寫。是JDK下面的一個包名。

J.U.C並發框架源碼閱讀（十）ConcurrentLinkedQueue

bstr ray there vol line explained 1.7 root run 基於版本jdk1.7.0_80 java.util.concurrent.ConcurrentLinkedQueue 代碼如下 /* * ORACLE PROPRIETA

【Java】J.U.C併發包 - AQS機制

簡介 Java併發包（java.util.concurrent）中提供了很多併發工具，這其中，很多我們耳熟能詳的併發工具，譬如ReentrantLock、Semaphore，CountDownLatch，CyclicBarrier，它們的實現都用到了一個共同的基類 - Abstrac

Java併發程式設計（6）- J.U.C元件拓展

J.U.C-FutureTask 在Java中一般通過繼承Thread類或者實現Runnable介面這兩種方式來建立執行緒，但是這兩種方式都有個缺陷，就是不能在執行完成後獲取執行的結果，因此Java 1.5之後提供了Callable和Future介面，通過它們就可以在任務執行完畢之後得到任務的執行結果。

高併發第五彈：安全釋出物件及單例模式

要正確的釋出一個物件首先要解決3個問題： 1.釋出的物件只需要被它需要被看見的執行緒或其它物件看見 2.避免逸出問題 3.避免其它執行緒拿到未初始化完全的物件 什麼是釋出 　　釋出一個物件是指，使物件能夠在當前作用域之外的程式碼中使用。比如，將建立的物件儲存到容器中，也可能通過某個方法返回物件的引用，或者將

jdk1.8 J.U.C併發原始碼閱讀------AQS之conditionObject內部類分析

一、繼承關係 public class ConditionObject implements Condition, java.io.Serializable 實現了Condition介面和Serializable介面，是AbstractQueuedSynchronize

【死磕Java併發】-----J.U.C之AQS：阻塞和喚醒執行緒

此篇部落格所有原始碼均來自JDK 1.8 線上程獲取同步狀態時如果獲取失敗，則加入CLH同步佇列，通過通過自旋的方式不斷獲取同步狀態，但是在自旋的過程中則需要判斷當前執行緒是否需要阻塞，其主要方法在acquireQueued()： if (sho

死磕Java併發：J.U.C之併發工具類：Semaphore

作者：chenssy來源：Java技術棧公眾號訊號量Semaphore是一個控制訪問多個共享資源

【死磕Java併發】-----J.U.C之阻塞佇列：ArrayBlockingQueue

ArrayBlockingQueue，一個由陣列實現的有界阻塞佇列。該佇列採用FIFO的原則對元素進行排序新增的。 ArrayBlockingQueue為有界且固定，其大小在構造時由建構函式來決定，確認之後就不能再改變了。ArrayBlockingQueu