摘要：AQS 的全稱為 Abstract Queued Synchronizer，是在 J.U.C（java.util.concurrent）下子包中的類。

AQS 的全稱為 Abstract Queued Synchronizer，是在 J.U.C（java.util.concurrent）下子包中的類。

一、AQS 的設(shè)計(jì)如下

（1）使用 Node 實(shí)現(xiàn) FIFO 隊(duì)列，可以用于構(gòu)建鎖或者其他同步裝置的基礎(chǔ)框架。

（2）利用了一個(gè) int 類型表示狀態(tài)

在 AQS 類中，有一個(gè)叫做 state 的成員變量。

基于 AQS 有一個(gè)同步組件 ReentrantLock，在 ReentrantLock 中，state 表示獲取鎖的線程數(shù)。如果 state=0, 則表示還沒有線程獲取鎖；如果 state=1, 則表示有線程獲取了鎖；如果 state>1，則表示重入鎖的數(shù)量。

（3）使用方法是繼承

設(shè)計(jì)上基于模板方法，使用時(shí)需要繼承 AQS，并覆寫其中的方法

（4）子類通過繼承并通過實(shí)現(xiàn)它的方法管理其狀態(tài) {acquire 和 release} 的方法操縱狀態(tài)

（5）可以同時(shí)實(shí)現(xiàn)排它鎖和共享鎖模式（獨(dú)占、共享）

站在使用者的角度，AQS 的功能主要分為兩類：獨(dú)占模式和共享模式。它的所有子類中要么實(shí)現(xiàn)并使用了它的獨(dú)占功能的 API，要么使用了共享鎖的功能，而不會同時(shí)使用兩套 API。即便是它最有名的子類 ——ReentrantReadWriteLock，也是通過兩個(gè)內(nèi)部類 ——ReadLock（讀鎖）和 WriteLock（寫鎖）兩套 API 來實(shí)現(xiàn)的。

二、AQS 內(nèi)部實(shí)現(xiàn)的大體思路

首先，AQS 內(nèi)部維護(hù)了一個(gè) CLH 隊(duì)列來管理鎖，線程會首先嘗試獲取鎖，如果失敗，就將當(dāng)前線程以及等待等信息封裝成一個(gè) Node 節(jié)點(diǎn)，加入到同步隊(duì)列 SyncQueue，接著會不斷循環(huán)嘗試獲取鎖，獲取鎖的條件是當(dāng)前節(jié)點(diǎn)為 Head 的直接后繼節(jié)點(diǎn)才會嘗試獲取鎖，如果失敗，就會阻塞自己，直到自己被喚醒。

而持有鎖的線程釋放鎖的時(shí)候，會喚醒隊(duì)列中的后繼線程。基于這些基礎(chǔ)的設(shè)計(jì)和思路，JDK 提供了許多基于 AQS 的子類，比如：CountDownLatch、Semaphore、CyclicBarrier、ReentrantLock、Condition、FutureTask 等

三、AQS 同步組件

CountDownLatch：閉鎖，通過一個(gè)計(jì)數(shù)，來保證線程是否一直阻塞

Semaphore：控制同一時(shí)間并發(fā)線程的數(shù)目

CyclicBarrier：與 CountDownLatch 類似，都能阻塞進(jìn)程；

ReentrantLock：可重入鎖

Condition: 在使用時(shí)需要 ReentrantLock

FutureTask: 對比 Runnable 和 Callable

1.CountDownLatch

同步輔助類，通過它可以阻塞當(dāng)前線程。也就是說，能夠?qū)崿F(xiàn)一個(gè)線程或者多個(gè)線程一直等待，直到其他線程執(zhí)行的操作完成。使用一個(gè)給定的計(jì)數(shù)器進(jìn)行初始化，該計(jì)數(shù)器的操作是原子操作，即同時(shí)只能有一個(gè)線程操作該計(jì)數(shù)器。
調(diào)用該類 await () 方法的線程會一直阻塞，直到其他線程調(diào)用該類的 countDown () 方法，使當(dāng)前計(jì)數(shù)器的值變?yōu)?0 為止。

