源碼倉庫Gitee · GitHub一.Semaphore1.概念一個計數(shù)信號量。在概念上信號量維持一組許可證。如果有必要每個acquire()都會阻塞直到許可證可用然后才能使用它。每個release()添加許可證潛在地釋放阻塞獲取方。但是沒有使用實際的許可證對象;Semaphore只保留可用數(shù)量的計數(shù)并相應地執(zhí)行。即一個Semaphore維護了一組permits【許可證】。每次調(diào)用acquire()方法都會阻塞直到獲取到許可證。每次調(diào)用release()方法都會添加一個許可證也就是釋放一個被阻塞的獲取者。但是實際上并不存在這個許可證Semaphore僅僅是記錄可用資源的數(shù)量并且做出對應的行為有資源就獲取沒有資源就阻塞。信號量通常用于限制線程數(shù)而不是訪問某些物理或邏輯資源。線程池控制的是線程數(shù)量而信號量控制的是并發(fā)數(shù)量雖然說看起來一樣但兩者還是有區(qū)別的。信號量類似于鎖機制信號量的調(diào)用當達到數(shù)量后線程還是存在的只是被掛起了而已。而線程池同時執(zhí)行的線程數(shù)量是固定的超過了數(shù)量的只能等待。在獲得項目之前每個線程必須從信號量獲取許可證以確保某個項目可用。當線程完成該項目后它將返回到池中并將許可證返回到信號量允許另一個線程獲取該項目。請注意當調(diào)用acquire()時不會保持同步鎖定因為這將阻止某個項目返回到池中。信號量封裝了限制對池的訪問所需的同步與保持池本身一致性所需的任何同步分開。【即將限制對池的訪問和對池中數(shù)據(jù)的操作所需要的鎖分開】。信號量被初始化為一個并且被使用使得它只有至多一個允許可用可以用作互斥鎖。這通常被稱為二進制信號量因為它只有兩個狀態(tài)一個許可證可用或零個許可證可用。當以這種方式使用時二進制信號量具有屬性與許多Lock實現(xiàn)不同“鎖”可以由除所有者之外的線程釋放因為信號量沒有所有權(quán)概念。這在某些專門的上下文中是有用的例如死鎖恢復。Semaphore(int permits) 創(chuàng)建一個 Semaphore與給定數(shù)量的許可證和非公平公平設置。 Semaphore(int permits, boolean fair) 創(chuàng)建一個 Semaphore與給定數(shù)量的許可證和給定的公平設置。此類的構(gòu)造函數(shù)可選擇接受公平參數(shù)。當設置為false時此類不會保證線程獲取許可的順序。特別是闖入是允許的也就是說一個線程調(diào)用acquire()可以提前已經(jīng)等待線程分配的許可證-在等待線程隊列的頭部邏輯新的線程將自己【新線程將自己放在等待線程隊列的最前面】。當公平設置為真時信號量保證調(diào)用acquire方法的線程被選擇以按照它們調(diào)用這些方法的順序獲得許可先進先出; FIFO【FIFO的順序是指是依據(jù)到達方法內(nèi)部的執(zhí)行點的時間并不是方法執(zhí)行的時間?！?。請注意FIFO排序必須適用于這些方法中的特定內(nèi)部執(zhí)行點。因此一個線程可以在另一個線程之前調(diào)用acquire但是在另一個線程之后到達排序點并且類似地從方法返回。另請注意未定義的tryAcquire方法不符合公平性設置但將采取任何可用的許可證。【不定時的tryAcquire()方法會任意選取可用的許可證。】【非公平鎖可以插隊獲取運行公平鎖按照線程順序執(zhí)行?！客ǔＳ糜诳刂瀑Y源訪問的信號量應該被公平地初始化以確保線程沒有被訪問資源【確保沒有線程因為長時間獲取不到許可證而餓死】。當使用信號量進行其他類型的同步控制時非正常排序的吞吐量優(yōu)勢往往超過公平性。2.用法線程可以通過acquire()方法獲取到一個許可然后對共享資源進行操作如果許可集已分配完了那么線程將進入等待狀態(tài)直到其他線程釋放許可才有機會再獲取許可線程釋放一個許可通過release()方法完成許可將被歸還給Semaphore。public static void main(String[] args) { ExecutorService service Executors.newCachedThreadPool(); final Semaphore sp new Semaphore(3); for (int i 0; i 7; i) { Runnable runnable () - { try { sp.acquire(); } catch (InterruptedException e1) { e1.printStackTrace(); } System.out.println(線程 Thread.currentThread().getName() 進入當前已有 (3 - sp.availablePermits()) 個并發(fā)); try { Thread.sleep((long) (Math.random() * 10000)); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(線程 Thread.currentThread().getName() 即將離開); sp.release(); //下面代碼有時候執(zhí)行不準確因為其沒有和上面的代碼合成原子單元 System.out.println(線程 Thread.currentThread().getName() 已離開當前已有 (3 - sp.availablePermits()) 個并發(fā)); }; service.execute(runnable); } } 結(jié)果 線程pool-1-thread-1進入當前已有1個并發(fā) 線程pool-1-thread-2進入當前已有2個并發(fā) 線程pool-1-thread-3進入當前已有3個并發(fā) 線程pool-1-thread-3即將離開 線程pool-1-thread-4進入當前已有3個并發(fā) 線程pool-1-thread-3已離開當前已有3個并發(fā) 線程pool-1-thread-1即將離開 線程pool-1-thread-1已離開當前已有2個并發(fā) 線程pool-1-thread-5進入當前已有3個并發(fā) 線程pool-1-thread-5即將離開 線程pool-1-thread-5已離開當前已有2個并發(fā) 線程pool-1-thread-6進入當前已有3個并發(fā) 線程pool-1-thread-4即將離開 線程pool-1-thread-4已離開當前已有2個并發(fā) 線程pool-1-thread-7進入當前已有3個并發(fā) 線程pool-1-thread-2即將離開 線程pool-1-thread-2已離開當前已有2個并發(fā) 線程pool-1-thread-7即將離開 線程pool-1-thread-7已離開當前已有1個并發(fā) 線程pool-1-thread-6即將離開 線程pool-1-thread-6已離開當前已有0個并發(fā)3.acquire解析acquire() 從該信號量獲取許可證阻止直到可用或線程為 interrupted 。 void acquire(int permits) 從該信號量獲取給定數(shù)量的許可證阻止直到所有可用否則線程為 interrupted 。public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1);//調(diào)用AQS的acquireSharedInterruptibly } /** * AQS的acquireSharedInterruptibly * Acquires in shared mode, aborting if interrupted. Implemented * by first checking interrupt status, then invoking at least once * {link #tryAcquireShared}, returning on success. Otherwise the * thread is queued, possibly repeatedly blocking and unblocking, * invoking {link #tryAcquireShared} until success or the thread * is interrupted. * 以共享模式獲取如果中斷被中止。 * 實現(xiàn)首先檢查中斷狀態(tài)然后至少調(diào)用一次tryacquirered成功返回。 * 否則,線程排隊可能會重復阻塞和取消阻塞 * 調(diào)用tryacquiremred直到成功或線程被打斷了。 */ public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted())//中斷拋出異常 throw new InterruptedException(); if (tryAcquireShared(arg) 0)//獲取失敗加入同步隊列等待 doAcquireSharedInterruptibly(arg); } //由Semaphore的FairSync或NonfairSync實現(xiàn)共享模式下資源可以被多個線程通知占用tryAcquireShared返回int類型表示還有多少個資源可以同時被占用用于共享模式下傳播喚醒。 protected int tryAcquireShared(int arg) { throw new UnsupportedOperationException(); } //以共享中斷模式獲取 private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node addWaiter(Node.SHARED);//創(chuàng)建當前線程的節(jié)點并且鎖的模型是共享鎖將其添加到AQS CLH隊列的末尾 boolean failed true; try { for (;;) {//自旋 final Node p node.predecessor();//獲得當前節(jié)點的前驅(qū)節(jié)點 if (p head) {//是頭節(jié)點沒有等待節(jié)點 int r tryAcquireShared(arg); if (r 0) {//獲取成功當前節(jié)點設置為頭節(jié)點并傳播【傳播指的是同步狀態(tài)剩余的許可數(shù)值不為0通知后續(xù)結(jié)點繼續(xù)獲取同步狀態(tài)】 setHeadAndPropagate(node, r); p.next null; // help GC failed false; return; } } //前繼節(jié)點非head節(jié)點沒資源獲取將前繼節(jié)點狀態(tài)設置為SIGNAL通過park掛起node節(jié)點的線程 if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node);//結(jié)束該結(jié)點線程的請求 } }public void acquire(int permits) throws InterruptedException { if (permits 0) throw new IllegalArgumentException();//數(shù)量小于0拋出異常 sync.acquireSharedInterruptibly(permits);//調(diào)用AQS的acquireSharedInterruptibly }tryAcquireSharedstatic final class FairSync extends Sync {//公平鎖獲取 protected int tryAcquireShared(int acquires) { for (;;) {//自旋 //有前驅(qū)節(jié)點表示當前線程前面有阻塞線程當前線程獲取失敗先讓前節(jié)點線程獲取運行【比非公平鎖獲取多了判斷前驅(qū)節(jié)點的操作】 if (hasQueuedPredecessors()) return -1; int available getState();//可獲取的許可證數(shù)量 int remaining available - acquires;//剩下的許可證數(shù)量 //如果剩余數(shù)量小于0或更新剩余數(shù)量成功返回剩余數(shù)量 if (remaining 0 || compareAndSetState(available, remaining)) return remaining; } } } static final class NonfairSync extends Sync {//非公平鎖獲取 protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires);//調(diào)用Semaphore的內(nèi)部類Sync的nonfairTryAcquireShared } } final int nonfairTryAcquireShared(int acquires) { for (;;) {//自旋 int available getState(); int remaining available - acquires; if (remaining 0 || compareAndSetState(available, remaining)) return remaining; } }4.release解析public void release() { sync.releaseShared(1);//調(diào)用AQS的releaseShared } public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {//釋放同步狀態(tài)成功 doReleaseShared();//喚醒同步隊列中后繼結(jié)點的線程 return true; } return false; } protected boolean tryReleaseShared(int arg) {//由Semaphore的Sync實現(xiàn) throw new UnsupportedOperationException(); } private void doReleaseShared() { /* * Ensure that a release propagates, even if there are other * in-progress acquires/releases. This proceeds in the usual * way of trying to unparkSuccessor of head if it needs * signal. But if it does not, status is set to PROPAGATE to * ensure that upon release, propagation continues. * Additionally, we must loop in case a new node is added * while we are doing this. Also, unlike other uses of * unparkSuccessor, we need to know if CAS to reset status * fails, if so rechecking. * 保證釋放動作(向同步等待隊列尾部)傳遞即使沒有其他正在進行的 * 請求或釋放動作。如果頭節(jié)點的后繼節(jié)點需要喚醒那么執(zhí)行喚醒 * 動作如果不需要將頭結(jié)點的等待狀態(tài)設置為PROPAGATE保證 * 喚醒傳遞。另外為了防止過程中有新節(jié)點進入(隊列)這里必 * 需做循環(huán)所以和其他unparkSuccessor方法使用方式不一樣 * 的是如果(頭結(jié)點)等待狀態(tài)設置失敗重新檢測。 */ for (;;) { Node h head; if (h ! null h ! tail) { int ws h.waitStatus;//頭節(jié)點狀態(tài) // 如果頭節(jié)點對應的線程是SIGNAL狀態(tài)則意味著頭 //結(jié)點的后繼結(jié)點所對應的線程需要被unpark-喚醒。 if (ws Node.SIGNAL) { // 修改頭結(jié)點對應的線程狀態(tài)設置為0。失敗的話則繼續(xù)循環(huán)。 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases // 喚醒頭結(jié)點h的后繼結(jié)點所對應的線程 unparkSuccessor(h); } else if (ws 0 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } // 如果頭結(jié)點發(fā)生變化則繼續(xù)循環(huán)。否則退出循環(huán)。 if (h head) // loop if head changed break; } } //喚醒傳入結(jié)點的后繼結(jié)點對應的線程 private void unparkSuccessor(Node node) { int ws node.waitStatus; if (ws 0) compareAndSetWaitStatus(node, ws, 0); //拿到后繼結(jié)點 Node s node.next; if (s null || s.waitStatus 0) { s null; for (Node t tail; t ! null t ! node; t t.prev) if (t.waitStatus 0) s t; } if (s ! null) //喚醒該線程 LockSupport.unpark(s.thread); } protected final boolean tryReleaseShared(int releases) { for (;;) {//自旋 int current getState();//獲取當前同步狀態(tài) int next current releases;//狀態(tài)1 if (next current) // overflow throw new Error(Maximum permit count exceeded); if (compareAndSetState(current, next))//更新狀態(tài)成功返回true return true; } }二.Exchanger1.概念線程可以在成對內(nèi)配對和交換元素的同步點。每個線程在輸入exchange方法時提供一些對象與合作者線程匹配并在返回時接收其合作伙伴的對象。交換器可以被視為一個的雙向形式SynchronousQueue。交換器在諸如遺傳算法和管道設計的應用中可能是有用的。一個用于兩個工作線程之間交換數(shù)據(jù)的封裝工具類簡單說就是一個線程在完成一定的事務后想與另一個線程交換數(shù)據(jù)則第一個先拿出數(shù)據(jù)的線程會一直等待第二個線程直到第二個線程拿著數(shù)據(jù)到來時才能彼此交換對應數(shù)據(jù)。2.用法ExchangerV 泛型類型其中 V 表示可交換的數(shù)據(jù)類型V exchange(V v)等待另一個線程到達此交換點除非當前線程被中斷然后將給定的對象傳送給該線程并接收該線程的對象。V exchange(V v, long timeout, TimeUnit unit)等待另一個線程到達此交換點除非當前線程被中斷或超出了指定的等待時間然后將給定的對象傳送給該線程并接收該線程的對象。ExchangerInteger exchanger new Exchanger(); ExecutorService executor Executors.newCachedThreadPool(); Runnable run () -{ try { int num new Random().nextInt(10); System.out.println(Thread.currentThread().getName()開始交換數(shù)據(jù):num); num exchanger.exchange(num);//交換數(shù)據(jù)并得到交換后的數(shù)據(jù) System.out.println(Thread.currentThread().getName()交換數(shù)據(jù)結(jié)束后的數(shù)據(jù)num); } catch (InterruptedException e) { e.printStackTrace(); } }; executor.execute(run); executor.execute(run); executor.shutdown(); 結(jié)果 pool-1-thread-2開始交換數(shù)據(jù):9 pool-1-thread-1開始交換數(shù)據(jù):8 pool-1-thread-2交換數(shù)據(jù)結(jié)束后的數(shù)據(jù)8 pool-1-thread-1交換數(shù)據(jù)結(jié)束后的數(shù)據(jù)93.exchange源碼解析 參考下面的博客寫的很詳細https://blog.csdn.net/qq_31865983/article/details/105620881https://www.xuebuyuan.com/2736097.html