并發(fā)框架

lvzishen 發(fā)布于2019-08-16 13:57 / 2746人閱讀

摘要：中提供了幾個(gè)比較常用的并發(fā)工具類，比如。是一個(gè)同步工具類，它允許一個(gè)或多個(gè)線程一直等待，直到其他線程的操作執(zhí)行完畢再執(zhí)行。從命名可以解讀到是倒數(shù)的意思，類似于我們倒計(jì)時(shí)的概念。

JUC中提供了幾個(gè)比較常用的并發(fā)工具類，比如CountDownLatch、CyclicBarrier、Semaphore。其實(shí)在以前我們課堂的演示代碼中，或多或少都有用到過(guò)這樣一些api，接下來(lái)我們會(huì)帶大家去深入研究一些常用的api。 CountDownLatch
countdownlatch是一個(gè)同步工具類，它允許一個(gè)或多個(gè)線程一直等待，直到其他線程的操作執(zhí)行完畢再執(zhí)行。從命名可以解讀到countdown是倒數(shù)的意思，類似于我們倒計(jì)時(shí)的概念。 countdownlatch提供了兩個(gè)方法，一個(gè)是countDown，一個(gè)是await， countdownlatch初始化的時(shí)候需要傳入一個(gè)整數(shù)，在這個(gè)整數(shù)倒數(shù)到0之前，調(diào)用了await方法的程序都必須要等待，然后通過(guò)countDown來(lái)倒數(shù)。

public static void main(String[] args) throws InterruptedException {
 
 ? ?CountDownLatch countDownLatch=new CountDownLatch(3);
 
 ? ?new Thread(()->{
 ? ?
 ? ? ? ?countDownLatch.countDown();
 
 ?  },"t1").start();
 
 ? ?new Thread(()->{
 ? ?
 ? ? ? ?countDownLatch.countDown();

 ?  },"t2").start();
 

 ? ?new Thread(()->{

 ? ? ? ?countDownLatch.countDown();

 ?  },"t3").start();

 ? ?countDownLatch.await();
 
 ? ?System.out.println("所有線程執(zhí)行完畢");
 
}

從代碼的實(shí)現(xiàn)來(lái)看，有點(diǎn)類似join的功能，但是比join更加靈活。CountDownLatch構(gòu)造函數(shù)會(huì)接收一個(gè)int類型的參數(shù)作為計(jì)數(shù)器的初始值，當(dāng)調(diào)用CountDownLatch的countDown方法時(shí)，這個(gè)計(jì)數(shù)器就會(huì)減一。通過(guò)await方法去阻塞去阻塞主流程
使用場(chǎng)景
1. 通過(guò)countdownlatch實(shí)現(xiàn)最大的并行請(qǐng)求，也就是可以讓N個(gè)線程同時(shí)執(zhí)行，這個(gè)我也是在課堂上寫(xiě)得比較多的
2. 比如應(yīng)用程序啟動(dòng)之前，需要確保相應(yīng)的服務(wù)已經(jīng)啟動(dòng)，比如我們之前在講zookeeper的時(shí)候，通過(guò)原生api連接的地方有用到countDownLatch 源碼分析
CountDownLatch類存在一個(gè)內(nèi)部類Sync，上節(jié)課我們講過(guò)，它是一個(gè)同步工具，一定繼承了 AbstractQueuedSynchronizer。很顯然，CountDownLatch實(shí)際上是是使得線程阻塞了，既然涉及到阻塞，就一定涉及到AQS隊(duì)列 await
await函數(shù)會(huì)使得當(dāng)前線程在countdownlatch倒計(jì)時(shí)到0之前一直等待，除非線程別中斷；從源碼中可以得知await 方法會(huì)轉(zhuǎn)發(fā)到Sync的acquireSharedInterruptibly
方法
public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } acquireSharedInterruptibly
這塊代碼主要是判斷當(dāng)前線程是否獲取到了共享鎖; 上一節(jié)課提到過(guò)，AQS有兩種鎖類型，一種是共享鎖，一種是獨(dú)占鎖，在這里用的是共享鎖；為什么要用共享鎖，因?yàn)镃ountDownLatch可以多個(gè)線程同時(shí)通過(guò)。

public final void acquireSharedInterruptibly(int arg)
 
 ? ? ? ?throws InterruptedException {
doAcquireSharedInterruptibly 
獲取共享鎖
 
 ? ?if (Thread.interrupted()) //判斷線程是否中斷
 
 ? ? ? ?throw new InterruptedException();
 
 ? ?if (tryAcquireShared(arg) < 0) //如果等于0則返回1，否則返回-1，返回-1表示需要阻塞
 
 ? ? ? ?doAcquireSharedInterruptibly(arg);
 
} 在這里，state的意義是count，如果計(jì)數(shù)器為0，表示不需要阻塞，否則，只有在滿足條件的情況下才會(huì)被喚醒

doAcquireSharedInterruptibly
獲取共享鎖

private void doAcquireSharedInterruptibly(int arg)
 
 ? ?throws InterruptedException {
 
 ? ?final Node node = addWaiter(Node.SHARED); //創(chuàng)建一個(gè)共享模式的節(jié)點(diǎn)添加到隊(duì)列中 ? ?boolean failed = true;
 
 ? ?try {
 
 ? ? ? ?for (;;) { //自旋等待共享鎖釋放，也就是等待計(jì)數(shù)器等于0。
 
 ? ? ? ? ? ?final Node p = node.predecessor(); //獲得當(dāng)前節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)
 
 ? ? ? ? ? ?if (p == head) {
 
 ? ? ? ? ? ? ? ?int r = tryAcquireShared(arg);//就判斷嘗試獲取鎖
 
 ? ? ? ? ? ? ? ?if (r >= 0) {//r>=0表示計(jì)數(shù)器已經(jīng)歸零了，則釋放當(dāng)前的共享鎖
 
 ? ? ? ? ? ? ? ? ? ?setHeadAndPropagate(node, r);
 
 ? ? ? ? ? ? ? ? ? ?p.next = null; // help GC
 
 ? ? ? ? ? ? ? ? ? ?failed = false;
 
 ? ? ? ? ? ? ? ? ? ?return;
 
 ? ? ? ? ? ? ?  }
 
 ? ? ? ? ?  } //當(dāng)前節(jié)點(diǎn)不是頭節(jié)點(diǎn)，則嘗試讓當(dāng)前線程阻塞，第一個(gè)方法是判斷是否需要阻塞，第二個(gè)方法是阻塞
 
 ? ? ? ? ? ?if (shouldParkAfterFailedAcquire(p, node) &&
 
 ? ? ? ? ? ? ? ?parkAndCheckInterrupt())
 
 ? ? ? ? ? ? ? ?throw new InterruptedException();
 ? ? ?  }
 
 ?  } finally {
 
 ? ? ? ?if (failed)
 
 ? ? ? ? ? ?cancelAcquire(node);
 
 ?  }
 
}

待續(xù)。。。。。。。。。