每次調(diào)用該類的 countDown () 方法，當(dāng)前計(jì)數(shù)器的值就會減 1。當(dāng)計(jì)數(shù)器的值減為 0 的時(shí)候，所有因調(diào)用 await () 方法而處于等待狀態(tài)的線程就會繼續(xù)往下執(zhí)行。這種操作只能出現(xiàn)一次，因?yàn)樵擃愔械挠?jì)數(shù)器不能被重置。如果需要一個(gè)可以重置計(jì)數(shù)次數(shù)的版本，可以考慮使用 CyclicBarrier 類。

CountDownLatch 支持給定時(shí)間的等待，超過一定的時(shí)間不再等待，使用時(shí)只需要在 await () 方法中傳入需要等待的時(shí)間即可。此時(shí)，await () 方法的方法簽名如下：

public boolean await(long timeout, TimeUnit unit)

CountDownLatch 使用場景 在某些業(yè)務(wù)場景中，程序執(zhí)行需要等待某個(gè)條件完成后才能繼續(xù)執(zhí)行后續(xù)的操作。典型的應(yīng)用為并行計(jì)算：當(dāng)某個(gè)處理的運(yùn)算量很大時(shí)，可以將該運(yùn)算任務(wù)拆分成多個(gè)子任務(wù)，等待所有的子任務(wù)都完成之后，父任務(wù)再拿到所有子任務(wù)的運(yùn)算結(jié)果進(jìn)行匯總。

調(diào)用 ExecutorService 類的 shutdown () 方法，并不會第一時(shí)間內(nèi)把所有線程全部都銷毀掉，而是讓當(dāng)前已有的線程全部執(zhí)行完，之后，再把線程池銷毀掉。 示例代碼如下：

package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CountDownLatchExample {
 private static final int threadCount = 200;
 public static void main(String[] args) throws InterruptedException {
 ExecutorService exec = Executors.newCachedThreadPool();
 final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
 for (int i = 0; i < threadCount; i++){
 final int threadNum = i;
 exec.execute(() -> {
 try {
 test(threadNum);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }finally {
 countDownLatch.countDown();
 }
 });
 }
 countDownLatch.await();
        log.info("finish");
 exec.shutdown();
 }
 private static void test(int threadNum) throws InterruptedException {
 Thread.sleep(100);
 log.info("{}", threadNum);
 Thread.sleep(100);
 }
}

支持給定時(shí)間等待的示例代碼如下：

package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Slf4j
public class CountDownLatchExample {
 private static final int threadCount = 200;
 public static void main(String[] args) throws InterruptedException {
 ExecutorService exec = Executors.newCachedThreadPool();
 final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
 for (int i = 0; i < threadCount; i++){
 final int threadNum = i;
 exec.execute(() -> {
 try {
 test(threadNum);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }finally {
 countDownLatch.countDown();
 }
 });
 }
 countDownLatch.await(10, TimeUnit.MICROSECONDS);
        log.info("finish");
 exec.shutdown();
 }
 private static void test(int threadNum) throws InterruptedException {
 Thread.sleep(100);
 log.info("{}", threadNum);
 }
}

2.Semaphore

控制同一時(shí)間并發(fā)線程的數(shù)目。能夠完成對于信號量的控制，可以控制某個(gè)資源可被同時(shí)訪問的個(gè)數(shù)。 提供了兩個(gè)核心方法 ——acquire () 方法和 release () 方法。acquire () 方法表示獲取一個(gè)許可，如果沒有則等待，release () 方法則是在操作完成后釋放對應(yīng)的許可。

Semaphore 維護(hù)了當(dāng)前訪問的個(gè)數(shù)，通過提供同步機(jī)制來控制同時(shí)訪問的個(gè)數(shù)。Semaphore 可以實(shí)現(xiàn)有限大小的鏈表。 Semaphore 使用場景如 Semaphore 常用于僅能提供有限訪問的資源，比如：數(shù)據(jù)庫連接數(shù) 每次獲取并釋放一個(gè)許可，示例代碼如下：

