【高并發(fā)】AQS詳解

大家好，我是冰河~~

AQS的全稱為AbstractQueuedSynchronizer，是在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，要么使用了共享鎖的功能，而不會(huì)同時(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ì)列來管理鎖，線程會(huì)首先嘗試獲取鎖，如果失敗，就將當(dāng)前線程以及等待等信息封裝成一個(gè)Node節(jié)點(diǎn)，加入到同步隊(duì)列SyncQueue，接著會(huì)不斷循環(huán)嘗試獲取鎖，獲取鎖的條件是當(dāng)前節(jié)點(diǎn)為Head的直接后繼節(jié)點(diǎn)才會(huì)嘗試獲取鎖，如果失敗，就會(huì)阻塞自己，直到自己被喚醒。而持有鎖的線程釋放鎖的時(shí)候，會(huì)喚醒隊(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()方法的線程會(huì)一直阻塞，直到其他線程調(diào)用該類的countDown()方法，使當(dāng)前計(jì)數(shù)器的值變?yōu)?為止。每次調(diào)用該類的countDown()方法，當(dāng)前計(jì)數(shù)器的值就會(huì)減1。當(dāng)計(jì)數(shù)器的值減為0的時(shí)候，所有因調(diào)用await()方法而處于等待狀態(tài)的線程就會(huì)繼續(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()方法，并不會(huì)第一時(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)的線程會(huì)被喚醒，繼續(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會(huì)優(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不會(huì)比synchronized差。

synchronized的優(yōu)勢：

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

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

（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í)行寫鎖，這就會(huì)導(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不會(huì)引發(fā)死鎖，其他的鎖使用不當(dāng)可能會(huì)引發(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(); } }

好了，今天就到這兒吧，我是冰河，我們下期見~~

Java JDK JVM 任務(wù)調(diào)度 多線程

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。