源碼倉(cāng)庫(kù)Gitee · GitHub一.Semaphore1.概念一個(gè)計(jì)數(shù)信號(hào)量。在概念上信號(hào)量維持一組許可證。如果有必要每個(gè)acquire()都會(huì)阻塞直到許可證可用然后才能使用它。每個(gè)release()添加許可證潛在地釋放阻塞獲取方。但是沒有使用實(shí)際的許可證對(duì)象;Semaphore只保留可用數(shù)量的計(jì)數(shù)并相應(yīng)地執(zhí)行。即一個(gè)Semaphore維護(hù)了一組permits【許可證】。每次調(diào)用acquire()方法都會(huì)阻塞直到獲取到許可證。每次調(diào)用release()方法都會(huì)添加一個(gè)許可證也就是釋放一個(gè)被阻塞的獲取者。但是實(shí)際上并不存在這個(gè)許可證Semaphore僅僅是記錄可用資源的數(shù)量并且做出對(duì)應(yīng)的行為有資源就獲取沒有資源就阻塞。信號(hào)量通常用于限制線程數(shù)而不是訪問某些物理或邏輯資源。線程池控制的是線程數(shù)量而信號(hào)量控制的是并發(fā)數(shù)量雖然說看起來一樣但兩者還是有區(qū)別的。信號(hào)量類似于鎖機(jī)制信號(hào)量的調(diào)用當(dāng)達(dá)到數(shù)量后線程還是存在的只是被掛起了而已。而線程池同時(shí)執(zhí)行的線程數(shù)量是固定的超過了數(shù)量的只能等待。在獲得項(xiàng)目之前每個(gè)線程必須從信號(hào)量獲取許可證以確保某個(gè)項(xiàng)目可用。當(dāng)線程完成該項(xiàng)目后它將返回到池中并將許可證返回到信號(hào)量允許另一個(gè)線程獲取該項(xiàng)目。請(qǐng)注意當(dāng)調(diào)用acquire()時(shí)不會(huì)保持同步鎖定因?yàn)檫@將阻止某個(gè)項(xiàng)目返回到池中。信號(hào)量封裝了限制對(duì)池的訪問所需的同步與保持池本身一致性所需的任何同步分開?！炯磳⑾拗茖?duì)池的訪問和對(duì)池中數(shù)據(jù)的操作所需要的鎖分開】。信號(hào)量被初始化為一個(gè)并且被使用使得它只有至多一個(gè)允許可用可以用作互斥鎖。這通常被稱為二進(jìn)制信號(hào)量因?yàn)樗挥袃蓚€(gè)狀態(tài)一個(gè)許可證可用或零個(gè)許可證可用。當(dāng)以這種方式使用時(shí)二進(jìn)制信號(hào)量具有屬性與許多Lock實(shí)現(xiàn)不同“鎖”可以由除所有者之外的線程釋放因?yàn)樾盘?hào)量沒有所有權(quán)概念。這在某些專門的上下文中是有用的例如死鎖恢復(fù)。Semaphore(int permits) 創(chuàng)建一個(gè) Semaphore與給定數(shù)量的許可證和非公平公平設(shè)置。 Semaphore(int permits, boolean fair) 創(chuàng)建一個(gè) Semaphore與給定數(shù)量的許可證和給定的公平設(shè)置。此類的構(gòu)造函數(shù)可選擇接受公平參數(shù)。當(dāng)設(shè)置為false時(shí)此類不會(huì)保證線程獲取許可的順序。特別是闖入是允許的也就是說一個(gè)線程調(diào)用acquire()可以提前已經(jīng)等待線程分配的許可證-在等待線程隊(duì)列的頭部邏輯新的線程將自己【新線程將自己放在等待線程隊(duì)列的最前面】。當(dāng)公平設(shè)置為真時(shí)信號(hào)量保證調(diào)用acquire方法的線程被選擇以按照它們調(diào)用這些方法的順序獲得許可先進(jìn)先出; FIFO【FIFO的順序是指是依據(jù)到達(dá)方法內(nèi)部的執(zhí)行點(diǎn)的時(shí)間并不是方法執(zhí)行的時(shí)間?！?。請(qǐng)注意FIFO排序必須適用于這些方法中的特定內(nèi)部執(zhí)行點(diǎn)。因此一個(gè)線程可以在另一個(gè)線程之前調(diào)用acquire但是在另一個(gè)線程之后到達(dá)排序點(diǎn)并且類似地從方法返回。另請(qǐng)注意未定義的tryAcquire方法不符合公平性設(shè)置但將采取任何可用的許可證?！静欢〞r(shí)的tryAcquire()方法會(huì)任意選取可用的許可證?！俊痉枪芥i可以插隊(duì)獲取運(yùn)行公平鎖按照線程順序執(zhí)行?！客ǔＳ糜诳刂瀑Y源訪問的信號(hào)量應(yīng)該被公平地初始化以確保線程沒有被訪問資源【確保沒有線程因?yàn)殚L(zhǎng)時(shí)間獲取不到許可證而餓死】。當(dāng)使用信號(hào)量進(jìn)行其他類型的同步控制時(shí)非正常排序的吞吐量?jī)?yōu)勢(shì)往往超過公平性。2.用法線程可以通過acquire()方法獲取到一個(gè)許可然后對(duì)共享資源進(jìn)行操作如果許可集已分配完了那么線程將進(jìn)入等待狀態(tài)直到其他線程釋放許可才有機(jī)會(huì)再獲取許可線程釋放一個(gè)許可通過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() 進(jìn)入當(dāng)前已有 (3 - sp.availablePermits()) 個(gè)并發(fā)); try { Thread.sleep((long) (Math.random() * 10000)); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(線程 Thread.currentThread().getName() 即將離開); sp.release(); //下面代碼有時(shí)候執(zhí)行不準(zhǔn)確因?yàn)槠錄]有和上面的代碼合成原子單元 System.out.println(線程 Thread.currentThread().getName() 已離開當(dāng)前已有 (3 - sp.availablePermits()) 個(gè)并發(fā)); }; service.execute(runnable); } } 結(jié)果 線程pool-1-thread-1進(jìn)入當(dāng)前已有1個(gè)并發(fā) 線程pool-1-thread-2進(jìn)入當(dāng)前已有2個(gè)并發(fā) 線程pool-1-thread-3進(jìn)入當(dāng)前已有3個(gè)并發(fā) 線程pool-1-thread-3即將離開 線程pool-1-thread-4進(jìn)入當(dāng)前已有3個(gè)并發(fā) 線程pool-1-thread-3已離開當(dāng)前已有3個(gè)并發(fā) 線程pool-1-thread-1即將離開 線程pool-1-thread-1已離開當(dāng)前已有2個(gè)并發(fā) 線程pool-1-thread-5進(jìn)入當(dāng)前已有3個(gè)并發(fā) 線程pool-1-thread-5即將離開 線程pool-1-thread-5已離開當(dāng)前已有2個(gè)并發(fā) 線程pool-1-thread-6進(jìn)入當(dāng)前已有3個(gè)并發(fā) 線程pool-1-thread-4即將離開 線程pool-1-thread-4已離開當(dāng)前已有2個(gè)并發(fā) 線程pool-1-thread-7進(jìn)入當(dāng)前已有3個(gè)并發(fā) 線程pool-1-thread-2即將離開 線程pool-1-thread-2已離開當(dāng)前已有2個(gè)并發(fā) 線程pool-1-thread-7即將離開 線程pool-1-thread-7已離開當(dāng)前已有1個(gè)并發(fā) 線程pool-1-thread-6即將離開 線程pool-1-thread-6已離開當(dāng)前已有0個(gè)并發(fā)3.acquire解析acquire() 從該信號(hào)量獲取許可證阻止直到可用或線程為 interrupted 。 void acquire(int permits) 從該信號(hào)量獲取給定數(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. * 以共享模式獲取如果中斷被中止。 * 實(shí)現(xiàn)首先檢查中斷狀態(tài)然后至少調(diào)用一次tryacquirered成功返回。 * 否則,線程排隊(duì)可能會(huì)重復(fù)阻塞和取消阻塞 * 調(diào)用tryacquiremred直到成功或線程被打斷了。 */ public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted())//中斷拋出異常 throw new InterruptedException(); if (tryAcquireShared(arg) 0)//獲取失敗加入同步隊(duì)列等待 doAcquireSharedInterruptibly(arg); } //由Semaphore的FairSync或NonfairSync實(shí)現(xiàn)共享模式下資源可以被多個(gè)線程通知占用tryAcquireShared返回int類型表示還有多少個(gè)資源可以同時(shí)被占用用于共享模式下傳播喚醒。 protected int tryAcquireShared(int arg) { throw new UnsupportedOperationException(); } //以共享中斷模式獲取 private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node addWaiter(Node.SHARED);//創(chuàng)建當(dāng)前線程的節(jié)點(diǎn)并且鎖的模型是共享鎖將其添加到AQS CLH隊(duì)列的末尾 boolean failed true; try { for (;;) {//自旋 final Node p node.predecessor();//獲得當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn) if (p head) {//是頭節(jié)點(diǎn)沒有等待節(jié)點(diǎn) int r tryAcquireShared(arg); if (r 0) {//獲取成功當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn)并傳播【傳播指的是同步狀態(tài)剩余的許可數(shù)值不為0通知后續(xù)結(jié)點(diǎn)繼續(xù)獲取同步狀態(tài)】 setHeadAndPropagate(node, r); p.next null; // help GC failed false; return; } } //前繼節(jié)點(diǎn)非head節(jié)點(diǎn)沒資源獲取將前繼節(jié)點(diǎn)狀態(tài)設(shè)置為SIGNAL通過park掛起node節(jié)點(diǎn)的線程 if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node);//結(jié)束該結(jié)點(diǎn)線程的請(qǐng)求 } }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é)點(diǎn)表示當(dāng)前線程前面有阻塞線程當(dāng)前線程獲取失敗先讓前節(jié)點(diǎn)線程獲取運(yùn)行【比非公平鎖獲取多了判斷前驅(qū)節(jié)點(diǎn)的操作】 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();//喚醒同步隊(duì)列中后繼結(jié)點(diǎn)的線程 return true; } return false; } protected boolean tryReleaseShared(int arg) {//由Semaphore的Sync實(shí)現(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. * 保證釋放動(dòng)作(向同步等待隊(duì)列尾部)傳遞即使沒有其他正在進(jìn)行的 * 請(qǐng)求或釋放動(dòng)作。如果頭節(jié)點(diǎn)的后繼節(jié)點(diǎn)需要喚醒那么執(zhí)行喚醒 * 動(dòng)作如果不需要將頭結(jié)點(diǎn)的等待狀態(tài)設(shè)置為PROPAGATE保證 * 喚醒傳遞。另外為了防止過程中有新節(jié)點(diǎn)進(jìn)入(隊(duì)列)這里必 * 需做循環(huán)所以和其他unparkSuccessor方法使用方式不一樣 * 的是如果(頭結(jié)點(diǎn))等待狀態(tài)設(shè)置失敗重新檢測(cè)。 */ for (;;) { Node h head; if (h ! null h ! tail) { int ws h.waitStatus;//頭節(jié)點(diǎn)狀態(tài) // 如果頭節(jié)點(diǎn)對(duì)應(yīng)的線程是SIGNAL狀態(tài)則意味著頭 //結(jié)點(diǎn)的后繼結(jié)點(diǎn)所對(duì)應(yīng)的線程需要被unpark-喚醒。 if (ws Node.SIGNAL) { // 修改頭結(jié)點(diǎn)對(duì)應(yīng)的線程狀態(tài)設(shè)置為0。失敗的話則繼續(xù)循環(huán)。 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases // 喚醒頭結(jié)點(diǎn)h的后繼結(jié)點(diǎn)所對(duì)應(yīng)的線程 unparkSuccessor(h); } else if (ws 0 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } // 如果頭結(jié)點(diǎn)發(fā)生變化則繼續(xù)循環(huán)。否則退出循環(huán)。 if (h head) // loop if head changed break; } } //喚醒傳入結(jié)點(diǎn)的后繼結(jié)點(diǎn)對(duì)應(yīng)的線程 private void unparkSuccessor(Node node) { int ws node.waitStatus; if (ws 0) compareAndSetWaitStatus(node, ws, 0); //拿到后繼結(jié)點(diǎn) 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();//獲取當(dāng)前同步狀態(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.概念線程可以在成對(duì)內(nèi)配對(duì)和交換元素的同步點(diǎn)。每個(gè)線程在輸入exchange方法時(shí)提供一些對(duì)象與合作者線程匹配并在返回時(shí)接收其合作伙伴的對(duì)象。交換器可以被視為一個(gè)的雙向形式SynchronousQueue。交換器在諸如遺傳算法和管道設(shè)計(jì)的應(yīng)用中可能是有用的。一個(gè)用于兩個(gè)工作線程之間交換數(shù)據(jù)的封裝工具類簡(jiǎn)單說就是一個(gè)線程在完成一定的事務(wù)后想與另一個(gè)線程交換數(shù)據(jù)則第一個(gè)先拿出數(shù)據(jù)的線程會(huì)一直等待第二個(gè)線程直到第二個(gè)線程拿著數(shù)據(jù)到來時(shí)才能彼此交換對(duì)應(yīng)數(shù)據(jù)。2.用法ExchangerV 泛型類型其中 V 表示可交換的數(shù)據(jù)類型V exchange(V v)等待另一個(gè)線程到達(dá)此交換點(diǎn)除非當(dāng)前線程被中斷然后將給定的對(duì)象傳送給該線程并接收該線程的對(duì)象。V exchange(V v, long timeout, TimeUnit unit)等待另一個(gè)線程到達(dá)此交換點(diǎn)除非當(dāng)前線程被中斷或超出了指定的等待時(shí)間然后將給定的對(duì)象傳送給該線程并接收該線程的對(duì)象。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源碼解析 參考下面的博客寫的很詳細(xì)https://blog.csdn.net/qq_31865983/article/details/105620881https://www.xuebuyuan.com/2736097.html