package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
 private static final int threadCount = 200;
 public static void main(String[] args) throws InterruptedException {
 ExecutorService exec = Executors.newCachedThreadPool();
 final Semaphore semaphore = new Semaphore(3);
 for (int i = 0; i < threadCount; i++){
 final int threadNum = i;
 exec.execute(() -> {
 try {
 semaphore.acquire(); //獲取一個(gè)許可
 test(threadNum);
 semaphore.release(); //釋放一個(gè)許可
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 });
 }
 exec.shutdown();
 }
 private static void test(int threadNum) throws InterruptedException {
 log.info("{}", threadNum);
 Thread.sleep(1000);
 }
}

每次獲取并釋放多個(gè)許可，示例代碼如下：

package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
 private static final int threadCount = 200;
 public static void main(String[] args) throws InterruptedException {
 ExecutorService exec = Executors.newCachedThreadPool();
 final Semaphore semaphore = new Semaphore(3);
 for (int i = 0; i < threadCount; i++){
 final int threadNum = i;
 exec.execute(() -> {
 try {
 semaphore.acquire(3); //獲取多個(gè)許可
 test(threadNum);
 semaphore.release(3); //釋放多個(gè)許可
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 });
 }
        log.info("finish");
 exec.shutdown();
 }
 private static void test(int threadNum) throws InterruptedException {
 log.info("{}", threadNum);
 Thread.sleep(1000);
 }
}

假設(shè)有這樣一個(gè)場景，并發(fā)太高了，即使使用 Semaphore 進(jìn)行控制，處理起來也比較棘手。假設(shè)系統(tǒng)當(dāng)前允許的最高并發(fā)數(shù)是 3，超過 3 后就需要丟棄，使用 Semaphore 也能實(shí)現(xiàn)這樣的場景，示例代碼如下：

