上期回顧：
上次博客我們主要說了我們juc並發包下面的ReetrantLock的一些簡單使用和底層的原理，是如何實現公平鎖、非公平鎖的。內部的雙向鏈表到底是什么意思，prev和next到底是什么，為什么要引入heap和tail來值向null的Node節點。高並發時候是如何保證state來記錄重入鎖的，在我們的上次博客都做了詳細的說明。這次我們來聊一些簡單易懂且實用的AQS中的工具類。
Semaphore信號量：
這個東西很簡單，別看字面意思，什么信號量，我也不懂得那個術語什么意思，Semaphore你可以這樣來理解，我們要去看電影，而且是3D電影（必須戴3D眼鏡才可以進入），但是比較不巧的是我們電影院只有兩個3D眼鏡了，也就是說，我們每次只能進去兩個人看電影，然后等待這兩個人看完電影以后把眼鏡還回來，后面的兩個人才能繼續觀看，就是說每次只允許最多進去兩個人，每次進入到線程獲取鎖，需要你得到前置的票據，才可以進行后續的流程。可以理解為一個簡單的限流吧。我們來一下代碼示例。
public class Test { public static void main(String[] args) throws InterruptedException { Semaphore semaphore = new Semaphore(2); for (int i = 0; i < 5; i++) { new Thread(new Task(semaphore,"xiaocaijishu"+i)).start(); } } static class Task extends Thread{ Semaphore semaphore; public Task(Semaphore semaphore,String tname){ this.semaphore = semaphore; this.setName(tname); } public void run() { try { semaphore.acquire(); System.out.println(Thread.currentThread().getName()+"拿着3D眼鏡進去了,時間是"+System.currentTimeMillis()); Thread.sleep(1000); semaphore.release(); System.out.println(Thread.currentThread().getName()+"出來了,將3D眼鏡還給了服務人員,時間是"+System.currentTimeMillis()); } catch (InterruptedException e) { e.printStackTrace(); } } } }
運行結果就是這樣的
我們來解釋一下運行結果，線程1和線程3同一時間去看電影了，然后1出來了，這時線程9馬上拿着我們的3D眼鏡進去了，過了一會線程3也看完電影了，出來了還了3D眼鏡，線程7又在同一時間拿着3D眼鏡進去看電影了，后續線程都是如此執行的，每次只是進入兩個線程。
簡單的使用看到了，我們來看看底層的源碼設計吧。開始的時候我們是創建一個Semaphore內部票據數目給予的是2。
//1.創建初始票據是2的Semaphore Semaphore semaphore = new Semaphore(2); //2.進入Semaphore,查看數據2是如何存儲的. public Semaphore(int permits) { sync = new NonfairSync(permits); } //3.底層還是基於sync 創建了一個對象,但不同於過去ReetrantLock的是,這次是一個非公平的鎖對象,我們再次進入NonfairSync看看那個數字2到底放在哪里了. Sync(int permits) { setState(permits); } //4.我們可以看到底層還是用State來存儲的.
這次沒有把所有代碼全部粘出來,感覺那樣像是湊篇幅一樣。
通過上述代碼，我們可以看到，我們的初始票據數，是上一次那個state來存的。
后續我們調用了acquire方法來嘗試獲取票據，acquire方法也可以傳入獲取票據數目的比如semaphore.acquire(2);也是可以的。我們進入acquire方法來看看到底是如何獲取的。
//從new Semaphore(2);點擊進入后續方法 public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); } //我們可以看到,當我們沒有傳需要獲取多少票據的時候,會默認給予1這個參數,我們來繼續看后續流程 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); } //Thread.interrupted()判斷當前線程是否已經中斷,如果中斷我直接拋出異常,電影都演完了,我拿3D眼鏡還有毛線用. //tryAcquireShared(arg)嘗試獲取票據,arg是1,剛才給予的默認1 final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } } //內部有實現關系,所以調用的是Semaphore類nonfairTryAcquireShared方法,我們來解讀一下 //直接就是一個死循環, int available = getState();獲取一下當前還有多少票據 // int remaining = available - acquires;計算出當前票據減去所需票據的一個剩余值 //if (remaining < 0 || compareAndSetState(available, remaining))我們現有2個票據,拿走1個,剩余1個,所以remaining < 0 一定是false的 //再來看另一半compareAndSetState,用原子計算(上次博客說過為什么要原子計算)方式來修改剩余票據,這個是可以修改成功的.所以滿足條件可以返回一個2-1 也就是返回一個正數1
是不是有點看懵圈了，很多小伙伴感覺if (remaining < 0 ||compareAndSetState(available, remaining))前面的remaining<0，這個或判斷貌似沒用啊，來張圖解釋一下。
有沒有感覺好點了，自己可以跟着源代碼走一走，獲取的過程就差一個doAcquireSharedInterruptibly還沒有看了，如果獲取超過了票據數，也就是不應該讓返回負數時運行doAcquireSharedInterruptibly方法，我們來看一下。
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.SHARED);//以共享方式添加節點 boolean failed = true; try { for (;;) { final Node p = node.predecessor();//判斷前驅節點是否為空 if (p == head) { int r = tryAcquireShared(arg);//再次嘗試獲取票據 if (r >= 0) {//>= 0表示獲取票據成功 setHeadAndPropagate(node, r);//更改頭節點 p.next = null; // help GC failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && //剔除不可用的Node節點 parkAndCheckInterrupt()) //阻塞當前線程 throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
經過兩次以上的嘗試，我們將該線程阻塞了，不至於一直for循環在運行，也就這樣，票據發放完畢了。
過程差不多就是這樣的，我們可以再仔細看一下是如何添加節點的，上次ReetrantLock說了一些，我們這次再來看一下。我們現已第一次塞節點為例，
private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; if (pred != null) {//第一次一定是空的，我們現在已初始塞節點為例。 node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node);//為空直接進入這個邏輯 return node; }
private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize //1.第一次一定是空 //二次循環不為空 進入else if (compareAndSetHead(new Node()))//2.創建一個空節點,並且作為head節點. tail = head;//3.tail指向那個head節點 } else { node.prev = t;//4. 將node節點的前驅指針指向 if (compareAndSetTail(t, node)) {//5.原子計算方式將node節點后驅節點指向tail t.next = node;//6.將t節點(空節點)的后驅指針指向node節點 return t; } } } }
第一次循環只是一個內部的初始空節點，第二次循環才是移動指針塞入的過程。
節點喚醒是在釋放票據時被喚醒的，代碼超級簡單，可以自己當做一份作業，自己去看一遍代碼吧~！提示流程就是先還票據，然后喚醒。Semaphore差不多就這些知識點，我也帶着大家簡單的看了一遍源碼。我們再來繼續看一下后面AQS的一些工具類。
CountDownLaunch的基本使用
CountDownLaunch很好理解，也是比較實用的，我們干王者農葯的時候就是一個很好的栗子，游戲選完人物大家一起加載地圖等游戲資料，有的人慢，有的人快，這時就印出來了CountDownLaunch，相當於我們5個玩家同時開啟5個線程，然后一起執行，執行完畢先等着，直到5個玩家全部執行完成時，才可以運行后續操作。我們來看一下代碼。
public class CountDownLaunchSample { public static void main(String[] args) throws InterruptedException { CountDownLatch countDownLatch = new CountDownLatch(2); new Thread(new playerOne(countDownLatch)).start(); new Thread(new playerTwo(countDownLatch)).start(); countDownLatch.await(); System.out.println("全部加載完成"); } static class playerOne implements Runnable { CountDownLatch countDownLatch; public playerOne(CountDownLatch countDownLatch) { this.countDownLatch = countDownLatch; } public void run() { try { System.out.println("玩家1開始加載..."); Thread.sleep(2000); System.out.println("玩家1加載完成"); } catch (InterruptedException e) { e.printStackTrace(); } finally { if (countDownLatch != null) countDownLatch.countDown(); } } } static class playerTwo implements Runnable { CountDownLatch countDownLatch; public playerTwo(CountDownLatch countDownLatch) { this.countDownLatch = countDownLatch; } public void run() { try { System.out.println("玩家2開始加載..."); Thread.sleep(10000); System.out.println("玩家2加載完成"); } catch (InterruptedException e) { e.printStackTrace(); } finally { if (countDownLatch != null) countDownLatch.countDown(); } } } }
實際項目中如果遇到讀取excel多個sheet頁簽然后匯總數據的情況也可以采用CountDownLanch。注意最后final的countDownLatch.countDown()方法，也是一個類似上面票據增減的方法。
CyclicBarrier柵欄的簡單使用：
CyclicBarrier和我們上面的CountDownLanch差不多，都是開啟多個任務一起去執行，不同的是CountDownLanch需要支線任務執行完成然后CountDownLanch做一個匯總，然后繼續運行后續程序。CyclicBarrier不需要做匯總。再就是CyclicBarrier是可以重復的。
public class CyclicBarrierTest implements Runnable { private CyclicBarrier cyclicBarrier; private int index ; public CyclicBarrierTest(CyclicBarrier cyclicBarrier, int index) { this.cyclicBarrier = cyclicBarrier; this.index = index; } public void run() { try { System.out.println("index: " + index); index--; cyclicBarrier.await(); } catch (Exception e) { e.printStackTrace(); } } public static void main(String[] args) throws Exception { CyclicBarrier cyclicBarrier = new CyclicBarrier(11, new Runnable() { public void run() { System.out.println("所有特工到達屏障，准備開始執行秘密任務"); } }); for (int i = 0; i < 10; i++) { new Thread(new CyclicBarrierTest(cyclicBarrier, i)).start(); } cyclicBarrier.await(); System.out.println("全部到達屏障...."); } }
這個需要注意的是CyclicBarrier cyclicBarrier = new CyclicBarrier(11, 這個11，就是說一定有11個線程執行完畢，我才可以執行后面的操作，我們下面for循環是10，而我們那里寫的是11啊，別忘記還有一個主線程呢，所以說每次計算一定加一個主線程啊。
Exchanger的簡單使用
最后就是我們Exchanger，平時使用的不多，我們了解一下就可以了，摟一眼代碼，就是線程之間的變量交換。
public static void main(String []args) { final Exchanger<Integer> exchanger = new Exchanger<Integer>(); for(int i = 0 ; i < 4 ; i++) { final Integer num = i; new Thread() { public void run() { System.out.println("我是線程：Thread_" + this.getName() + "我的數據是：" + num); try { Integer exchangeNum = exchanger.exchange(num); Thread.sleep(1000); System.out.println("我是線程：Thread_" + this.getName() + "我原先的數據為：" + num + " , 交換后的數據為：" + exchangeNum); } catch (InterruptedException e) { e.printStackTrace(); } } }.start(); } }
總結：
這次我們核心梳理了我們的Semaphore的執行流程，內部是如何來實現我們的票據計數，獲取，歸還等操作的，再就是我們for無限循環會在兩次以后自動阻塞的設計思想，還有我們的CountDownLanch、CyclicBarrier、Executors的基本使用，並賦予大家簡單的代碼流程，今天就說到這，明天我們繼續來說我們的多線程。
