蒼穹之邊砸彬，浩瀚之摯颠毙，眰恦之美斯入； 悟心悟性，善始善終蛀蜜，惟善惟道刻两！ —— 朝槿《朝槿兮年說》
寫在開頭
在并發(fā)編程領(lǐng)域，有兩大核心問題：一個(gè)是互斥滴某，即同一時(shí)刻只允許一個(gè)線程訪問共享資源磅摹；另一個(gè)是同步，即線程之間如何通信霎奢、協(xié)作偏瓤。<br />主要原因是，對于多線程實(shí)現(xiàn)實(shí)現(xiàn)并發(fā)椰憋，一直以來厅克，多線程都存在2個(gè)問題：
- 線程之間內(nèi)存共享，需要通過加鎖進(jìn)行控制橙依，但是加鎖會(huì)導(dǎo)致性能下降证舟，同時(shí)復(fù)雜的加鎖機(jī)制也會(huì)增加編程編碼難度
- 過多線程造成線程之間的上下文切換，導(dǎo)致效率低下
因此窗骑，在并發(fā)編程領(lǐng)域中女责，一直有一個(gè)很重要的設(shè)計(jì)原則： “ 不要通過內(nèi)存共享來實(shí)現(xiàn)通信，而應(yīng)該通過通信來實(shí)現(xiàn)內(nèi)存共享创译〉种”<br />簡單來說，就是盡可能通過消息通信软族，而不是內(nèi)存共享來實(shí)現(xiàn)進(jìn)程或者線程之間的同步刷喜。
關(guān)健術(shù)語
<br />本文用到的一些關(guān)鍵詞語以及常用術(shù)語，主要如下：
- 并發(fā)(Concurrent): 在操作系統(tǒng)中立砸，是指一個(gè)時(shí)間段中有幾個(gè)程序都處于已啟動(dòng)運(yùn)行到運(yùn)行完畢之間掖疮，且這幾個(gè)程序都是在同一個(gè)處理機(jī)上運(yùn)行。
- 并行(Parallel): 當(dāng)系統(tǒng)有一個(gè)以上CPU時(shí)颗祝，當(dāng)一個(gè)CPU執(zhí)行一個(gè)進(jìn)程時(shí)浊闪，另一個(gè)CPU可以執(zhí)行另一個(gè)進(jìn)程，兩個(gè)進(jìn)程互不搶占CPU資源螺戳，可以同時(shí)進(jìn)行搁宾。
- 信號(hào)量(Semaphore): 是在多線程環(huán)境下使用的一種設(shè)施，是可以用來保證兩個(gè)或多個(gè)關(guān)鍵代碼段不被并發(fā)調(diào)用倔幼，也是作系統(tǒng)用來解決并發(fā)中的互斥和同步問題的一種方法盖腿。
- 信號(hào)量機(jī)制(Semaphores)： 用來解決同步/互斥的問題的，它是1965年,荷蘭學(xué)者 Dijkstra提出了一種卓有成效的實(shí)現(xiàn)進(jìn)程互斥與同步的方法凤藏。
- 管程(Monitor) : 一般是指管理共享變量以及對共享變量的操作過程奸忽，讓它們支持并發(fā)的一種機(jī)制堕伪。
- 互斥(Mutual Exclusion)：一個(gè)公共資源同一時(shí)刻只能被一個(gè)進(jìn)程或線程使用揖庄，多個(gè)進(jìn)程或線程不能同時(shí)使用公共資源栗菜。即就是同一時(shí)刻只允許一個(gè)線程訪問共享資源的問題。
- 同步(Synchronization)：兩個(gè)或兩個(gè)以上的進(jìn)程或線程在運(yùn)行過程中協(xié)同步調(diào)蹄梢，按預(yù)定的先后次序運(yùn)行疙筹。即就是線程之間如何通信、協(xié)作的問題禁炒。
- 對象池(Object Pool): 指的是一次性創(chuàng)建出 N 個(gè)對象而咆，之后所有的線程重復(fù)利用這 N 個(gè)對象，當(dāng)然對象在被釋放前幕袱，也是不允許其他線程使用的, 一般指保存實(shí)例對象的容器暴备。
基本概述
在Java領(lǐng)域中，我們可以將鎖大致分為基于Java語法層面(關(guān)鍵詞)實(shí)現(xiàn)的鎖和基于JDK層面實(shí)現(xiàn)的鎖们豌。
在Java領(lǐng)域中, 尤其是在并發(fā)編程領(lǐng)域涯捻，對于多線程并發(fā)執(zhí)行一直有兩大核心問題：同步和互斥。其中：
- 互斥(Mutual Exclusion)：一個(gè)公共資源同一時(shí)刻只能被一個(gè)進(jìn)程或線程使用望迎，多個(gè)進(jìn)程或線程不能同時(shí)使用公共資源障癌。即就是同一時(shí)刻只允許一個(gè)線程訪問共享資源的問題。
- 同步(Synchronization)：兩個(gè)或兩個(gè)以上的進(jìn)程或線程在運(yùn)行過程中協(xié)同步調(diào)辩尊，按預(yù)定的先后次序運(yùn)行涛浙。即就是線程之間如何通信、協(xié)作的問題摄欲。
針對對于這兩大核心問題轿亮，利用管程是能夠解決和實(shí)現(xiàn)的，因此可以說胸墙，管程是并發(fā)編程的萬能鑰匙哀托。<br />雖然，Java在基于語法層面(synchronized 關(guān)鍵字)實(shí)現(xiàn)了對管程技術(shù),但是從使用方式和性能上來說劳秋，內(nèi)置鎖(synchronized 關(guān)鍵字)的粒度相對過大仓手，不支持超時(shí)和中斷等問題。<br />為了彌補(bǔ)這些問題玻淑，從JDK層面對其“重復(fù)造輪子”嗽冒，在JDK內(nèi)部對其重新設(shè)計(jì)和定義，甚至實(shí)現(xiàn)了新的特性补履。<br />在Java領(lǐng)域中添坊，從JDK源碼分析來看，基于JDK層面實(shí)現(xiàn)的鎖大致主要可以分為以下4種方式：
- 基于Lock接口實(shí)現(xiàn)的鎖：JDK1.5版本提供的ReentrantLock類
- 基于ReadWriteLock接口實(shí)現(xiàn)的鎖：JDK1.5版本提供的ReentrantReadWriteLock類
- 基于AQS基礎(chǔ)同步器實(shí)現(xiàn)的鎖：JDK1.5版本提供的并發(fā)相關(guān)的同步器Semaphore箫锤，CyclicBarrier以及CountDownLatch等
- 基于自定義API操作實(shí)現(xiàn)的鎖：JDK1.8版本中提供的StampedLock類
從閱讀源碼不難發(fā)現(xiàn)贬蛙，在Java SDK 并發(fā)包主要通過AbstractQueuedSynchronizer(AQS)實(shí)現(xiàn)多線程同步機(jī)制的封裝與定義雨女，而通過Lock 和 Condition 兩個(gè)接口來實(shí)現(xiàn)管程罢低，其中 Lock 用于解決互斥問題束凑，Condition 用于解決同步問題。
一.AQS基礎(chǔ)同步器基本理論
在Java領(lǐng)域中,同步器是專門為多線程并發(fā)設(shè)計(jì)的同步機(jī)制边坤，主要是多線程并發(fā)執(zhí)行時(shí)線程之間通過某種共享狀態(tài)來實(shí)現(xiàn)同步野蝇，只有當(dāng)狀態(tài)滿足這種條件時(shí)線程才往下執(zhí)行的一種同步機(jī)制讼稚。
<br />一個(gè)標(biāo)準(zhǔn)的AQS同步器主要有同步狀態(tài)機(jī)制，等待隊(duì)列绕沈，條件隊(duì)列锐想，獨(dú)占模式，共享模式等五大核心要素組成乍狐。<br />在Java領(lǐng)域中赠摇，JDK的JUC(java.util.concurrent.)包中提供了各種并發(fā)工具，但是大部分同步工具的實(shí)現(xiàn)基于AbstractQueuedSynchronizer類實(shí)現(xiàn)浅蚪，其內(nèi)部結(jié)構(gòu)主要如下：
- 同步狀態(tài)機(jī)制(Synchronization Status)：主要用于實(shí)現(xiàn)鎖(Lock)機(jī)制藕帜，是指同步狀態(tài)，其要求對于狀態(tài)的更新必須原子性的
- 等待隊(duì)列(Wait Queue)：主要用于存放等待線程獲取到的鎖資源掘鄙，并且把線程維護(hù)到一個(gè)Node(節(jié)點(diǎn))里面和維護(hù)一個(gè)非阻塞的CHL Node FIFO(先進(jìn)先出)隊(duì)列耘戚，主要是采用自旋鎖+CAS操作來保證節(jié)點(diǎn)插入和移除的原子性操作。
- 條件隊(duì)列(Condition Queue)：用于實(shí)現(xiàn)鎖的條件機(jī)制操漠，一般主要是指替換“等待-通知”工作機(jī)制收津，主要是通過ConditionObject對象實(shí)現(xiàn)Condition接口提供的方法實(shí)現(xiàn)。
- 獨(dú)占模式(Exclusive Mode)：主要用于實(shí)現(xiàn)獨(dú)占鎖浊伙，主要是基于靜態(tài)內(nèi)部類Node的常量標(biāo)志EXCLUSIVE來標(biāo)識(shí)該節(jié)點(diǎn)是獨(dú)占模式
- 共享模式(Shared Mode)：主要用于實(shí)現(xiàn)共享鎖撞秋，主要是基于靜態(tài)內(nèi)部類Node的常量標(biāo)志SHARED來標(biāo)識(shí)該節(jié)點(diǎn)是共享模式
- 條件變量(Conditional Variable)： 利用線程間共享的變量進(jìn)行同步的一種工作機(jī)制
- 共享變量((Shared Variable))：一般指對象實(shí)體對象的成員變量和屬性
- 阻塞隊(duì)列(Blocking Queue)：共享變量(Shared Variable)及其對共享變量的操作統(tǒng)一封裝
- 等待隊(duì)列(Wait Queue)：每個(gè)條件變量都對應(yīng)有一個(gè)等待隊(duì)列(Wait Queue),內(nèi)部需要實(shí)現(xiàn)入隊(duì)操作(Enqueue)和出隊(duì)操作(Dequeue)方法
- 變量狀態(tài)描述機(jī)(Synchronization Status)：描述條件變量和共享變量之間狀態(tài)變化吻贿，又可以稱其為同步狀態(tài)
- 工作模式(Operation Mode)： 線程資源具有排他性，因此定義獨(dú)占模式和共享模式兩種工作模式
綜上所述哑子，條件變量和等待隊(duì)列的作用是解決線程之間的同步問題舅列；共享變量與阻塞隊(duì)列的作用是解決線程之間的互斥問題。
二. JDK顯式鎖統(tǒng)一概念模型
在并發(fā)編程領(lǐng)域卧蜓，有兩大核心問題：一個(gè)是互斥帐要，即同一時(shí)刻只允許一個(gè)線程訪問共享資源；另一個(gè)是同步弥奸，即線程之間如何通信榨惠、協(xié)作。
綜合Java領(lǐng)域中的并發(fā)鎖的各種實(shí)現(xiàn)與應(yīng)用分析來看，一把鎖或者一種鎖赠橙，基本上都會(huì)包含以下幾個(gè)方面：
- 鎖的同步器工作機(jī)制：主要是考慮共享模式還是獨(dú)享模式耽装，是否支持超時(shí)機(jī)制，以及是否支持超時(shí)機(jī)制期揪？
- 鎖的同步器工作模式：主要是基于AQS基礎(chǔ)同步器封裝內(nèi)部同步器掉奄，是否考慮公平/非公平模式？
- 鎖的狀態(tài)變量機(jī)制： 主要鎖的狀態(tài)設(shè)置横侦，是否共享狀態(tài)變量挥萌？
- 鎖的隊(duì)列封裝定義：主要是指等待隊(duì)列和條件隊(duì)列绰姻，是否需要條件隊(duì)列或者等待隊(duì)列定義枉侧？
- 鎖的底層實(shí)現(xiàn)操作： 主要是指底層CL鎖和CAS操作，是否需要考慮自旋鎖或者CAS操作實(shí)例對象方法狂芋？
- 鎖的組合實(shí)現(xiàn)新鎖： 主要是基于獨(dú)占鎖和共享鎖榨馁，是否考慮對應(yīng)API自定義操作實(shí)現(xiàn)？
綜上所述帜矾，大致可以根據(jù)上述這些方向翼虫，我們便可以清楚???知道Java領(lǐng)域中各種鎖實(shí)現(xiàn)的基本理論時(shí)和實(shí)現(xiàn)思想。
六.CountDownLatch(閉鎖)的設(shè)計(jì)與實(shí)現(xiàn)
在Java領(lǐng)域中屡萤，CountDownLatch(閉鎖)是針對于Java多線程并發(fā)控制中倒計(jì)數(shù)器的具體數(shù)量珍剑，主要是采用遞減計(jì)數(shù)方式的倒計(jì)數(shù)器思想和基于AQS基礎(chǔ)同步器來實(shí)現(xiàn)的一種同步器工具類。
CountDownLatch(閉鎖)是Java多線程并發(fā)中最常見的一種同步器死陆，從鎖的性質(zhì)上來看招拙，屬于共享鎖，其功能相當(dāng)于一個(gè)多線程環(huán)境下的倒數(shù)門閂措译。<br />CountDownLatch通過定義一個(gè)倒計(jì)數(shù)器别凤，在并發(fā)環(huán)境下由線程進(jìn)行遞減1操作，當(dāng)計(jì)數(shù)值變?yōu)?之后领虹，被await方法阻塞的線程將會(huì)喚醒规哪。<br />通過CountDownLatch可以實(shí)現(xiàn)線程間的計(jì)數(shù)同步。
1. 設(shè)計(jì)思想
一般來說塌衰，通過定義一個(gè)倒計(jì)數(shù)器诉稍，為了讓某個(gè)線程或者多個(gè)線程在某個(gè)運(yùn)行節(jié)點(diǎn)上等待N個(gè)條件都滿足后，才讓所有的線程繼續(xù)往下執(zhí)行最疆，其中倒計(jì)數(shù)器的數(shù)量則為N杯巨，每滿足一個(gè)條件，倒計(jì)數(shù)器就依次逐漸遞減1肚菠，直到N-1=0的時(shí)舔箭，所有等待的線程才往下繼續(xù)執(zhí)行。<br />CountDownLatch類最早是在JDK1.5版本提供的，從設(shè)計(jì)思想上來看层扶，主要包括倒計(jì)數(shù)器的同步器箫章，控制阻塞等待的方法，倒計(jì)數(shù)器的遞減操作方法等3個(gè)核心要素镜会。其中：
- 倒計(jì)數(shù)器的同步器：基于AQS基礎(chǔ)抽象隊(duì)列同步器封裝內(nèi)置實(shí)現(xiàn)一個(gè)靜態(tài)的內(nèi)置同步類,主要用于設(shè)置倒計(jì)數(shù)器的初始值以及定制AQS基礎(chǔ)同步器的獲取和釋放共享鎖檬寂。
- 倒計(jì)數(shù)器的初始值： 一般在構(gòu)建CountDownLatch類時(shí)指定，表示的是需要等待條件的個(gè)數(shù)戳表，即就是倒計(jì)數(shù)器的具體的資源數(shù)量Source(N)桶至。
- 控制線程阻塞等待的方法：定義一個(gè)控制線程阻塞等待的方法，當(dāng)?shù)褂?jì)數(shù)器的具體的資源數(shù)量 Source(N)>0時(shí)匾旭，調(diào)用方法使其線程進(jìn)入阻塞等待狀態(tài)镣屹。
- 倒計(jì)數(shù)器的遞減操作方法：定義一個(gè)倒計(jì)數(shù)器的遞減操作方法，調(diào)用方法就會(huì)把倒計(jì)數(shù)器遞減1价涝，當(dāng)?shù)褂?jì)數(shù)器的具體的資源數(shù)量 Source(N)-1=0時(shí)女蜈，所有等待的線程才往下繼續(xù)執(zhí)行。
簡單來說色瘩，CountDownLatch主要是讓某個(gè)線程或者多個(gè)線程伪窖，等待其他線程完成某件事情或者某個(gè)任務(wù)結(jié)束之后才能繼續(xù)執(zhí)行。
2. 基本實(shí)現(xiàn)
在CountDownLatch類的JDK1.8版本中居兆，對于CountDownLatch的基本實(shí)現(xiàn)如下：
public class CountDownLatch {
private final Sync sync;
/**
* CountDownLatch鎖-構(gòu)造一個(gè)倒計(jì)數(shù)器
*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
/**
* CountDownLatch鎖-基于AQS定義支持同步器實(shí)現(xiàn)
*/
private static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860 L;
//......其他方法代碼
}
/**
* CountDownLatch鎖-線程等待方法
*/
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* CountDownLatch鎖-倒計(jì)數(shù)器遞減操作
*/
public void countDown() {
sync.releaseShared(1);
}
//... 其他代碼
}
- 倒計(jì)數(shù)器同步器：基于AQS基礎(chǔ)定義支持同步器實(shí)現(xiàn)一個(gè)靜態(tài)私有化的同步器Sync類覆山，其中定義了獲取和釋放共享鎖的兩個(gè)方法
- 線程等待方法：主要是提供了一個(gè)await()方法，其本質(zhì)是調(diào)用的是AQS基礎(chǔ)同步器中的acquireSharedInterruptibly(int arg)方法泥栖，否則throws InterruptedException異常
- 倒計(jì)數(shù)器遞減操作方法： 主要是提供了一個(gè)countDown()方法簇宽，其本質(zhì)是調(diào)用的是AQS基礎(chǔ)同步器中的releaseShared(int arg) 方法
2.1 基于AQS同步器封裝靜態(tài)內(nèi)部Sync抽象類
/**
* CountDownLatch鎖-基于AQS同步器封裝一個(gè)內(nèi)部的同步器
*/
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
/**
* CountDownLatch鎖-獲取共享鎖方法
*/
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
/**
* CountDownLatch鎖-釋放共享鎖方法
*/
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;
}
}
}
- 實(shí)現(xiàn)方式： 主要基于AQS封裝的內(nèi)部靜態(tài)抽象Sync同步類實(shí)現(xiàn)，使用的AQS的共享模式
- 主要方法： 主要定制適配提供了tryAcquireShared()和tryReleaseShared()方法聊倔，即就是tryAcquireShared()用于獲取共享鎖晦毙，tryReleaseShared()方法用于釋放共享鎖，其中：
- 獲取共享鎖tryAcquireShared()方法：首先獲取狀態(tài)變量status耙蔑，這里是指倒計(jì)數(shù)器中的數(shù)量见妒，當(dāng)status=0時(shí)，返回值=1甸陌，表示獲取鎖成功须揣；否則，status !=0 時(shí)钱豁，返回值=-1耻卡，表示獲取共享鎖失敗進(jìn)行入隊(duì)。
- 釋放共享鎖tryReleaseShared()方法： 通過自旋來實(shí)現(xiàn)遞減操作牲尺，其中會(huì)獲取狀態(tài)變量status卵酪，將其遞減1后使用compareAndSetState(c, nextc)方法通過CAS修改狀態(tài)值
- 鎖獲取方式： 主要是利用getCount()來獲取倒計(jì)數(shù)器中的數(shù)量幌蚊，同時(shí)還可以利用構(gòu)造方法指導(dǎo)一個(gè)倒計(jì)數(shù)器中的數(shù)量。
3. 具體實(shí)現(xiàn)
public class CountDownLatch {
private final Sync sync;
/**
* CountDownLatch鎖-基于AQS基礎(chǔ)同步器實(shí)現(xiàn)一個(gè)內(nèi)部同步器
*/
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374 L;
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;
}
}
}
/**
* CountDownLatch鎖-構(gòu)造一個(gè)倒計(jì)數(shù)器
*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
/**
* CountDownLatch鎖-基于AQS定義支持同步器實(shí)現(xiàn)
*/
private static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860 L;
//......其他方法代碼
}
/**
* CountDownLatch鎖-線程等待方法
*/
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* CountDownLatch鎖-返回當(dāng)前計(jì)數(shù)器
*/
public long getCount() {
return sync.getCount();
}
/**
* CountDownLatch鎖-線程等待方法(支持超時(shí)機(jī)制)
*/
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
/**
* CountDownLatch鎖-倒計(jì)數(shù)器遞減操作
*/
public void countDown() {
sync.releaseShared(1);
}
}
- 倒計(jì)數(shù)初始值：通過構(gòu)造方法CountDownLatch(int count)指定一個(gè)倒計(jì)數(shù)器的初始值溃卡，其必須大于0溢豆，否則會(huì)throw new IllegalArgumentException("count < 0")
- 線程等待方法： 主要提供了await() 方法和await(long timeout, TimeUnit unit)方法，其中：
- 無參數(shù)await() 方法： 一般默認(rèn)的方法瘸羡，其本質(zhì)是調(diào)用AQS同步器中的acquireSharedInterruptibly()方法漩仙，主要表示支持中斷機(jī)制
- 有參數(shù)await(long timeout, TimeUnit unit)方法： 是用于實(shí)現(xiàn)超時(shí)機(jī)制，其本質(zhì)是調(diào)用AQS同步器中的tryAcquireSharedNanos(int arg, long nanosTimeout)方法
- 倒計(jì)數(shù)遞減操作方法：主要是countDown() 方法犹赖， 其本質(zhì)是調(diào)用AQS同步器中的releaseShared(int arg) 方法队他，核心實(shí)現(xiàn)是AQS基礎(chǔ)同步器的doReleaseShared方法。
- 其他方法： 主要是getCount() 方法峻村，用來獲取倒計(jì)數(shù)個(gè)數(shù)麸折，其本質(zhì)是調(diào)用AQS同步器中g(shù)etCount()方法，來獲取狀態(tài)變量
綜上所述雀哨，從一定意義上講磕谅，CountDownLatch是一種共享鎖私爷，屬于AQS基礎(chǔ)抽象隊(duì)列同步器中共享模式孵化的產(chǎn)物雾棺，沒有支持公平模式與非公平模式的實(shí)現(xiàn)。
七.CyclicBarrier(循環(huán)屏障)的設(shè)計(jì)與實(shí)現(xiàn)
在Java領(lǐng)域中衬浑，CyclicBarrier(循環(huán)屏障)是針對于Java多線程并發(fā)控制中倒計(jì)數(shù)器的線程數(shù)量捌浩，主要是采用遞減計(jì)數(shù)方式的倒計(jì)數(shù)器思想和基于AQS基礎(chǔ)同步器實(shí)現(xiàn)的ReentrantLock鎖來實(shí)現(xiàn)的一種同步器工具類。
CyclicBarrier(循環(huán)屏障)是Java中通過對線程預(yù)定義設(shè)置一個(gè)屏障工秩，只有當(dāng)?shù)竭_(dá)屏障的線程數(shù)量到達(dá)指定的最大屏障時(shí)尸饺，屏障才會(huì)讓這些線程通過執(zhí)行。<br />從一定意義上來講助币，這里的屏障本質(zhì)上還是一個(gè)倒計(jì)數(shù)器浪听，倒計(jì)數(shù)器的最大限度支持的數(shù)量就是我們?yōu)榫€程設(shè)置屏障大小，其工作原理與CountDownLatch(閉鎖)類似眉菱，都是通過讓線程阻塞等待時(shí)迹栓，倒計(jì)數(shù)器執(zhí)行遞減1運(yùn)算。<br />但是與CountDownLatch不同是俭缓，CyclicBarrier(循環(huán)屏障)是基于ReentrantLock(可重入鎖)來實(shí)現(xiàn)的克伊，更準(zhǔn)確的說，CyclicBarrier是對ReentrantLock的應(yīng)用實(shí)例华坦。
1. 設(shè)計(jì)思想
一般來說愿吹，通過定義一個(gè)倒計(jì)數(shù)器，為了讓某個(gè)線程或者多個(gè)線程在某個(gè)運(yùn)行節(jié)點(diǎn)上約束N個(gè)線程惜姐，需要讓指定數(shù)量的線程共同到達(dá)某一個(gè)節(jié)點(diǎn)之后犁跪，這些線程才會(huì)一起被執(zhí)行。<br />CyclicBarrier(循環(huán)屏障)最早是在JDK1.5版本中提供的，從設(shè)計(jì)思想上來看坷衍，主要包括倒計(jì)數(shù)器的最大屏障撵颊，控制阻塞等待的方法，倒計(jì)數(shù)器的遞減操作方法惫叛，和觸發(fā)點(diǎn)線程任務(wù)等4個(gè)核心要素倡勇。其中：
- 倒計(jì)數(shù)器的同步器： 主要基于ReentrantLock來實(shí)現(xiàn)控制線程對象，其本質(zhì)還是基于AQS基礎(chǔ)同步器實(shí)現(xiàn)嘉涌。
- 倒計(jì)數(shù)器的最大屏障數(shù)量：一般是在構(gòu)建CyclicBarrier(循環(huán)屏障)對象是預(yù)定義設(shè)置妻熊，表示需要在某個(gè)運(yùn)行節(jié)點(diǎn)上約束的線程數(shù)量。
- 控制線程阻塞等待的方法：定義一個(gè)方法仑最，使得實(shí)現(xiàn)阻塞線程讓其進(jìn)入等待狀態(tài)扔役。
- 倒計(jì)數(shù)器的遞減操作方法：定義一個(gè)方法，使得讓倒計(jì)數(shù)器進(jìn)行遞減1運(yùn)算警医，直到達(dá)到屏障時(shí)亿胸，等待的線程才繼續(xù)執(zhí)行。
- 觸發(fā)點(diǎn)線程任務(wù)：一般指的是當(dāng)指定數(shù)量的線程達(dá)到設(shè)置的屏障時(shí)预皇，才會(huì)去觸發(fā)執(zhí)行的任務(wù)侈玄。
簡單來說，CyclicBarrier(循環(huán)屏障)是讓多個(gè)線程互相等待吟温，直到達(dá)到一個(gè)同步的運(yùn)行節(jié)點(diǎn)序仙。再繼續(xù)一起執(zhí)行。
2. 基本實(shí)現(xiàn)
在CyclicBarrier類的JDK1.8版本中鲁豪，對于CountDownLatch的基本實(shí)現(xiàn)如下：
public class CyclicBarrier {
/** CyclicBarrier鎖—屏障lock實(shí)體 */
private final ReentrantLock lock = new ReentrantLock();
/** CyclicBarrier鎖—屏障條件隊(duì)列 */
private final Condition trip = lock.newCondition();
/** CyclicBarrier鎖—屏障最大值 */
private final int parties;
/** CyclicBarrier鎖—屏障觸發(fā)線程任務(wù)目標(biāo) */
private final Runnable barrierCommand;
/** CyclicBarrier鎖—當(dāng)前計(jì)數(shù)器的最大值屏障實(shí)例 */
private Generation generation = new Generation();
/** CyclicBarrier鎖—當(dāng)前計(jì)數(shù)器的最大值屏障實(shí)例 */
private int count;
/** CyclicBarrier鎖—屏障實(shí)例 */
private static class Generation {
boolean broken = false;
}
/** CyclicBarrier鎖—構(gòu)造一個(gè)屏障實(shí)例(不帶觸發(fā)任務(wù)的) */
public CyclicBarrier(int parties) {
this(parties, null);
}
/** CyclicBarrier鎖—構(gòu)造一個(gè)屏障實(shí)例(帶觸發(fā)任務(wù)的) */
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
/** CyclicBarrier鎖—無參數(shù)構(gòu)造一個(gè)等待方法(默認(rèn)模式) */
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
/** CyclicBarrier鎖—有參數(shù)構(gòu)造一個(gè)等待方法(支持超時(shí)機(jī)制) */
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
/** CyclicBarrier鎖—更新狀態(tài)變量 */
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}
/** CyclicBarrier鎖—阻塞屏障 */
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
//...其他代碼
}
- 預(yù)定義設(shè)置屏障最大值： 主要是通過變量parties來實(shí)現(xiàn)預(yù)定義設(shè)置屏障最大值
- 設(shè)置當(dāng)前屏障數(shù)量：主要是通過變量count來實(shí)現(xiàn)
- 控制線程的對象實(shí)例： 主要是通過ReentrantLock和Condition來控制線程間通信
- 觸發(fā)目標(biāo)任務(wù)對象： 主要是通過Runable來定義barrierCommand變量
- 提供了兩個(gè)構(gòu)造方法：都需要預(yù)定義指定屏障最大值parties潘悼，其中一個(gè)需要傳入barrierAction觸發(fā)點(diǎn)任務(wù)
- 線程阻塞等待方法：主要提供了2個(gè)await()方法，其中：
- 無參數(shù)await()方法：默認(rèn)處理方式爬橡，不支持超時(shí)機(jī)制治唤，其核心處理邏輯在dowait(boolean timed, long nanos)方法中實(shí)現(xiàn)
- 有參數(shù)await()方法：指定參數(shù)處理，支持超時(shí)機(jī)制糙申，其核心處理邏輯在dowait(boolean timed, long nanos)方法中實(shí)現(xiàn)
- 屏障設(shè)置關(guān)健方法：主要是breakBarrier() 來實(shí)現(xiàn)宾添，其中：
- 通知到達(dá)屏障的所有線程：主要是通過Condition中的signalAll()來通知屏障中所有線程已經(jīng)滿足條件
- 屏障設(shè)置：默認(rèn)預(yù)定義設(shè)置屏障最大值與設(shè)置當(dāng)前屏障數(shù)相同，主要設(shè)置count = parties
- 更新屏障狀態(tài)：主要是通過generation.broken = true來實(shí)現(xiàn)
- 更新屏障的狀態(tài)：主要是提供了nextGeneration() 方法郭宝，表示已經(jīng)到達(dá)預(yù)定義設(shè)置屏障最大值辞槐，其中：
- 通知到達(dá)屏障的所有線程：主要是通過Condition中的signalAll()來通知屏障中所有線程已經(jīng)滿足條件
- 準(zhǔn)備下一輪屏障設(shè)置：意味著預(yù)定義設(shè)置屏障最大值與設(shè)置當(dāng)前屏障數(shù)相同，主要設(shè)置count = parties
- 重置屏障狀態(tài)：主要是通過generation = new Generation()來實(shí)現(xiàn)
2.1 構(gòu)造Generation屏障實(shí)例標(biāo)記
private static class Generation {
boolean broken = false;
}
主要是構(gòu)造了一個(gè)靜態(tài)私有化的Generation類，其中定義了一個(gè)broken變量來作為屏障標(biāo)記衔统，默認(rèn)初始值為false鹿榜，表示還沒達(dá)到屏障最大值海雪。
2.1 線程阻塞等待核心dowait方法
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
// [1].實(shí)例化構(gòu)建ReentrantLock的對象
final ReentrantLock lock = this.lock;
// [2].通過lock()獲取鎖或者說加鎖操作
lock.lock();
try {
// [3].實(shí)例化構(gòu)建Generation屏障實(shí)例對象
final Generation g = generation;
// [4].判斷Generation屏障實(shí)例標(biāo)記狀態(tài)
if (g.broken)
throw new BrokenBarrierException();
// [5].判斷Thread是包含中斷標(biāo)志位
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
// [6].對倒計(jì)數(shù)器的屏障數(shù)量遞減1運(yùn)算
int index = --count;
// [7].依據(jù)結(jié)果index == 0表示當(dāng)前指定的線程數(shù)量到達(dá)屏障最大值，需要觸發(fā)Runnable任務(wù)
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 進(jìn)行下一輪屏障設(shè)置
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// [7].自旋操作
for (;;) {
try {
// 判斷是否超時(shí)
if (!timed)
trip.await();
else if (nanos > 0L)
// 進(jìn)行下一輪屏障設(shè)置
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
// 是否發(fā)生線程中斷
Thread.currentThread().interrupt();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
// 如果等待時(shí)間超過指定超時(shí)時(shí)間舱殿，throw new TimeoutException
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
// 最后釋放鎖操作
lock.unlock();
}
}
- 加鎖操作： 實(shí)例化構(gòu)建ReentrantLock的對象奥裸，通過lock()方法進(jìn)行加鎖操作
- 判斷屏障實(shí)例標(biāo)記狀態(tài)：實(shí)例化構(gòu)建Generation實(shí)例標(biāo)記，判斷屏障實(shí)例標(biāo)記狀態(tài)是否一致沪袭，如果不一致則throw new BrokenBarrierException();
- 判斷當(dāng)前線程是否被中斷： 判斷Thread是包含中斷標(biāo)志位湾宙，如果中斷throw new InterruptedException()并調(diào)用breakBarrier()重新設(shè)置屏障
- 屏障倒計(jì)數(shù)器遞減運(yùn)算：對倒計(jì)數(shù)器的屏障數(shù)量遞減1運(yùn)算，即就是對當(dāng)前倒計(jì)數(shù)器的當(dāng)前值減去1
- 觸發(fā)節(jié)點(diǎn)線程任務(wù)： 當(dāng)前倒計(jì)數(shù)器的當(dāng)前值為0時(shí)冈绊，需要觸發(fā)Runnable任務(wù)侠鳄，并調(diào)用nextGeneration方法開啟下一輪操作；否則死宣，當(dāng)前倒計(jì)數(shù)器的當(dāng)前值不為0時(shí)伟恶，調(diào)用awaitNanos(nanos)方法進(jìn)入等待狀態(tài)
- 自旋操作判斷超時(shí)： 如果使用了超時(shí)參數(shù)，調(diào)用awaitNanos(nanos)方法進(jìn)入等待狀態(tài)毅该，其中如果發(fā)生中斷則調(diào)用Thread.currentThread().interrupt()設(shè)置中斷標(biāo)記博秫。如果等待時(shí)間> 指定超時(shí)時(shí)間，拋出throw new TimeoutException()異常
- 釋放鎖： 通過unlock()方法進(jìn)行解鎖操作眶掌，并釋放鎖
3. 具體實(shí)現(xiàn)
在CyclicBarrier類的JDK1.8版本中挡育，對于CyclicBarrier的具體實(shí)現(xiàn)如下：
public class CyclicBarrier {
/** CyclicBarrier鎖—屏障lock實(shí)體 */
private final ReentrantLock lock = new ReentrantLock();
/** CyclicBarrier鎖—屏障條件隊(duì)列 */
private final Condition trip = lock.newCondition();
/** CyclicBarrier鎖—屏障最大值 */
private final int parties;
/** CyclicBarrier鎖—屏障觸發(fā)線程任務(wù)目標(biāo) */
private final Runnable barrierCommand;
/** CyclicBarrier鎖—當(dāng)前計(jì)數(shù)器的最大值屏障實(shí)例 */
private Generation generation = new Generation();
/** CyclicBarrier鎖—當(dāng)前計(jì)數(shù)器的最大值屏障實(shí)例 */
private int count;
/** CyclicBarrier鎖—屏障實(shí)例 */
private static class Generation {
boolean broken = false;
}
/** CyclicBarrier鎖—構(gòu)造一個(gè)屏障實(shí)例(不帶觸發(fā)任務(wù)的) */
public CyclicBarrier(int parties) {
this(parties, null);
}
/** CyclicBarrier鎖—構(gòu)造一個(gè)屏障實(shí)例(帶觸發(fā)任務(wù)的) */
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
/** CyclicBarrier鎖—無參數(shù)構(gòu)造一個(gè)等待方法(默認(rèn)模式) */
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
/** CyclicBarrier鎖—有參數(shù)構(gòu)造一個(gè)等待方法(支持超時(shí)機(jī)制) */
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
/** CyclicBarrier鎖—更新狀態(tài)變量 */
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}
/** CyclicBarrier鎖—阻塞屏障 */
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
/** CyclicBarrier鎖—阻塞屏障 */
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
// [1].實(shí)例化構(gòu)建ReentrantLock的對象
final ReentrantLock lock = this.lock;
// [2].通過lock()獲取鎖或者說加鎖操作
lock.lock();
try {
// [3].實(shí)例化構(gòu)建Generation屏障實(shí)例對象
final Generation g = generation;
// [4].判斷Generation屏障實(shí)例標(biāo)記狀態(tài)是否為true
if (g.broken)
throw new BrokenBarrierException();
// [5].判斷Thread是包含中斷標(biāo)志位
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
// [6].對倒計(jì)數(shù)器的屏障數(shù)量遞減1運(yùn)算
int index = --count;
// [7].依據(jù)結(jié)果index == 0表示當(dāng)前指定的線程數(shù)量到達(dá)屏障最大值，需要觸發(fā)Runnable任務(wù)
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 進(jìn)行下一輪屏障設(shè)置
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// [7].自旋操作
for (;;) {
try {
// 判斷是否超時(shí)
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
Thread.currentThread().interrupt();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
// 如果等待時(shí)間超過指定超時(shí)時(shí)間畏线，throw new TimeoutException
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
// 最后釋放鎖操作
lock.unlock();
}
}
/** CyclicBarrier鎖—獲取當(dāng)前等屏障等待數(shù)量 */
public int getNumberWaiting() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return parties - count;
} finally {
lock.unlock();
}
}
/** CyclicBarrier鎖—獲取當(dāng)前等屏障數(shù)量 */
public int getParties() {
return parties;
}
/** CyclicBarrier鎖—判斷當(dāng)前屏障 */
public boolean isBroken() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return generation.broken;
} finally {
lock.unlock();
}
}
/** CyclicBarrier鎖—重置屏障數(shù)量 */
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}
}
主要需要注意如下幾個(gè)方法静盅，都是基于ReentrantLock來實(shí)現(xiàn)加鎖和解鎖操作的，其中：
- getNumberWaiting()方法： 獲取當(dāng)前屏障中等待的線程數(shù)量
- reset() 方法：當(dāng)一輪屏障操作結(jié)束寝殴，需要重置屏障中最大線程數(shù)量
- isBroken() 方法：判斷是否到達(dá)屏障最大值
綜上所述，從一定意義上講明垢，CyclicBarrier是一種可重入鎖蚣常，屬于ReentrantLock的應(yīng)用實(shí)例，其中加鎖和解鎖操作都是獨(dú)占模式的痊银。
八.Semaphore(信號(hào)量)的設(shè)計(jì)與實(shí)現(xiàn)
在Java領(lǐng)域中抵蚊，Semaphore(信號(hào)量)是針對于Java多線程并發(fā)控制中實(shí)現(xiàn)對公共資源的線程數(shù)量進(jìn)行并發(fā)同時(shí)訪問控制，主要是采用指定一個(gè)最大許可數(shù)的思想和基于AQS基礎(chǔ)同步器來實(shí)現(xiàn)的一種同步器工具類溯革。
Semaphore可以用來控制在同一時(shí)刻訪問共享資源的線程數(shù)量贞绳，通過協(xié)調(diào)各個(gè)線程以保證共享資源的合理使用。<br />Semaphore維護(hù)了一組虛擬許可致稀，它的數(shù)量可以通過構(gòu)造器的參數(shù)指定冈闭。<br />線程在訪問共享資源前，必須調(diào)用Semaphore的acquire()方法獲得許可抖单，如果許可數(shù)量為0萎攒，該線程就一直阻塞遇八。<br />線程在訪問共享資源后，必須調(diào)用Semaphore的release()方法釋放許可耍休。
1. 設(shè)計(jì)思想
一般來說刃永，通過定義一個(gè)倒計(jì)數(shù)器，為了控制最多N個(gè)線程同時(shí)訪問公共資源羊精，其計(jì)數(shù)器的最大值Max(N)是被許可的最多N個(gè)線程數(shù)量斯够，即就是許可的最大值N。<br />Semaphore類最早是在JDK1.5版本提供的喧锦，從設(shè)計(jì)思想上來看雳刺，主要包括倒計(jì)數(shù)器的最大許可數(shù)，同步器工作模式裸违，獲取鎖方法掖桦，釋放鎖方法等4個(gè)核心要素。其中：
- 同步器工作模式：基于AQS基礎(chǔ)抽象隊(duì)列同步器封裝內(nèi)置實(shí)現(xiàn)一個(gè)靜態(tài)的內(nèi)置同步器抽象類供汛，然后基于這個(gè)抽象類分別實(shí)現(xiàn)了公平同步器和非公平同步器枪汪，用來指定和描述同步器工作模式是公平模式還是非公平模式。
- 公平/非公平模式：主要描述的是多個(gè)線程在同時(shí)獲取鎖時(shí)是否按照先到先得的順序獲取鎖怔昨，如果是則為公平模式雀久，否則為非公平模式。
- 獲取鎖方法：主要定義了一個(gè)lock()方法來獲取鎖趁舀，表示假如鎖已經(jīng)被其他線程占有或持有赖捌，其當(dāng)前獲取鎖的線程則進(jìn)入等待狀態(tài)。
- 釋放鎖方法：主要定義了一個(gè)unlock()方法來釋放鎖矮烹，表示假如鎖已經(jīng)被其他線程放棄或釋放越庇，其當(dāng)前獲取鎖的線程則獲得該鎖。
2. 基本實(shí)現(xiàn)
在JDK1.8版本中奉狈，對于Semaphore的基本實(shí)現(xiàn)如下：
public class Semaphore implements java.io.Serializable {
private static final long serialVersionUID = -3222578661600680210 L;
/**
* Semaphore鎖- 封裝同步器
*/
private final Sync sync;
/**
* Semaphore鎖- 封裝同步器
*/
abstract static class Sync extends AbstractQueuedSynchronizer {
//....其他代碼
}
/**
* Semaphore鎖- 構(gòu)造一個(gè)令牌許可(默認(rèn)非公模式)
*/
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
/**
* Semaphore鎖- 構(gòu)造一個(gè)令牌許可(可選公平/非公模式)
*/
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
/**
* Semaphore鎖- 獲取鎖方法(默認(rèn)一個(gè)且可中斷機(jī)制)
*/
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* Semaphore鎖- 獲取鎖方法(可選指定多個(gè)且可中斷機(jī)制)
*/
public void acquire(int permits) throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
/**
* Semaphore鎖- 獲取鎖方法(默認(rèn)多個(gè)且不可中斷機(jī)制)
*/
public void acquireUninterruptibly() {
sync.acquireShared(1);
}
/**
* Semaphore鎖- 獲取鎖方法(指定多個(gè)且不可中斷機(jī)制)
*/
public void acquireUninterruptibly(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireShared(permits);
}
/**
* Semaphore鎖-釋放鎖方法(默認(rèn)一個(gè))
*/
public void release() {
sync.releaseShared(1);
}
/**
* Semaphore鎖-釋放鎖方法(可選指定多個(gè))
*/
public void release(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.releaseShared(permits);
}
}
- 內(nèi)部同步器：基于AQS基礎(chǔ)同步器封裝和定義了一個(gè)靜態(tài)內(nèi)部Sync抽象類卤唉，其中抽象了一個(gè)內(nèi)置鎖lock()方法
- 同步器工作模式：主要提供了 2個(gè)構(gòu)造方法，其中無參數(shù)構(gòu)造方法表示的是默認(rèn)的工作模式仁期，有參數(shù)構(gòu)造方法主要依據(jù)參數(shù)來實(shí)現(xiàn)指定的工作模式
- 獲取鎖方法： 主要提供了3個(gè)基于acquire方法桑驱，用于獲取鎖共享鎖，其中：
- 無參數(shù)acquire()方法：獲取共享鎖的一般模式跛蛋，默認(rèn)指定一個(gè)許可和支持可中斷機(jī)制
- 有參數(shù)acquire()方法：獲取共享鎖的指定模式熬的，可選指定多個(gè)許可且支持可中斷機(jī)制
- 無參數(shù)acquireUninterruptibly()方法：獲取共享鎖的指定模式，默認(rèn)指定一個(gè)許可且不支持可中斷機(jī)制
- 釋放鎖方法： 主要是提供了2個(gè)release()方法用于釋放鎖共享鎖赊级，其中：
- 無參數(shù)release()方法：釋放共享鎖的一般模式押框，默認(rèn)指定一個(gè)許可和支持可中斷機(jī)制
- 有參數(shù)release()方法：釋放共享鎖的指定模式，可選指定多個(gè)許可且支持可中斷機(jī)制
2.1 基于AQS同步器封裝靜態(tài)內(nèi)部Sync抽象類
/**
* Semaphore鎖- 基于AQS基礎(chǔ)同步器封裝同步器
*/
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933 L;
Sync(int permits) {
setState(permits);
}
final int getPermits() {
return getState();
}
/**
* Semaphore鎖- 非公平模式獲取共享鎖
*/
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
/**
* Semaphore鎖- 釋放共享鎖
*/
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
/**
* Semaphore鎖- 自旋+compareAndSetState通過CAS操作計(jì)算操作令牌許可數(shù)
*/
final void reducePermits(int reductions) {
for (;;) {
int current = getState();
int next = current - reductions;
if (next > current) // underflow
throw new Error("Permit count underflow");
if (compareAndSetState(current, next))
return;
}
}
/**
* Semaphore鎖- 自旋+compareAndSetState通過CAS操作重置令牌許可數(shù)
*/
final int drainPermits() {
for (;;) {
int current = getState();
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}
}
- 實(shí)現(xiàn)方式：主要是基于AQS基礎(chǔ)同步器封裝了一個(gè)靜態(tài)的的Sync抽象類此衅，通過構(gòu)造方法指定一個(gè)最大的令牌許可數(shù)量
- 主要方法：主要是看共享鎖的獲取nonfairTryAcquireShared()方法和釋放鎖tryReleaseShared()方法强戴，其中：
- 獲取鎖nonfairTryAcquireShared()方法：非公平模式下獲取共享鎖亭螟，利用自旋+compareAndSetState()方法通過CAS操作，保證并發(fā)修改令牌許可數(shù)量
- 釋放鎖tryReleaseShared(i)方法： 公平/非公平模式下釋放共享鎖骑歹，利用自旋+compareAndSetState()方法通過CAS操作釋放预烙，會(huì)把釋放的令牌許可數(shù)量增加到當(dāng)前剩余的令牌許可數(shù)量中。
- 令牌許可操作方法：主要提供了drainPermits() 方法 和reducePermits() 方法道媚，其中：
- drainPermits() 方法：主要是利用自旋+compareAndSetState()方法通過CAS操作重置令牌許可數(shù)
- reducePermits() 方法：主要是自旋+compareAndSetState)方法通過CAS操作遞減計(jì)算操作令牌許可數(shù)
- 獲取鎖方式：令牌許可數(shù)量QS基礎(chǔ)同步器狀態(tài)變量對應(yīng)扁掸，通過getPermits() 方法來獲取令牌許可數(shù)量，本質(zhì)是調(diào)用AQS基礎(chǔ)同步器中的getState()來獲取狀態(tài)變量最域。
<br />即就意味著斋射，剩余令牌許可數(shù)量等于當(dāng)前可用令牌許可數(shù)量與消耗令牌許可數(shù)量之差呀闻。<br />由此可見倒信，在公平/非公平模式下憎账，我們對于對于獲取鎖和釋放鎖時(shí)萌朱，對于剩余令牌許可數(shù)量Remanent(N)計(jì)算都滿足以下公式：
- 首先壶运，在線程在訪問共享資源前喜命，我們可以允許的最大值為Available(N),自旋獲取鎖的數(shù)量為Acquires(N)牌里，那么我們在獲取鎖時(shí)：
- 其次，在線程在訪問共享資源后务甥，自旋釋放鎖的數(shù)量為Releases(N)牡辽，那么我們在釋放鎖時(shí)：
<br />當(dāng)然喳篇，需要注意的的一個(gè)問題陪每，就是當(dāng)剩余令牌許可數(shù)量Remanent(N) < 0時(shí)臭挽，表示當(dāng)前線程會(huì)進(jìn)入阻塞等待狀態(tài)。
2.2 基于Sync抽象類封裝FairSync公平同步器
/**
* Semaphore鎖- 基于Sync抽象類封裝FairSync公平同步器
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944 L;
/**
* Semaphore鎖- Semaphore鎖- 通過構(gòu)造方法指定令牌許可
*/
FairSync(int permits) {
super(permits);
}
/**
* Semaphore鎖- Semaphore鎖- 公平模式釋放共享鎖
*/
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}
- 實(shí)現(xiàn)方式： 主要是在基于靜態(tài)內(nèi)部Sync抽象類來實(shí)現(xiàn)铜跑，構(gòu)造了一個(gè)可指定大小的的令牌許可
- 主要方法： 主要是提供了一個(gè)tryAcquireShared方法因妙，其中利用hasQueuedPredecessors()來保證公平性
- 工作機(jī)制： 通過基于AQS基礎(chǔ)同步器中的等待隊(duì)列來實(shí)現(xiàn)公平機(jī)制
需要注意的是痰憎，在未達(dá)到最大的令牌許可數(shù)量時(shí)，所有線程都不會(huì)進(jìn)入等待隊(duì)列中攀涵。
2.3 基于Sync抽象類封裝NonfairSync非公平同步器
/**
* Semaphore鎖- 基于Sync抽象類封裝NonfairSync非公平同步器
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898 L;
/**
* Semaphore鎖- Semaphore鎖- 通過構(gòu)造方法指定令牌許可
*/
NonfairSync(int permits) {
super(permits);
}
/**
* Semaphore鎖- Semaphore鎖- 非公平模式釋放共享鎖
*/
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
}
- 實(shí)現(xiàn)方式： 主要是在基于靜態(tài)內(nèi)部Sync抽象類來實(shí)現(xiàn)铣耘，構(gòu)造了一個(gè)可指定大小的的令牌許可
- 主要方法： 主要是提供了一個(gè)tryAcquireShared方法，其中主要是調(diào)用了靜態(tài)內(nèi)部Sync抽象類nonfairTryAcquireShared方法以故。
- 工作機(jī)制： 通過自旋操作讓所有線程競爭獲取令牌許可蜗细，本質(zhì)還是采用了AQS基礎(chǔ)同步器中闖入策略到打破公平的
3. 具體實(shí)現(xiàn)
在JDK1.8版本中，對于Semaphore的具體實(shí)現(xiàn)如下：
public class Semaphore implements java.io.Serializable {
private static final long serialVersionUID = -3222578661600680210 L;
/**
* Semaphore鎖- 封裝同步器
*/
private final Sync sync;
/**
* Semaphore鎖- 基于AQS基礎(chǔ)同步器封裝同步器
*/
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933 L;
Sync(int permits) {
setState(permits);
}
final int getPermits() {
return getState();
}
/**
* Semaphore鎖- 非公平模式獲取共享鎖
*/
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
/**
* Semaphore鎖- 釋放共享鎖
*/
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
/**
* Semaphore鎖- 自旋+compareAndSetState通過CAS操作計(jì)算操作令牌許可數(shù)
*/
final void reducePermits(int reductions) {
for (;;) {
int current = getState();
int next = current - reductions;
if (next > current) // underflow
throw new Error("Permit count underflow");
if (compareAndSetState(current, next))
return;
}
}
/**
* Semaphore鎖- 自旋+compareAndSetState通過CAS操作重置令牌許可數(shù)
*/
final int drainPermits() {
for (;;) {
int current = getState();
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}
}
/**
* Semaphore鎖- 基于Sync抽象類封裝FairSync公平同步器
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944 L;
/**
* Semaphore鎖- Semaphore鎖- 通過構(gòu)造方法指定令牌許可
*/
FairSync(int permits) {
super(permits);
}
/**
* Semaphore鎖- Semaphore鎖- 公平模式釋放共享鎖
*/
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}
/**
* Semaphore鎖- 基于Sync抽象類封裝NonfairSync非公平同步器
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898 L;
/**
* Semaphore鎖- Semaphore鎖- 通過構(gòu)造方法指定令牌許可
*/
NonfairSync(int permits) {
super(permits);
}
/**
* Semaphore鎖- Semaphore鎖- 非公平模式釋放共享鎖
*/
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
}
/**
* Semaphore鎖- 構(gòu)造一個(gè)令牌許可(默認(rèn)非公模式)
*/
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
/**
* Semaphore鎖- 構(gòu)造一個(gè)令牌許可(可選公平/非公模式)
*/
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
/**
* Semaphore鎖- 獲取鎖方法(默認(rèn)一個(gè)且可中斷機(jī)制)
*/
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* Semaphore鎖- 獲取鎖方法(可選指定多個(gè)且可中斷機(jī)制)
*/
public void acquire(int permits) throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
/**
* Semaphore鎖- 獲取鎖方法(默認(rèn)多個(gè)且不可中斷機(jī)制)
*/
public void acquireUninterruptibly() {
sync.acquireShared(1);
}
/**
* Semaphore鎖- 獲取鎖方法(指定多個(gè)且不可中斷機(jī)制)
*/
public void acquireUninterruptibly(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireShared(permits);
}
/**
* Semaphore鎖-釋放鎖方法(默認(rèn)一個(gè))
*/
public void release() {
sync.releaseShared(1);
}
/**
* Semaphore鎖-釋放鎖方法(可選指定多個(gè))
*/
public void release(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.releaseShared(permits);
}
/**
* Semaphore鎖-嘗試獲取鎖方法(默認(rèn)一個(gè))
*/
public boolean tryAcquire() {
return sync.nonfairTryAcquireShared(1) >= 0;
}
/**
* Semaphore鎖-嘗試獲取鎖方法(可選指定多個(gè))
*/
public boolean tryAcquire(int permits) {
if (permits < 0) throw new IllegalArgumentException();
return sync.nonfairTryAcquireShared(permits) >= 0;
}
/**
* Semaphore鎖-嘗試獲取鎖方法(可選指定多個(gè)并且支持超時(shí)機(jī)制)
*/
public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
}
/**
* Semaphore鎖-嘗試獲取鎖方法(默認(rèn)一個(gè)并且支持超時(shí)機(jī)制)
*/
public boolean tryAcquire(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
/**
* Semaphore鎖-統(tǒng)計(jì)可以令牌許可數(shù)
*/
public int availablePermits() {
return sync.getPermits();
}
/**
* Semaphore鎖-重置令牌許可數(shù)
*/
public int drainPermits() {
return sync.drainPermits();
}
/**
* Semaphore鎖-遞減計(jì)算令牌許可數(shù)
*/
protected void reducePermits(int reduction) {
if (reduction < 0) throw new IllegalArgumentException();
sync.reducePermits(reduction);
}
/**
* Semaphore鎖-判斷是否公平模式
*/
public boolean isFair() {
return sync instanceof FairSync;
}
/**
* Semaphore鎖-判斷隊(duì)列中是否存在線程對象
*/
public final boolean hasQueuedThreads() {
return sync.hasQueuedThreads();
}
/**
* Semaphore鎖-獲取隊(duì)列長度
*/
public final int getQueueLength() {
return sync.getQueueLength();
}
/**
* Semaphore鎖-獲取隊(duì)列的線程對象
*/
protected Collection < Thread > getQueuedThreads() {
return sync.getQueuedThreads();
}
}
- 信號(hào)量同步器： 主要是提供了2個(gè)構(gòu)造方法來實(shí)現(xiàn)令牌許可的管理怒详，其中：
- 默認(rèn)非公平模式：依據(jù)指定傳入的令牌許可數(shù)量permits直接實(shí)例化NonfairSync非公平同步器
- 可選公平/非公平模式：依據(jù)指定傳入的令牌許可數(shù)量permits和公平標(biāo)記fair來實(shí)例化NonfairSync非公平同步器和FairSync公平同步器炉媒，其中，當(dāng)fair=true時(shí)昆烁，是公平平模式吊骤，否則為非公平模式
- 支持可中斷機(jī)制：主要是提供了2個(gè)acquire()方法來獲取鎖，其中：
- 無參數(shù)acquire()方法：一般模式獲取共享鎖静尼，主要是基于AQS基礎(chǔ)同步器中的acquireSharedInterruptibly(int arg)來實(shí)現(xiàn)白粉，其核心邏輯是doAcquireSharedInterruptibly(int arg)來操縱。
- 有參數(shù)acquire()方法：依據(jù)指定傳入的令牌許可數(shù)量permits來判斷鼠渺，當(dāng)permits< 0時(shí)鸭巴，直接throw new IllegalArgumentException()；否則拦盹，直接調(diào)用acquireSharedInterruptibly(permits)方法鹃祖。
- 支持不可中斷機(jī)制：主要是提供了2個(gè)acquireUninterruptibly() 方法，其中：
- 無參數(shù)acquireUninterruptibly() 方法：一般模式獲取共享鎖普舆，主要是基于AQS基礎(chǔ)同步器中acquireShared(int arg)方法來實(shí)現(xiàn)恬口，其核心邏輯是doAcquireShared(int arg) 來操縱。
- 有參數(shù)acquireUninterruptibly() 方法：依據(jù)指定傳入的令牌許可數(shù)量permits來判斷沼侣，當(dāng)permits< 0時(shí)楷兽，直接throw new IllegalArgumentException()；否則华临，直接調(diào)用acquireShared(int arg)方法。
- 非公平模式獲取鎖方式： 主要提供了2個(gè)tryAcquire() 方法端考，其中：
- 無參數(shù)tryAcquire() 方法：非公平模式嘗試獲取共享鎖雅潭，直接調(diào)用的是非公平同步器中的nonfairTryAcquireShared(int acquires) 方法揭厚。
- 有參數(shù)tryAcquire() 方法：依據(jù)指定傳入的令牌許可數(shù)量permits來判斷，當(dāng)permits< 0時(shí)扶供，直接throw new IllegalArgumentException()筛圆；否則，直接調(diào)用nonfairTryAcquireShared(int acquires) 方法椿浓。
- 公平模式獲取鎖方式：主要提供了2個(gè)tryAcquire() 方法太援，支持超時(shí)機(jī)制。其中：
- 無參數(shù)tryAcquire() 方法：公平模式嘗試獲取共享鎖扳碍，依據(jù)指定傳入的令牌許可數(shù)量permits來判斷提岔，當(dāng)permits< 0時(shí)，直接throw new IllegalArgumentException()笋敞；否則碱蒙，直接調(diào)用的是AQS基礎(chǔ)同步器中的tryAcquire(int permits, long timeout, TimeUnit unit)方法，其核心邏輯是tryAcquireSharedNanos(int arg, long nanosTimeout)來操縱夯巷。
- 有參數(shù)tryAcquire() 方法：公平模式嘗試獲取共享鎖赛惩，默認(rèn)支持一個(gè)許可，直接調(diào)用的是AQS基礎(chǔ)同步器中的tryAcquire(1趁餐，long timeout, TimeUnit unit)方法喷兼，其核心邏輯是tryAcquireSharedNanos(int arg, long nanosTimeout)來操縱。
- 釋放鎖操作方式：主要提供了2個(gè)release()方法,其中：
- 無參數(shù)release() 方法：公平/非公平模式示范鎖操作后雷，默認(rèn)支持一個(gè)許可季惯，主要是直接調(diào)用AQS基礎(chǔ)同步器中的releaseShared(int arg) 方法
- 有參數(shù)release() 方法：公平/非公平模式示范鎖操作，依據(jù)指定傳入的令牌許可數(shù)量permits來判斷喷面，當(dāng)permits< 0時(shí)星瘾，直接throw new IllegalArgumentException()；否則惧辈，主要是直接調(diào)用AQS基礎(chǔ)同步器中的releaseShared(int arg) 方法
- 令牌許可操作方法：主要提供了availablePermits() 方法琳状，reducePermits(int reduction)方法 以及drainPermits() 方法，其中：
- availablePermits() 方法：獲取可用的令牌許可數(shù)量盒齿，主要是調(diào)用內(nèi)部同步器中g(shù)etPermits()方法念逞。
- reducePermits()方法：計(jì)算剩余可用令牌許可數(shù)量，依據(jù)指定傳入的令牌許可數(shù)量reduction來判斷边翁，當(dāng)reduction< 0時(shí)翎承，直接throw new IllegalArgumentException()；否則符匾，調(diào)用內(nèi)部同步器中reducePermits()方法叨咖。
- drainPermits() 方法：重置可用令牌許可數(shù)量，主要是調(diào)用內(nèi)部同步器中drainPermits()方法。
- 隊(duì)列操作方法：主要提供了hasQueuedThreads()方法甸各，getQueuedThreads() 方法以及getQueueLength() 方法垛贤，其中：
- hasQueuedThreads()方法：主要是用于獲取隊(duì)列中是否存在等待獲取令牌許可的線程對象，主要是直接使用AQS基礎(chǔ)同步器的hasQueuedThreads()來實(shí)現(xiàn)趣倾。
- getQueuedThreads() 方法：主要是用于獲取隊(duì)列中等待獲取令牌許可的線程對象聘惦，主要是直接使用AQS基礎(chǔ)同步器的getQueuedThreads()來實(shí)現(xiàn)。
- getQueueLength() 方法：主要是用于獲取隊(duì)列中等待獲取令牌許可的數(shù)量儒恋，主要是直接使用AQS基礎(chǔ)同步器的getQueueLength()來實(shí)現(xiàn)善绎。
綜上所述，從一定意義上講诫尽，Semaphore是一種共享鎖禀酱，屬于AQS基礎(chǔ)抽象隊(duì)列同步器中共享模式孵化的產(chǎn)物，支持公平模式與非公平模式箱锐，默認(rèn)是使用非公平模式比勉。
寫在最后
通過對Java領(lǐng)域中，JDK內(nèi)部提供的各種鎖的實(shí)現(xiàn)來看驹止，一直圍繞的核心主要還是基于AQS基礎(chǔ)同步器來實(shí)現(xiàn)的浩聋，但是AQS基礎(chǔ)同步器不是一種非它不可的技術(shù)標(biāo)準(zhǔn)規(guī)范，更多的只是一套技術(shù)參考指南臊恋。
但是衣洁，實(shí)際上，Java對于鎖的實(shí)現(xiàn)與運(yùn)用遠(yuǎn)遠(yuǎn)不止這些抖仅，還有相位器(Phaser)和交換器(Exchanger),以及在Java JDK1.8版本之前并發(fā)容器ConcurrentHashMap中使用的分段鎖(Segment)坊夫。
不論是何種實(shí)現(xiàn)和應(yīng)用，在Java并發(fā)編程領(lǐng)域來講撤卢，都是圍繞線程安全問題的角度去考慮的环凿，只是針對于各種各樣的業(yè)務(wù)場景做的具體的實(shí)現(xiàn)。
一定意義上來講放吩，對線程加鎖只是并發(fā)編程的實(shí)現(xiàn)方式之一智听，相對于實(shí)際應(yīng)用來說，Java領(lǐng)域中的鎖都只是一種單一應(yīng)用的鎖渡紫，只是給我們掌握J(rèn)ava并發(fā)編程提供一種思想沒到推，三言兩語也不可能詳盡。
到此為止惕澎，這算是對于Java領(lǐng)域中并發(fā)鎖的最終章莉测，文中表述均為個(gè)人看法和個(gè)人理解，如有不到之處唧喉，忘請諒解也請給予批評指正捣卤。
最后忍抽，技術(shù)研究之路任重而道遠(yuǎn)，愿我們熬的每一個(gè)通宵腌零，都撐得起我們想在這條路上走下去的勇氣梯找，未來仍然可期，與各位程序編程君共勉益涧！
版權(quán)聲明：本文為博主原創(chuàng)文章，遵循相關(guān)版權(quán)協(xié)議驯鳖，如若轉(zhuǎn)載或者分享請附上原文出處鏈接和鏈接來源闲询。