package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
 private static final int threadCount = 200;
 public static void main(String[] args) throws InterruptedException {
 ExecutorService exec = Executors.newCachedThreadPool();
 final Semaphore semaphore = new Semaphore(3);
 for (int i = 0; i < threadCount; i++){
 final int threadNum = i;
 exec.execute(() -> {
 try {
 //嘗試獲取一個(gè)許可，也可以嘗試獲取多個(gè)許可，
 //支持嘗試獲取許可超時(shí)設(shè)置，超時(shí)后不再等待后續(xù)線程的執(zhí)行
 //具體可以參見Semaphore的源碼
 if (semaphore.tryAcquire()) { 
 test(threadNum);
 semaphore.release(); //釋放一個(gè)許可
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 });
 }
        log.info("finish");
 exec.shutdown();
 }
 private static void test(int threadNum) throws InterruptedException {
 log.info("{}", threadNum);
 Thread.sleep(1000);
 }
}

3.CyclicBarrier

是一個(gè)同步輔助類，允許一組線程相互等待，直到到達(dá)某個(gè)公共的屏障點(diǎn)，通過它可以完成多個(gè)線程之間相互等待，只有當(dāng)每個(gè)線程都準(zhǔn)備就緒后，才能各自繼續(xù)往下執(zhí)行后面的操作。

與 CountDownLatch 有相似的地方，都是使用計(jì)數(shù)器實(shí)現(xiàn)，當(dāng)某個(gè)線程調(diào)用了 CyclicBarrier 的 await () 方法后，該線程就進(jìn)入了等待狀態(tài)，而且計(jì)數(shù)器執(zhí)行加 1 操作，當(dāng)計(jì)數(shù)器的值達(dá)到了設(shè)置的初始值，調(diào)用 await () 方法進(jìn)入等待狀態(tài)的線程會被喚醒，繼續(xù)執(zhí)行各自后續(xù)的操作。

CyclicBarrier 在釋放等待線程后可以重用，所以，CyclicBarrier 又被稱為循環(huán)屏障。

CyclicBarrier 使用場景

可以用于多線程計(jì)算數(shù)據(jù)，最后合并計(jì)算結(jié)果的場景

CyclicBarrier 與 CountDownLatch 的區(qū)別

（1）CountDownLatch 的計(jì)數(shù)器只能使用一次，而 CyclicBarrier 的計(jì)數(shù)器可以使用 reset () 方法進(jìn)行重置，并且可以循環(huán)使用

（2）CountDownLatch 主要實(shí)現(xiàn) 1 個(gè)或 n 個(gè)線程需要等待其他線程完成某項(xiàng)操作之后，才能繼續(xù)往下執(zhí)行，描述的是 1 個(gè)或 n 個(gè)線程等待其他線程的關(guān)系。而 CyclicBarrier 主要實(shí)現(xiàn)了多個(gè)線程之間相互等待，直到所有的線程都滿足了條件之后，才能繼續(xù)執(zhí)行后續(xù)的操作，描述的是各個(gè)線程內(nèi)部相互等待的關(guān)系。

（3）CyclicBarrier 能夠處理更復(fù)雜的場景，如果計(jì)算發(fā)生錯(cuò)誤，可以重置計(jì)數(shù)器讓線程重新執(zhí)行一次。

CyclicBarrier 中提供了很多有用的方法，比如：可以通過 getNumberWaiting () 方法獲取阻塞的線程數(shù)量，通過 isBroken () 方法判斷阻塞的線程是否被中斷。 示例代碼如下：

package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {
 private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
 public static void main(String[] args) throws Exception {
 ExecutorService executorService = Executors.newCachedThreadPool();
 for (int i = 0; i < 10; i++){
 final int threadNum = i;
 Thread.sleep(1000);
 executorService.execute(() -> {
 try {
 race(threadNum);
 } catch (Exception e) {
 e.printStackTrace();
 }
 });
 }
executorService.shutdown();
 }
 private static void race(int threadNum) throws Exception{
 Thread.sleep(1000);
 log.info("{} is ready", threadNum);
 cyclicBarrier.await();
 log.info("{} continue", threadNum);
 }
}

設(shè)置等待超時(shí)示例代碼如下：

package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
@Slf4j
public class CyclicBarrierExample {
 private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
 public static void main(String[] args) throws Exception {
 ExecutorService executorService = Executors.newCachedThreadPool();
 for (int i = 0; i < 10; i++){
 final int threadNum = i;
 Thread.sleep(1000);
 executorService.execute(() -> {
 try {
 race(threadNum);
 } catch (Exception e) {
 e.printStackTrace();
 }
 });
 }
 executorService.shutdown();
 }
 private static void race(int threadNum) throws Exception{
 Thread.sleep(1000);
 log.info("{} is ready", threadNum);
 try{
 cyclicBarrier.await(2000, TimeUnit.MILLISECONDS);
 }catch (BrokenBarrierException | TimeoutException e){
 log.warn("BarrierException", e);
 }
 log.info("{} continue", threadNum);
 }
}

在聲明 CyclicBarrier 的時(shí)候，還可以指定一個(gè) Runnable，當(dāng)線程達(dá)到屏障的時(shí)候，可以優(yōu)先執(zhí)行 Runnable 中的方法。 示例代碼如下：

package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {
 private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> {
 log.info("callback is running");
 });
 public static void main(String[] args) throws Exception {
 ExecutorService executorService = Executors.newCachedThreadPool();
 for (int i = 0; i < 10; i++){
 final int threadNum = i;
 Thread.sleep(1000);
 executorService.execute(() -> {
 try {
 race(threadNum);
 } catch (Exception e) {
 e.printStackTrace();
 }
 });
 }
 executorService.shutdown();
 }
 private static void race(int threadNum) throws Exception{
 Thread.sleep(1000);
 log.info("{} is ready", threadNum);
 cyclicBarrier.await();
 log.info("{} continue", threadNum);
 }
}

4.ReentrantLock 與鎖

Java 中主要分為兩類鎖，一類是 synchronized 修飾的鎖，另外一類就是 J.U.C 中提供的鎖。J.U.C 中提供的核心鎖就是 ReentrantLock。 ReentrantLock（可重入鎖）與 synchronized 區(qū)別：

（1）可重入性

二者都是同一個(gè)線程進(jìn)入 1 次，鎖的計(jì)數(shù)器就自增 1，需要等到鎖的計(jì)數(shù)器下降為 0 時(shí)，才能釋放鎖。

（2）鎖的實(shí)現(xiàn)

synchronized 是基于 JVM 實(shí)現(xiàn)的，而 ReentrantLock 是 JDK 實(shí)現(xiàn)的

（3）性能的區(qū)別

synchronized 優(yōu)化之前性能比 ReentrantLock 差很多，但是自從 synchronized 引入了偏向鎖，輕量級鎖也就是自旋鎖后，性能就差不多了。

（4）功能區(qū)別

便利性：synchronized 使用起來比較方便，并且由編譯器保證加鎖和釋放鎖；ReentrantLock 需要手工聲明加鎖和釋放鎖，最好是在 finally 代碼塊中聲明釋放鎖。

鎖的靈活度和細(xì)粒度：在這點(diǎn)上 ReentrantLock 會優(yōu)于 synchronized

ReentrantLock 獨(dú)有的功能如下： （1）ReentrantLock 可指定是公平鎖還是非公平鎖。而 synchronized 只能是非公平鎖。所謂的公平鎖就是先等待的線程先獲得鎖。

（2）提供了一個(gè) Condition 類，可以分組喚醒需要喚醒的線程。而 synchronized 只能隨機(jī)喚醒一個(gè)線程，或者喚醒全部的線程

（3）提供能夠中斷等待鎖的線程的機(jī)制，lock.lockInterruptibly ()。ReentrantLock 實(shí)現(xiàn)是一種自旋鎖，通過循環(huán)調(diào)用 CAS 操作來實(shí)現(xiàn)加鎖，性能上比較好是因?yàn)楸苊饬耸咕€程進(jìn)入內(nèi)核態(tài)的阻塞狀態(tài)。 synchronized 能做的事情 ReentrantLock 都能做，而 ReentrantLock 有些能做的事情，synchronized 不能做。

在性能上，ReentrantLock 不會比 synchronized 差。 synchronized 的優(yōu)勢：

（1）不用手動(dòng)釋放鎖，JVM 自動(dòng)處理，如果出現(xiàn)異常，JVM 也會自動(dòng)釋放鎖

（2）JVM 用 synchronized 進(jìn)行管理鎖定請求和釋放時(shí)，JVM 在生成線程轉(zhuǎn)儲時(shí)能夠鎖定信息，這些對調(diào)試非常有價(jià)值，因?yàn)樗鼈兡軜?biāo)識死鎖或者其他異常行為的來源。而 ReentrantLock 只是普通的類，JVM 不知道具體哪個(gè)線程擁有 lock 對象。

（3）synchronized 可以在所有 JVM 版本中工作，ReentrantLock 在某些 1.5 之前版本的 JVM 中可能不支持 ReentrantLock 中的部分方法說明：

boolean tryLock (): 僅在調(diào)用時(shí)鎖定未被另一個(gè)線程保持的情況下才獲取鎖定

boolean tryLock (long, TimeUnit): 如果鎖定在給定的等待時(shí)間內(nèi)沒有被另一個(gè)線程保持，且當(dāng)前線程沒有被中斷，則獲取這個(gè)鎖定。

void lockInterruptibly (): 如果當(dāng)前線程沒有被中斷，就獲取鎖定；如果被中斷，就拋出異常

boolean isLocked (): 查詢此鎖定是否由任意線程保持

boolean isHeldByCurrentThread (): 查詢當(dāng)前線程是否保持鎖定狀態(tài)；

boolean isFair (): 判斷是否是公平鎖

boolean hasQueuedThread (Thread)：查詢指定線程是否在等待獲取此鎖定

boolean hasQueuedThreads (): 查詢是否有線程正在等待獲取此鎖定

boolean getHoldCount (): 查詢當(dāng)前線程保持鎖定的個(gè)數(shù)

示例代碼如下：

package io.binghe.concurrency.example.lock;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class LockExample {
 //請求總數(shù)
 public static int clientTotal = 5000;
 //同時(shí)并發(fā)執(zhí)行的線程數(shù)
 public static int threadTotal = 200;
 public static int count = 0;
 private static final Lock lock = new ReentrantLock();
 public static void main(String[] args) throws InterruptedException {
 ExecutorService executorService = Executors.newCachedThreadPool();
 final Semaphore semaphore = new Semaphore(threadTotal);
 final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
 for(int i = 0; i < clientTotal; i++){
 executorService.execute(() -> {
 try{
 semaphore.acquire();
 add();
 semaphore.release();
 }catch (Exception e){
 log.error("exception", e);
 }
 countDownLatch.countDown();
 });
 }
 countDownLatch.await();
 executorService.shutdown();
        log.info("count:{}", count);
 }
 private static void add(){
 lock.lock();
 try{
            count ++;
 }finally {
 lock.unlock();
 }
 }
}

5.ReentrantReadWriteLock

在沒有任何讀寫鎖的時(shí)候，才可以取得寫鎖。如果一直有讀鎖存在，則無法執(zhí)行寫鎖，這就會導(dǎo)致寫鎖饑餓。

示例代碼如下：

package io.binghe.concurrency.example.lock;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@Slf4j
public class LockExample {
 private final Map map = new TreeMap<>();
 private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
 private final Lock readLock = lock.readLock();
 private final Lock writeLock = lock.writeLock();
 public Data get(String key){
 readLock.lock();
 try{
 return map.get(key);
 }finally {
 readLock.unlock();
 }
 }
 public Set getAllKeys(){
 readLock.lock();
 try{
 return map.keySet();
 }finally {
 readLock.unlock();
 }
 }
 public Data put(String key, Data value){
 writeLock.lock();
 try{
 return map.put(key, value);
 }finally {
 writeLock.unlock();
 }
 }
 class Data{
 }
}

6.StampedLock

控制鎖三種模式：寫、讀、樂觀讀。 StampedLock 的狀態(tài)由版本和模式兩個(gè)部分組成，鎖獲取方法返回的是一個(gè)數(shù)字作為票據(jù)，用相應(yīng)的鎖狀態(tài)來表示并控制相關(guān)的訪問，數(shù)字 0 表示沒有寫鎖被授權(quán)訪問。

在讀鎖上分為悲觀鎖和樂觀鎖，樂觀讀就是在讀操作很多，寫操作很少的情況下，可以樂觀的認(rèn)為寫入和讀取同時(shí)發(fā)生的幾率很小。

因此，不悲觀的使用完全的讀取鎖定。程序可以查看讀取資料之后，是否遭到寫入進(jìn)行了變更，再采取后續(xù)的措施，這樣的改進(jìn)可以大幅度提升程序的吞吐量。

總之，在讀線程越來越多的場景下，StampedLock 大幅度提升了程序的吞吐量。 StampedLock 源碼中的案例如下，這里加上了注釋

class Point {
private double x, y;
private final StampedLock sl = new StampedLock();
void move(double deltaX, double deltaY) { // an exclusively locked method
long stamp = sl.writeLock();
try {
x += deltaX;
y += deltaY;
} finally {
sl.unlockWrite(stamp);
}
}
//下面看看樂觀讀鎖案例
double distanceFromOrigin() { // A read-only method
long stamp = sl.tryOptimisticRead(); //獲得一個(gè)樂觀讀鎖
double currentX = x, currentY = y; //將兩個(gè)字段讀入本地局部變量
if (!sl.validate(stamp)) { //檢查發(fā)出樂觀讀鎖后同時(shí)是否有其他寫鎖發(fā)生？
stamp = sl.readLock(); //如果沒有，我們再次獲得一個(gè)讀悲觀鎖
try {
currentX = x; // 將兩個(gè)字段讀入本地局部變量
currentY = y; // 將兩個(gè)字段讀入本地局部變量
} finally {
sl.unlockRead(stamp);
}
}
return Math.sqrt(currentX * currentX + currentY * currentY);
}
//下面是悲觀讀鎖案例
void moveIfAtOrigin(double newX, double newY) { // upgrade
// Could instead start with optimistic, not read mode
long stamp = sl.readLock();
try {
while (x == 0.0 && y == 0.0) { //循環(huán)，檢查當(dāng)前狀態(tài)是否符合
long ws = sl.tryConvertToWriteLock(stamp); //將讀鎖轉(zhuǎn)為寫鎖
if (ws != 0L) { //這是確認(rèn)轉(zhuǎn)為寫鎖是否成功
stamp = ws; //如果成功 替換票據(jù)
x = newX; //進(jìn)行狀態(tài)改變
y = newY; //進(jìn)行狀態(tài)改變
break;
} else { //如果不能成功轉(zhuǎn)換為寫鎖
sl.unlockRead(stamp); //我們顯式釋放讀鎖
stamp = sl.writeLock(); //顯式直接進(jìn)行寫鎖 然后再通過循環(huán)再試
}
}
} finally {
sl.unlock(stamp); //釋放讀鎖或?qū)戞i
}
}
}

示例代碼如下：

package io.binghe.concurrency.example.lock;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.StampedLock;
@Slf4j
public class LockExample {
 //請求總數(shù)
 public static int clientTotal = 5000;
 //同時(shí)并發(fā)執(zhí)行的線程數(shù)
 public static int threadTotal = 200;
 public static int count = 0;
 private static final StampedLock lock = new StampedLock();
 public static void main(String[] args) throws InterruptedException {
 ExecutorService executorService = Executors.newCachedThreadPool();
 final Semaphore semaphore = new Semaphore(threadTotal);
 final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
 for(int i = 0; i < clientTotal; i++){
 executorService.execute(() -> {
 try{
 semaphore.acquire();
 add();
 semaphore.release();
 }catch (Exception e){
 log.error("exception", e);
 }
 countDownLatch.countDown();
 });
 }
 countDownLatch.await();
 executorService.shutdown();
        log.info("count:{}", count);
 }
 private static void add(){
//加鎖時(shí)返回一個(gè)long類型的票據(jù)
 long stamp = lock.writeLock();
 try{
            count ++;
 }finally {
 //釋放鎖的時(shí)候帶上加鎖時(shí)返回的票據(jù)
 lock.unlock(stamp);
 }
 }
}

總結(jié)：

（1）當(dāng)只有少量競爭者時(shí)，synchronized 是一個(gè)很好的通用鎖實(shí)現(xiàn)

（2）競爭者不少，但是線程的增長趨勢是可預(yù)估的，此時(shí)，ReentrantLock 是一個(gè)很好的通用鎖實(shí)現(xiàn)

（3）synchronized 不會引發(fā)死鎖，其他的鎖使用不當(dāng)可能會引發(fā)死鎖。

7.Condition

Condition 是一個(gè)多線程間協(xié)調(diào)通信的工具類，Condition 除了實(shí)現(xiàn) wait 和 notify 的功能以外，它的好處在于一個(gè) lock 可以創(chuàng)建多個(gè) Condition，可以選擇性的通知 wait 的線程 特點(diǎn)：

（1）Condition 的前提是 Lock，由 AQS 中 newCondition () 方法 創(chuàng)建 Condition 的對象

（2）Condition await 方法表示線程從 AQS 中移除，并釋放線程獲取的鎖，并進(jìn)入 Condition 等待隊(duì)列中等待，等待被 signal

（3）Condition signal 方法表示喚醒對應(yīng) Condition 等待隊(duì)列中的線程節(jié)點(diǎn)，并加入 AQS 中，準(zhǔn)備去獲取鎖。

示例代碼如下：

package io.binghe.concurrency.example.lock;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class LockExample {
 public static void main(String[] args) {
 ReentrantLock reentrantLock = new ReentrantLock();
 Condition condition = reentrantLock.newCondition();
 new Thread(() -> {
 try {
 reentrantLock.lock();
 log.info("wait signal"); // 1
 condition.await();
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("get signal"); // 4
 reentrantLock.unlock();
 }).start();
 new Thread(() -> {
 reentrantLock.lock();
 log.info("get lock"); // 2
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 condition.signalAll();
 log.info("send signal ~ "); // 3
 reentrantLock.unlock();
 }).start();
 }
}

審核編輯：劉清