第一部分: synchronized
臨界資源
在多線程並發過程中,有可能會出現多個線程同時出現訪問同一個共享,可變資源的情況。這個資源可能是變量、文件、對象等。
共享:資源可以由多個線程同時訪問
可變:資源可以在其生命周期內修改
引發的問題:
由於線程的過程是不可控的,所以需要采用同步機制來對協同對象可變狀態的訪問。
Java 中,提供了兩種方式來實現同步互斥訪問:synchronized 和 Lock
同步器的本質就是加鎖
加鎖目的
序列化訪問臨界資源:即在任一時刻,只能有一個線程訪問臨界資源,也稱為 同步互斥訪問。
JAVA鎖體系
JAVA線程生命狀態
synchronized原理詳解
synchronized內置鎖是一種對象鎖,(鎖的是對象而非引用),作用粒度是對象,可以用來實現對臨界資源的同步互斥訪問,是可重入的。
加鎖的方式:
1. 同步實例方法 鎖是當前實例對象
2. 同步類方法 鎖是當前類對象
3. 同步代碼塊 鎖是括號里面的對象
synchronized不能跨方法保證原子性,那如何實現跨方法保證? --- Unsafe類monitorenter和monitorexit來實現。
synchronized底層原理
synchronized是基於底層JVM內置鎖實現,通過內部對象Monitor(監控器鎖)實現,基於進入和退出Monitor對象實現方法和代碼塊同步,監視器鎖的實現依賴底層操作系統的Mutex Lock(互斥鎖)實現,它是一個重量級鎖性能較低。
synchronized關鍵字被編譯成字節碼后會被翻譯成monitorenter 和 monitorexit 兩條指令分別在同步塊邏輯代碼的起始位置與結束位置。
每個同步對象都有一個自己的Monitor(監視器鎖),加鎖過程如下圖所示:
問題:synchronized加鎖加在對象上,對象是如何記錄鎖狀態的呢?
-- 鎖狀態是被記錄在每個對象的對象頭(Mark Word)中.
對象的內存布局
HotSpot虛擬機中,對象在內存中存儲的布局可以分為三塊區域:對象頭(Header)、實例數據(Instance Data)和對齊填充(Padding).
-- 對象頭:比如 hash碼,對象所屬的年代,對象鎖,鎖狀態標志,偏向鎖(線程)ID,偏向時間,數組長度(數組對象)等
-- 實例數據:即創建對象時,對象中成員變量,方法等
-- 對齊填充:對象的大小必須是8字節的整數倍
JVM內置鎖在1.5之后版本做了重大的優化
如鎖粗化(Lock Coarsening)、鎖消除(Lock Elimination)、輕量級鎖(Lightweight Locking)、偏向鎖(Biased Locking)、適應性自旋(Adaptive Spinning)等技術來減少鎖操作的開銷,,內置鎖的並發性能已經基本與Lock持平.
鎖粗化舉例:
===鎖粗化===》
鎖消除舉例:
逃逸分析
使用逃逸分析,編譯器可以對代碼做如下優化:
一、同步省略。如果一個對象被發現只能從一個線程被訪問到,那么對於這個對象的操作可以不考慮同步。
二、將堆分配轉化為棧分配。如果一個對象在子程序中被分配,要使指向該對象的指針永遠 不會逃逸,對象可能是棧分配的候選,而不是堆分配。
三、分離對象或標量替換。有的對象可能不需要作為一個連續的內存結構存在也可以被訪問 到,那么對象的部分(或全部)可以不存儲在內存,而是存儲在CPU寄存器中
在Java代碼運行時,通過JVM參數可指定是否開啟逃逸分析,
-XX:+DoEscapeAnalysis : 表示開啟逃逸分析
-XX:DoEscapeAnalysis : 表示關閉逃逸分析
從jdk 1.7開始已經默認開始逃逸分析,如需關閉,需要指定-XX:DoEscapeAnalysis
public class StackAllocTest { /** * 進行兩種測試 * 關閉逃逸分析,同時調大堆空間,避免堆內GC的發生,如果有GC信息將會被打印出來 * VM運行參數:-Xmx4G -Xms4G -XX:-DoEscapeAnalysis -XX:+PrintGCDetails -XX:+HeapDumpOnOutOfMemoryError * * 開啟逃逸分析 * VM運行參數:-Xmx4G -Xms4G -XX:+DoEscapeAnalysis -XX:+PrintGCDetails -XX:+HeapDumpOnOutOfMemoryError * * 執行main方法后 * jps 查看進程 * jmap -histo 進程ID */ public static void main(String[] args) { long start = System.currentTimeMillis(); for (int i = 0; i < 500000; i++) { alloc(); } long end = System.currentTimeMillis(); //查看執行時間 System.out.println("cost-time " + (end - start) + " ms"); try { Thread.sleep(100000); } catch (InterruptedException e1) { e1.printStackTrace(); } } private static TulingStudent alloc() { //Jit對編譯時會對代碼進行 逃逸分析 //並不是所有對象存放在堆區,有的一部分存在線程棧空間 TulingStudent student = new TulingStudent(); return student; } static class TulingStudent { private String name; private int age; } }
關閉逃逸分析:
關閉逃逸分析,同時調大堆空間,避免堆內GC的發生,如果有GC信息將會被打印出來
VM運行參數:-Xmx4G -Xms4G -XX:-DoEscapeAnalysis -XX:+PrintGCDetails -XX:+HeapDumpOnOutOfMemoryError
執行結果:
查看線程 jps
分析進程 jmap -histo + 進程號
打開逃逸分析:
開啟逃逸分析,同時調大堆空間,避免堆內GC的發生,如果有GC信息將會被打印出來
VM運行參數:-Xmx4G -Xms4G -XX:+DoEscapeAnalysis -XX:+PrintGCDetails -XX:+HeapDumpOnOutOfMemoryError
問題: 是不是實例對象都存放在堆區?
-- 不一定,如果實例對象沒有線程逃逸行為,實例對象存放在堆區;如果有線程逃逸行為,則有可能部分存在線程棧中。
如果實例對象存儲在堆區,實例對象內存存在堆區,實例的引用存在棧上,實例的元數據class存放在方法區或元空間。
輕量級鎖使用場景:
鎖的升級過程拆分
JVM鎖的膨脹升級過程場景一:
JVM鎖的膨脹升級過程場景二:
鎖的升級過程明細如下:
第二部分: LOCK&AQS -- 如 獨占鎖:ReentrantLock 讀寫鎖:ReentrantReadWriterLock
AbstractQueuedSynchronizer(AQS) 同步框架器
並發之父 Doug Lea
Java並發編程核心在於java.concurrent.util包而juc當中的大多數同步器實現都是圍繞着共同的基礎行為,比如等待隊列、條件隊列、獨占獲取、共享獲取等,而這個行為的抽象就是基於AbstractQueuedSynchronizer簡稱AQS,
AQS定義了一套多線程訪問共享資源的同步器框架,是一個依賴狀態(state)的同步器。
Java.concurrent.util當中同步器的實現如Lock,Latch,Barrier等,都是基於AQS框架實現
- 一般通過定義內部類Sync繼承AQS
- 將同步器所有調用都映射到Sync對應的方法
AQS內部維護屬性 volatile int state (32位)
- state表示資源的可用狀態
State三種訪問方式
- getState()、setState()、compareAndSetState()
AQS定義兩種資源共享方式
- Exclusive-獨占,只有一個線程能執行,如ReentrantLock
- Share-共享,多個線程可以同時執行,如Semaphore/CountDownLatch
AQS定義兩種隊列
- 同步等待隊列 CLH對列(雙向鏈表)
- 條件等待隊列
不同的自定義同步器爭用共享資源的方式也不同。自定義同步器在實現時只需要實現共享資源state的獲取與釋放方式即可,至於具體線程等待隊列的維護(如獲取資源失敗入隊/喚醒出隊等),AQS已經在頂層實現好了。
自定義同步器實現時主要實現以下幾種方法:
- isHeldExclusively():該線程是否正在獨占資源。只有用到condition才需要去實現它。
- tryAcquire(int):獨占方式。嘗試獲取資源,成功則返回true,失敗則返回false。
- tryRelease(int):獨占方式。嘗試釋放資源,成功則返回true,失敗則返回false。
- tryAcquireShared(int):共享方式。嘗試獲取資源。負數表示失敗;0表示成功,但沒有剩余可用資源;正數表示成功,且有剩余資源。
- tryReleaseShared(int):共享方式。嘗試釋放資源,如果釋放后允許喚醒后續等待結點返回true,否則返回false
AQS具備特性:
- 阻塞等待隊列
- 共享/獨占
- 公平/非公平
- 可重入
- 允許中斷
問題 : 阻塞等待隊列,是如何實現的? -- 通過魔術類 UnSafe.park() / UnSafe.unpark()
AbstractQueuedSynchronizer.java
問題 : 公平和非公平鎖如何實現?
公平鎖: private ReentrantLock lock = new ReentrantLock(true);
非公平鎖: private ReentrantLock lock = new ReentrantLock(false);
問題 : 共享鎖和獨占鎖如何區分的?
AbstractQueuedSynchronizer.java

問題 : AQS定義兩種資源共享方式? --- 共享 和 獨占
- Exclusive -- 獨占,只有一個線程能執行,如ReentrantLock
- share -- 共享,多個線程可以同時執行,如Semaphore / CountDownLatch
AQS定義的兩種對列
- 同步等待隊列 CLH(雙向鏈表)
- 條件等待隊列
同步等待隊列詳解
AQS當中的同步等待隊列也稱CLH隊列,CLH隊列是Craig、Landin、Hagersten三人發明的一種基於雙向鏈表數據結構的隊列,
是FIFO先入先出線程等待隊列,Java中的CLH隊列是原CLH隊列的一個變種,線程由原自旋機制改為阻塞機制。
條件等待隊列
Condition是一個多線程間協調通信的工具類,使得某個,或者某些線程一起等待某個條件(Condition),只有當該條件具備時,這些等待線程才會被喚醒,從而重新爭奪鎖。
常見各種鎖詳解:
可重入鎖舉例:
import java.util.concurrent.locks.ReentrantLock; /** * 可重入鎖 */ public class LockTemplete { private Integer counter = 0; /** * 可重入鎖,公平鎖 * 公平鎖, * 非公平鎖 * 需要保證多個線程使用的是同一個鎖 * * * synchronized是否可重入? * 虛擬機,在ObjectMonitor.hpp定義了synchronized他怎么取重入加鎖 ..。hotspot源碼 * counter +1 * 基於AQS 去實現加鎖與解鎖 */ private ReentrantLock lock = new ReentrantLock(true); /** * 需要保證多個線程使用的是同一個ReentrantLock對象 * @return */ public void modifyResources(String threadName){ System.out.println("通知《管理員》線程:--->"+threadName+"准備打水"); //默認創建的是獨占鎖,排它鎖;同一時刻讀或者寫只允許一個線程獲取鎖 lock.lock(); System.out.println("線程:--->"+threadName+"第一次加鎖"); counter++; System.out.println("線程:"+threadName+"打第"+counter+"桶水"); //重入該鎖,我還有一件事情要做,沒做完之前不能把鎖資源讓出去 lock.lock(); System.out.println("線程:--->"+threadName+"第二次加鎖"); counter++; System.out.println("線程:"+threadName+"打第"+counter+"桶水"); lock.unlock(); System.out.println("線程:"+threadName+"釋放一個鎖"); lock.unlock(); System.out.println("線程:"+threadName+"釋放一個鎖"); } public static void main(String[] args){ LockTemplete tp = new LockTemplete(); new Thread(()->{ String threadName = Thread.currentThread().getName(); tp.modifyResources(threadName); },"Thread A").start(); new Thread(()->{ String threadName = Thread.currentThread().getName(); tp.modifyResources(threadName); },"Thread B").start(); } }
源碼解析及中文解析:
ReentrantLock.java

package com.it.edu.aqs; /* * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms. */ /* * * * * * * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at * http://creativecommons.org/publicdomain/zero/1.0/ */ import java.util.Collection; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; /** * A reentrant mutual exclusion {@link Lock} with the same basic * behavior and semantics as the implicit monitor lock accessed using * {@code synchronized} methods and statements, but with extended * capabilities. * * <p>A {@code ReentrantLock} is <em>owned</em> by the thread last * successfully locking, but not yet unlocking it. A thread invoking * {@code lock} will return, successfully acquiring the lock, when * the lock is not owned by another thread. The method will return * immediately if the current thread already owns the lock. This can * be checked using methods {@link #isHeldByCurrentThread}, and {@link * #getHoldCount}. * * <p>The constructor for this class accepts an optional * <em>fairness</em> parameter. When set {@code true}, under * contention, locks favor granting access to the longest-waiting * thread. Otherwise this lock does not guarantee any particular * access order. Programs using fair locks accessed by many threads * may display lower overall throughput (i.e., are slower; often much * slower) than those using the default setting, but have smaller * variances in times to obtain locks and guarantee lack of * starvation. Note however, that fairness of locks does not guarantee * fairness of thread scheduling. Thus, one of many threads using a * fair lock may obtain it multiple times in succession while other * active threads are not progressing and not currently holding the * lock. * Also note that the untimed {@link #tryLock()} method does not * honor the fairness setting. It will succeed if the lock * is available even if other threads are waiting. * * <p>It is recommended practice to <em>always</em> immediately * follow a call to {@code lock} with a {@code try} block, most * typically in a before/after construction such as: * * <pre> {@code * class X { * private final ReentrantLock lock = new ReentrantLock(); * // ... * * public void m() { * lock.lock(); // block until condition holds * try { * // ... method body * } finally { * lock.unlock() * } * } * }}</pre> * * <p>In addition to implementing the {@link Lock} interface, this * class defines a number of {@code public} and {@code protected} * methods for inspecting the state of the lock. Some of these * methods are only useful for instrumentation and monitoring. * * <p>Serialization of this class behaves in the same way as built-in * locks: a deserialized lock is in the unlocked state, regardless of * its state when serialized. * * <p>This lock supports a maximum of 2147483647 recursive locks by * the same thread. Attempts to exceed this limit result in * {@link Error} throws from locking methods. * * @since 1.5 * @author Doug Lea */ public class ReentrantLock implements Lock, java.io.Serializable { private static final long serialVersionUID = 7373984872572414699L; /** * 內部調用AQS的動作,都基於該成員屬性實現 */ private final Sync sync; /** * ReentrantLock鎖同步操作的基礎類Sync,繼承自AQS框架. * 該類有兩個繼承類,1、NonfairSync 非公平鎖,2、FairSync公平鎖 */ abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = -5179523762034025860L; /** * 加鎖的具體行為由子類實現 */ abstract void lock(); /** * 嘗試獲取非公平鎖 */ final boolean nonfairTryAcquire(int acquires) { //acquires = 1 final Thread current = Thread.currentThread(); int c = getState(); /** * 不需要判斷同步隊列(CLH)中是否有排隊等待線程 * 判斷state狀態是否為0,不為0可以加鎖 */ if (c == 0) { //unsafe操作,cas修改state狀態 if (compareAndSetState(0, acquires)) { //獨占狀態鎖持有者指向當前線程 setExclusiveOwnerThread(current); return true; } } /** * state狀態不為0,判斷鎖持有者是否是當前線程, * 如果是當前線程持有 則state+1 */ else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } //加鎖失敗 return false; } /** * 釋放鎖 */ protected final boolean tryRelease(int releases) { int c = getState() - releases; if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; if (c == 0) { free = true; setExclusiveOwnerThread(null); } setState(c); return free; } /** * 判斷持有獨占鎖的線程是否是當前線程 */ protected final boolean isHeldExclusively() { return getExclusiveOwnerThread() == Thread.currentThread(); } //返回條件對象 final ConditionObject newCondition() { return new ConditionObject(); } final Thread getOwner() { return getState() == 0 ? null : getExclusiveOwnerThread(); } final int getHoldCount() { return isHeldExclusively() ? getState() : 0; } final boolean isLocked() { return getState() != 0; } /** * Reconstitutes the instance from a stream (that is, deserializes it). */ private void readObject(java.io.ObjectInputStream s) throws java.io.IOException, ClassNotFoundException { s.defaultReadObject(); setState(0); // reset to unlocked state } } /** * 非公平鎖 */ static final class NonfairSync extends Sync { private static final long serialVersionUID = 7316153563782823691L; /** * 加鎖行為 */ final void lock() { /** * 第一步:直接嘗試加鎖 * 與公平鎖實現的加鎖行為一個最大的區別在於,此處不會去判斷同步隊列(CLH隊列)中 * 是否有排隊等待加鎖的節點,上來直接加鎖(判斷state是否為0,CAS修改state為1) * ,並將獨占鎖持有者 exclusiveOwnerThread 屬性指向當前線程 * 如果當前有人占用鎖,再嘗試去加一次鎖 */ if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else //AQS定義的方法,加鎖 acquire(1); } /** * 父類AbstractQueuedSynchronizer.acquire()中調用本方法 */ protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } } /** * 公平鎖 */ static final class FairSync extends Sync { private static final long serialVersionUID = -3000897897090466540L; final void lock() { acquire(1); } /** * 重寫aqs中的方法邏輯 * 嘗試加鎖,被AQS的acquire()方法調用 */ protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { /** * 與非公平鎖中的區別,需要先判斷隊列當中是否有等待的節點 * 如果沒有則可以嘗試CAS獲取鎖 */ if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { //獨占線程指向當前線程 setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; } } /** * 默認構造函數,創建非公平鎖對象 */ public ReentrantLock() { sync = new NonfairSync(); } /** * 根據要求創建公平鎖或非公平鎖 */ public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); } /** * 加鎖 */ public void lock() { sync.lock(); } /** * 嘗試獲去取鎖,獲取失敗被阻塞,線程被中斷直接拋出異常 */ public void lockInterruptibly() throws InterruptedException { sync.acquireInterruptibly(1); } /** * 嘗試加鎖 */ public boolean tryLock() { return sync.nonfairTryAcquire(1); } /** * 指定等待時間內嘗試加鎖 */ public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireNanos(1, unit.toNanos(timeout)); } /** * 嘗試去釋放鎖 */ public void unlock() { sync.release(1); } /** * 返回條件對象 */ public Condition newCondition() { return sync.newCondition(); } /** * 返回當前線程持有的state狀態數量 */ public int getHoldCount() { return sync.getHoldCount(); } /** * 查詢當前線程是否持有鎖 */ public boolean isHeldByCurrentThread() { return sync.isHeldExclusively(); } /** * 狀態表示是否被Thread加鎖持有 */ public boolean isLocked() { return sync.isLocked(); } /** * 是否公平鎖?是返回true 否則返回 false */ public final boolean isFair() { return sync instanceof FairSync; } /** * Returns the thread that currently owns this lock, or * {@code null} if not owned. When this method is called by a * thread that is not the owner, the return value reflects a * best-effort approximation of current lock status. For example, * the owner may be momentarily {@code null} even if there are * threads trying to acquire the lock but have not yet done so. * This method is designed to facilitate construction of * subclasses that provide more extensive lock monitoring * facilities. * * @return the owner, or {@code null} if not owned */ protected Thread getOwner() { return sync.getOwner(); } /** * 判斷隊列當中是否有在等待獲取鎖的Thread節點 */ public final boolean hasQueuedThreads() { return sync.hasQueuedThreads(); } /** * 當前線程是否在同步隊列中等待 */ public final boolean hasQueuedThread(Thread thread) { return sync.isQueued(thread); } /** * Returns an estimate of the number of threads waiting to * acquire this lock. The value is only an estimate because the number of * threads may change dynamically while this method traverses * internal data structures. This method is designed for use in * monitoring of the system state, not for synchronization * control. * * @return the estimated number of threads waiting for this lock */ public final int getQueueLength() { return sync.getQueueLength(); } /** * 返回Thread集合,排隊中的所有節點Thread會被返回 */ protected Collection<Thread> getQueuedThreads() { return sync.getQueuedThreads(); } /** * 條件隊列當中是否有正在等待的節點 */ public boolean hasWaiters(Condition condition) { if (condition == null) throw new NullPointerException(); if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject)) throw new IllegalArgumentException("not owner"); return sync.hasWaiters((AbstractQueuedSynchronizer.ConditionObject)condition); } /** * Returns an estimate of the number of threads waiting on the * given condition associated with this lock. Note that because * timeouts and interrupts may occur at any time, the estimate * serves only as an upper bound on the actual number of waiters. * This method is designed for use in monitoring of the system * state, not for synchronization control. * * @param condition the condition * @return the estimated number of waiting threads * @throws IllegalMonitorStateException if this lock is not held * @throws IllegalArgumentException if the given condition is * not associated with this lock * @throws NullPointerException if the condition is null */ public int getWaitQueueLength(Condition condition) { if (condition == null) throw new NullPointerException(); if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject)) throw new IllegalArgumentException("not owner"); return sync.getWaitQueueLength((AbstractQueuedSynchronizer.ConditionObject)condition); } /** * Returns a collection containing those threads that may be * waiting on the given condition associated with this lock. * Because the actual set of threads may change dynamically while * constructing this result, the returned collection is only a * best-effort estimate. The elements of the returned collection * are in no particular order. This method is designed to * facilitate construction of subclasses that provide more * extensive condition monitoring facilities. * * @param condition the condition * @return the collection of threads * @throws IllegalMonitorStateException if this lock is not held * @throws IllegalArgumentException if the given condition is * not associated with this lock * @throws NullPointerException if the condition is null */ protected Collection<Thread> getWaitingThreads(Condition condition) { if (condition == null) throw new NullPointerException(); if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject)) throw new IllegalArgumentException("not owner"); return sync.getWaitingThreads((AbstractQueuedSynchronizer.ConditionObject)condition); } /** * Returns a string identifying this lock, as well as its lock state. * The state, in brackets, includes either the String {@code "Unlocked"} * or the String {@code "Locked by"} followed by the * {@linkplain Thread#getName name} of the owning thread. * * @return a string identifying this lock, as well as its lock state */ public String toString() { Thread o = sync.getOwner(); return super.toString() + ((o == null) ? "[Unlocked]" : "[Locked by thread " + o.getName() + "]"); } }
ReentrantReadWriteLock.java

/* * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms. * * * * * * * * * * * * * * * * * * * * */ /* * * * * * * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at * http://creativecommons.org/publicdomain/zero/1.0/ */ package com.it.edu.aqs; import java.util.concurrent.TimeUnit; import java.util.Collection; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; /** * An implementation of {@link ReadWriteLock} supporting similar * semantics to {@link ReentrantLock}. * <p>This class has the following properties: * * <ul> * <li><b>Acquisition order</b> * * <p>This class does not impose a reader or writer preference * ordering for lock access. However, it does support an optional * <em>fairness</em> policy. * * <dl> * <dt><b><i>Non-fair mode (default)</i></b> * <dd>When constructed as non-fair (the default), the order of entry * to the read and write lock is unspecified, subject to reentrancy * constraints. A nonfair lock that is continuously contended may * indefinitely postpone one or more reader or writer threads, but * will normally have higher throughput than a fair lock. * * <dt><b><i>Fair mode</i></b> * <dd>When constructed as fair, threads contend for entry using an * approximately arrival-order policy. When the currently held lock * is released, either the longest-waiting single writer thread will * be assigned the write lock, or if there is a group of reader threads * waiting longer than all waiting writer threads, that group will be * assigned the read lock. * * <p>A thread that tries to acquire a fair read lock (non-reentrantly) * will block if either the write lock is held, or there is a waiting * writer thread. The thread will not acquire the read lock until * after the oldest currently waiting writer thread has acquired and * released the write lock. Of course, if a waiting writer abandons * its wait, leaving one or more reader threads as the longest waiters * in the queue with the write lock free, then those readers will be * assigned the read lock. * * <p>A thread that tries to acquire a fair write lock (non-reentrantly) * will block unless both the read lock and write lock are free (which * implies there are no waiting threads). (Note that the non-blocking * {@link ReadLock#tryLock()} and {@link WriteLock#tryLock()} methods * do not honor this fair setting and will immediately acquire the lock * if it is possible, regardless of waiting threads.) * <p> * </dl> * * <li><b>Reentrancy</b> * * <p>This lock allows both readers and writers to reacquire read or * write locks in the style of a {@link ReentrantLock}. Non-reentrant * readers are not allowed until all write locks held by the writing * thread have been released. * * <p>Additionally, a writer can acquire the read lock, but not * vice-versa. Among other applications, reentrancy can be useful * when write locks are held during calls or callbacks to methods that * perform reads under read locks. If a reader tries to acquire the * write lock it will never succeed. * * <li><b>Lock downgrading</b> * <p>Reentrancy also allows downgrading from the write lock to a read lock, * by acquiring the write lock, then the read lock and then releasing the * write lock. However, upgrading from a read lock to the write lock is * <b>not</b> possible. * * <li><b>Interruption of lock acquisition</b> * <p>The read lock and write lock both support interruption during lock * acquisition. * * <li><b>{@link Condition} support</b> * <p>The write lock provides a {@link Condition} implementation that * behaves in the same way, with respect to the write lock, as the * {@link Condition} implementation provided by * {@link ReentrantLock#newCondition} does for {@link ReentrantLock}. * This {@link Condition} can, of course, only be used with the write lock. * * <p>The read lock does not support a {@link Condition} and * {@code readLock().newCondition()} throws * {@code UnsupportedOperationException}. * * <li><b>Instrumentation</b> * <p>This class supports methods to determine whether locks * are held or contended. These methods are designed for monitoring * system state, not for synchronization control. * </ul> * * <p>Serialization of this class behaves in the same way as built-in * locks: a deserialized lock is in the unlocked state, regardless of * its state when serialized. * * <p><b>Sample usages</b>. Here is a code sketch showing how to perform * lock downgrading after updating a cache (exception handling is * particularly tricky when handling multiple locks in a non-nested * fashion): * * <pre> {@code * class CachedData { * Object data; * volatile boolean cacheValid; * final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(); * * void processCachedData() { * rwl.readLock().lock(); * if (!cacheValid) { * // Must release read lock before acquiring write lock * rwl.readLock().unlock(); * rwl.writeLock().lock(); * try { * // Recheck state because another thread might have * // acquired write lock and changed state before we did. * if (!cacheValid) { * data = ... * cacheValid = true; * } * // Downgrade by acquiring read lock before releasing write lock * rwl.readLock().lock(); * } finally { * rwl.writeLock().unlock(); // Unlock write, still hold read * } * } * * try { * use(data); * } finally { * rwl.readLock().unlock(); * } * } * }}</pre> * * ReentrantReadWriteLocks can be used to improve concurrency in some * uses of some kinds of Collections. This is typically worthwhile * only when the collections are expected to be large, accessed by * more reader threads than writer threads, and entail operations with * overhead that outweighs synchronization overhead. For example, here * is a class using a TreeMap that is expected to be large and * concurrently accessed. * * <pre> {@code * class RWDictionary { * private final Map<String, Data> m = new TreeMap<String, Data>(); * private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(); * private final Lock r = rwl.readLock(); * private final Lock w = rwl.writeLock(); * * public Data get(String key) { * r.lock(); * try { return m.get(key); } * finally { r.unlock(); } * } * public String[] allKeys() { * r.lock(); * try { return m.keySet().toArray(); } * finally { r.unlock(); } * } * public Data put(String key, Data value) { * w.lock(); * try { return m.put(key, value); } * finally { w.unlock(); } * } * public void clear() { * w.lock(); * try { m.clear(); } * finally { w.unlock(); } * } * }}</pre> * * <h3>Implementation Notes</h3> * * <p>This lock supports a maximum of 65535 recursive write locks * and 65535 read locks. Attempts to exceed these limits result in * {@link Error} throws from locking methods. * * @since 1.5 * @author Doug Lea */ public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { private static final long serialVersionUID = -6992448646407690164L; /** Inner class providing readlock */ private final ReentrantReadWriteLock.ReadLock readerLock; /** Inner class providing writelock */ private final ReentrantReadWriteLock.WriteLock writerLock; /** Performs all synchronization mechanics */ final Sync sync; /** * Creates a new {@code ReentrantReadWriteLock} with * default (nonfair) ordering properties. */ public ReentrantReadWriteLock() { this(false); } /** * Creates a new {@code ReentrantReadWriteLock} with * the given fairness policy. * * @param fair {@code true} if this lock should use a fair ordering policy */ public ReentrantReadWriteLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); readerLock = new ReadLock(this); writerLock = new WriteLock(this); } public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; } public ReentrantReadWriteLock.ReadLock readLock() { return readerLock; } /** * Synchronization implementation for ReentrantReadWriteLock. * Subclassed into fair and nonfair versions. */ abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 6317671515068378041L; /* * Read vs write count extraction constants and functions. * Lock state is logically divided into two unsigned shorts: * The lower one representing the exclusive (writer) lock hold count, * and the upper the shared (reader) hold count. */ static final int SHARED_SHIFT = 16; static final int SHARED_UNIT = (1 << SHARED_SHIFT); static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1; static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; /** Returns the number of shared holds represented in count */ static int sharedCount(int c) { return c >>> SHARED_SHIFT; } /** Returns the number of exclusive holds represented in count */ static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; } /** * A counter for per-thread read hold counts. * Maintained as a ThreadLocal; cached in cachedHoldCounter */ static final class HoldCounter { int count = 0; // Use id, not reference, to avoid garbage retention final long tid = getThreadId(Thread.currentThread()); } /** * ThreadLocal subclass. Easiest to explicitly define for sake * of deserialization mechanics. */ static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> { public HoldCounter initialValue() { return new HoldCounter(); } } /** * The number of reentrant read locks held by current thread. * Initialized only in constructor and readObject. * Removed whenever a thread's read hold count drops to 0. */ private transient ThreadLocalHoldCounter readHolds; /** * The hold count of the last thread to successfully acquire * readLock. This saves ThreadLocal lookup in the common case * where the next thread to release is the last one to * acquire. This is non-volatile since it is just used * as a heuristic, and would be great for threads to cache. * * <p>Can outlive the Thread for which it is caching the read * hold count, but avoids garbage retention by not retaining a * reference to the Thread. * * <p>Accessed via a benign data race; relies on the memory * model's final field and out-of-thin-air guarantees. */ private transient HoldCounter cachedHoldCounter; /** * firstReader is the first thread to have acquired the read lock. * firstReaderHoldCount is firstReader's hold count. * * <p>More precisely, firstReader is the unique thread that last * changed the shared count from 0 to 1, and has not released the * read lock since then; null if there is no such thread. * * <p>Cannot cause garbage retention unless the thread terminated * without relinquishing its read locks, since tryReleaseShared * sets it to null. * * <p>Accessed via a benign data race; relies on the memory * model's out-of-thin-air guarantees for references. * * <p>This allows tracking of read holds for uncontended read * locks to be very cheap. */ private transient Thread firstReader = null; private transient int firstReaderHoldCount; Sync() { readHolds = new ThreadLocalHoldCounter(); setState(getState()); // ensures visibility of readHolds } /* * Acquires and releases use the same code for fair and * nonfair locks, but differ in whether/how they allow barging * when queues are non-empty. */ /** * Returns true if the current thread, when trying to acquire * the read lock, and otherwise eligible to do so, should block * because of policy for overtaking other waiting threads. */ abstract boolean readerShouldBlock(); /** * Returns true if the current thread, when trying to acquire * the write lock, and otherwise eligible to do so, should block * because of policy for overtaking other waiting threads. */ abstract boolean writerShouldBlock(); /* * Note that tryRelease and tryAcquire can be called by * Conditions. So it is possible that their arguments contain * both read and write holds that are all released during a * condition wait and re-established in tryAcquire. */ protected final boolean tryRelease(int releases) { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); int nextc = getState() - releases; boolean free = exclusiveCount(nextc) == 0; if (free) setExclusiveOwnerThread(null); setState(nextc); return free; } protected final boolean tryAcquire(int acquires) { /* * Walkthrough: * 1. If read count nonzero or write count nonzero * and owner is a different thread, fail. * 2. If count would saturate, fail. (This can only * happen if count is already nonzero.) * 3. Otherwise, this thread is eligible for lock if * it is either a reentrant acquire or * queue policy allows it. If so, update state * and set owner. */ Thread current = Thread.currentThread(); int c = getState(); int w = exclusiveCount(c); if (c != 0) { // (Note: if c != 0 and w == 0 then shared count != 0) if (w == 0 || current != getExclusiveOwnerThread()) return false; if (w + exclusiveCount(acquires) > MAX_COUNT) throw new Error("Maximum lock count exceeded"); // Reentrant acquire setState(c + acquires); return true; } if (writerShouldBlock() || !compareAndSetState(c, c + acquires)) return false; setExclusiveOwnerThread(current); return true; } /** * 重寫aqs當中業務邏輯 * @param unused * @return */ protected final boolean tryReleaseShared(int unused) { Thread current = Thread.currentThread(); if (firstReader == current) { // assert firstReaderHoldCount > 0; if (firstReaderHoldCount == 1) firstReader = null; else firstReaderHoldCount--; } else { HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) rh = readHolds.get(); int count = rh.count; if (count <= 1) { readHolds.remove(); if (count <= 0) throw unmatchedUnlockException(); } --rh.count; } for (;;) { int c = getState(); int nextc = c - SHARED_UNIT; if (compareAndSetState(c, nextc)) // Releasing the read lock has no effect on readers, // but it may allow waiting writers to proceed if // both read and write locks are now free. return nextc == 0; } } private IllegalMonitorStateException unmatchedUnlockException() { return new IllegalMonitorStateException( "attempt to unlock read lock, not locked by current thread"); } /** * 嘗試加共享鎖 */ protected final int tryAcquireShared(int unused) { Thread current = Thread.currentThread(); int c = getState(); /** * 如果state狀態當前被加獨占寫鎖,則返回加共享鎖失敗 */ if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1; int r = sharedCount(c); //讀鎖數量 if (!readerShouldBlock() && //讀鎖是否應該阻塞 r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) { if (r == 0) { //第一次加讀鎖 firstReader = current; firstReaderHoldCount = 1;//第一個讀鎖Thread持有鎖數量 } else if (firstReader == current) { firstReaderHoldCount++; //當前Thread是加鎖,讀鎖數量加1 } else { HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) //每一個加鎖的thread對應一個HoldCounter cachedHoldCounter = rh = readHolds.get();//從threadLocal中取出變量 else if (rh.count == 0) readHolds.set(rh); rh.count++;// } return 1; } return fullTryAcquireShared(current); } /** * Full version of acquire for reads, that handles CAS misses * and reentrant reads not dealt with in tryAcquireShared. */ final int fullTryAcquireShared(Thread current) { /* * This code is in part redundant with that in * tryAcquireShared but is simpler overall by not * complicating tryAcquireShared with interactions between * retries and lazily reading hold counts. */ HoldCounter rh = null; for (;;) { int c = getState(); if (exclusiveCount(c) != 0) { if (getExclusiveOwnerThread() != current) return -1; // else we hold the exclusive lock; blocking here // would cause deadlock. } else if (readerShouldBlock()) { // Make sure we're not acquiring read lock reentrantly if (firstReader == current) { // assert firstReaderHoldCount > 0; } else { if (rh == null) { rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) { rh = readHolds.get(); if (rh.count == 0) readHolds.remove(); } } if (rh.count == 0) return -1; } } if (sharedCount(c) == MAX_COUNT) throw new Error("Maximum lock count exceeded"); if (compareAndSetState(c, c + SHARED_UNIT)) { if (sharedCount(c) == 0) { firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) { firstReaderHoldCount++; } else { if (rh == null) rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) rh = readHolds.get(); else if (rh.count == 0) readHolds.set(rh); rh.count++; cachedHoldCounter = rh; // cache for release } return 1; } } } /** * Performs tryLock for write, enabling barging in both modes. * This is identical in effect to tryAcquire except for lack * of calls to writerShouldBlock. */ final boolean tryWriteLock() { Thread current = Thread.currentThread(); int c = getState(); if (c != 0) { int w = exclusiveCount(c); if (w == 0 || current != getExclusiveOwnerThread()) return false; if (w == MAX_COUNT) throw new Error("Maximum lock count exceeded"); } if (!compareAndSetState(c, c + 1)) return false; setExclusiveOwnerThread(current); return true; } /** * Performs tryLock for read, enabling barging in both modes. * This is identical in effect to tryAcquireShared except for * lack of calls to readerShouldBlock. */ final boolean tryReadLock() { Thread current = Thread.currentThread(); for (;;) { int c = getState(); if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return false; int r = sharedCount(c); if (r == MAX_COUNT) throw new Error("Maximum lock count exceeded"); if (compareAndSetState(c, c + SHARED_UNIT)) { if (r == 0) { firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) { firstReaderHoldCount++; } else { HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) cachedHoldCounter = rh = readHolds.get(); else if (rh.count == 0) readHolds.set(rh); rh.count++; } return true; } } } protected final boolean isHeldExclusively() { // While we must in general read state before owner, // we don't need to do so to check if current thread is owner return getExclusiveOwnerThread() == Thread.currentThread(); } // Methods relayed to outer class final ConditionObject newCondition() { return new ConditionObject(); } final Thread getOwner() { // Must read state before owner to ensure memory consistency return ((exclusiveCount(getState()) == 0) ? null : getExclusiveOwnerThread()); } final int getReadLockCount() { return sharedCount(getState()); } final boolean isWriteLocked() { return exclusiveCount(getState()) != 0; } final int getWriteHoldCount() { return isHeldExclusively() ? exclusiveCount(getState()) : 0; } final int getReadHoldCount() { if (getReadLockCount() == 0) return 0; Thread current = Thread.currentThread(); if (firstReader == current) return firstReaderHoldCount; HoldCounter rh = cachedHoldCounter; if (rh != null && rh.tid == getThreadId(current)) return rh.count; int count = readHolds.get().count; if (count == 0) readHolds.remove(); return count; } /** * Reconstitutes the instance from a stream (that is, deserializes it). */ private void readObject(java.io.ObjectInputStream s) throws java.io.IOException, ClassNotFoundException { s.defaultReadObject(); readHolds = new ThreadLocalHoldCounter(); setState(0); // reset to unlocked state } final int getCount() { return getState(); } } /** * Nonfair version of Sync */ static final class NonfairSync extends Sync { private static final long serialVersionUID = -8159625535654395037L; final boolean writerShouldBlock() { return false; // writers can always barge } final boolean readerShouldBlock() { /* As a heuristic to avoid indefinite writer starvation, * block if the thread that momentarily appears to be head * of queue, if one exists, is a waiting writer. This is * only a probabilistic effect since a new reader will not * block if there is a waiting writer behind other enabled * readers that have not yet drained from the queue. */ return apparentlyFirstQueuedIsExclusive(); } } /** * Fair version of Sync */ static final class FairSync extends Sync { private static final long serialVersionUID = -2274990926593161451L; final boolean writerShouldBlock() { return hasQueuedPredecessors(); } final boolean readerShouldBlock() { return hasQueuedPredecessors(); } } /** * The lock returned by method {@link ReentrantReadWriteLock#readLock}. */ public static class ReadLock implements Lock, java.io.Serializable { private static final long serialVersionUID = -5992448646407690164L; private final Sync sync; /** * Constructor for use by subclasses * * @param lock the outer lock object * @throws NullPointerException if the lock is null */ protected ReadLock(ReentrantReadWriteLock lock) { sync = lock.sync; } /** * Acquires the read lock. * * <p>Acquires the read lock if the write lock is not held by * another thread and returns immediately. * * <p>If the write lock is held by another thread then * the current thread becomes disabled for thread scheduling * purposes and lies dormant until the read lock has been acquired. */ public void lock() { sync.acquireShared(1); } /** * Acquires the read lock unless the current thread is * {@linkplain Thread#interrupt interrupted}. * * <p>Acquires the read lock if the write lock is not held * by another thread and returns immediately. * * <p>If the write lock is held by another thread then the * current thread becomes disabled for thread scheduling * purposes and lies dormant until one of two things happens: * * <ul> * * <li>The read lock is acquired by the current thread; or * * <li>Some other thread {@linkplain Thread#interrupt interrupts} * the current thread. * * </ul> * * <p>If the current thread: * * <ul> * * <li>has its interrupted status set on entry to this method; or * * <li>is {@linkplain Thread#interrupt interrupted} while * acquiring the read lock, * * </ul> * * then {@link InterruptedException} is thrown and the current * thread's interrupted status is cleared. * * <p>In this implementation, as this method is an explicit * interruption point, preference is given to responding to * the interrupt over normal or reentrant acquisition of the * lock. * * @throws InterruptedException if the current thread is interrupted */ public void lockInterruptibly() throws InterruptedException { sync.acquireSharedInterruptibly(1); } /** * Acquires the read lock only if the write lock is not held by * another thread at the time of invocation. * * <p>Acquires the read lock if the write lock is not held by * another thread and returns immediately with the value * {@code true}. Even when this lock has been set to use a * fair ordering policy, a call to {@code tryLock()} * <em>will</em> immediately acquire the read lock if it is * available, whether or not other threads are currently * waiting for the read lock. This "barging" behavior * can be useful in certain circumstances, even though it * breaks fairness. If you want to honor the fairness setting * for this lock, then use {@link #tryLock(long, TimeUnit) * tryLock(0, TimeUnit.SECONDS) } which is almost equivalent * (it also detects interruption). * * <p>If the write lock is held by another thread then * this method will return immediately with the value * {@code false}. * * @return {@code true} if the read lock was acquired */ public boolean tryLock() { return sync.tryReadLock(); } /** * Acquires the read lock if the write lock is not held by * another thread within the given waiting time and the * current thread has not been {@linkplain Thread#interrupt * interrupted}. * * <p>Acquires the read lock if the write lock is not held by * another thread and returns immediately with the value * {@code true}. If this lock has been set to use a fair * ordering policy then an available lock <em>will not</em> be * acquired if any other threads are waiting for the * lock. This is in contrast to the {@link #tryLock()} * method. If you want a timed {@code tryLock} that does * permit barging on a fair lock then combine the timed and * un-timed forms together: * * <pre> {@code * if (lock.tryLock() || * lock.tryLock(timeout, unit)) { * ... * }}</pre> * * <p>If the write lock is held by another thread then the * current thread becomes disabled for thread scheduling * purposes and lies dormant until one of three things happens: * * <ul> * * <li>The read lock is acquired by the current thread; or * * <li>Some other thread {@linkplain Thread#interrupt interrupts} * the current thread; or * * <li>The specified waiting time elapses. * * </ul> * * <p>If the read lock is acquired then the value {@code true} is * returned. * * <p>If the current thread: * * <ul> * * <li>has its interrupted status set on entry to this method; or * * <li>is {@linkplain Thread#interrupt interrupted} while * acquiring the read lock, * * </ul> then {@link InterruptedException} is thrown and the * current thread's interrupted status is cleared. * * <p>If the specified waiting time elapses then the value * {@code false} is returned. If the time is less than or * equal to zero, the method will not wait at all. * * <p>In this implementation, as this method is an explicit * interruption point, preference is given to responding to * the interrupt over normal or reentrant acquisition of the * lock, and over reporting the elapse of the waiting time. * * @param timeout the time to wait for the read lock * @param unit the time unit of the timeout argument * @return {@code true} if the read lock was acquired * @throws InterruptedException if the current thread is interrupted * @throws NullPointerException if the time unit is null */ public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } /** * Attempts to release this lock. * * <p>If the number of readers is now zero then the lock * is made available for write lock attempts. */ public void unlock() { sync.releaseShared(1); } /** * Throws {@code UnsupportedOperationException} because * {@code ReadLocks} do not support conditions. * * @throws UnsupportedOperationException always */ public Condition newCondition() { throw new UnsupportedOperationException(); } /** * Returns a string identifying this lock, as well as its lock state. * The state, in brackets, includes the String {@code "Read locks ="} * followed by the number of held read locks. * * @return a string identifying this lock, as well as its lock state */ public String toString() { int r = sync.getReadLockCount(); return super.toString() + "[Read locks = " + r + "]"; } } /** * The lock returned by method {@link ReentrantReadWriteLock#writeLock}. */ public static class WriteLock implements Lock, java.io.Serializable { private static final long serialVersionUID = -4992448646407690164L; private final Sync sync; /** * Constructor for use by subclasses * * @param lock the outer lock object * @throws NullPointerException if the lock is null */ protected WriteLock(ReentrantReadWriteLock lock) { sync = lock.sync; } /** * Acquires the write lock. * * <p>Acquires the write lock if neither the read nor write lock * are held by another thread * and returns immediately, setting the write lock hold count to * one. * * <p>If the current thread already holds the write lock then the * hold count is incremented by one and the method returns * immediately. * * <p>If the lock is held by another thread then the current * thread becomes disabled for thread scheduling purposes and * lies dormant until the write lock has been acquired, at which * time the write lock hold count is set to one. */ public void lock() { sync.acquire(1); } /** * Acquires the write lock unless the current thread is * {@linkplain Thread#interrupt interrupted}. * * <p>Acquires the write lock if neither the read nor write lock * are held by another thread * and returns immediately, setting the write lock hold count to * one. * * <p>If the current thread already holds this lock then the * hold count is incremented by one and the method returns * immediately. * * <p>If the lock is held by another thread then the current * thread becomes disabled for thread scheduling purposes and * lies dormant until one of two things happens: * * <ul> * * <li>The write lock is acquired by the current thread; or * * <li>Some other thread {@linkplain Thread#interrupt interrupts} * the current thread. * * </ul> * * <p>If the write lock is acquired by the current thread then the * lock hold count is set to one. * * <p>If the current thread: * * <ul> * * <li>has its interrupted status set on entry to this method; * or * * <li>is {@linkplain Thread#interrupt interrupted} while * acquiring the write lock, * * </ul> * * then {@link InterruptedException} is thrown and the current * thread's interrupted status is cleared. * * <p>In this implementation, as this method is an explicit * interruption point, preference is given to responding to * the interrupt over normal or reentrant acquisition of the * lock. * * @throws InterruptedException if the current thread is interrupted */ public void lockInterruptibly() throws InterruptedException { sync.acquireInterruptibly(1); } /** * Acquires the write lock only if it is not held by another thread * at the time of invocation. * * <p>Acquires the write lock if neither the read nor write lock * are held by another thread * and returns immediately with the value {@code true}, * setting the write lock hold count to one. Even when this lock has * been set to use a fair ordering policy, a call to * {@code tryLock()} <em>will</em> immediately acquire the * lock if it is available, whether or not other threads are * currently waiting for the write lock. This "barging" * behavior can be useful in certain circumstances, even * though it breaks fairness. If you want to honor the * fairness setting for this lock, then use {@link * #tryLock(long, TimeUnit) tryLock(0, TimeUnit.SECONDS) } * which is almost equivalent (it also detects interruption). * * <p>If the current thread already holds this lock then the * hold count is incremented by one and the method returns * {@code true}. * * <p>If the lock is held by another thread then this method * will return immediately with the value {@code false}. * * @return {@code true} if the lock was free and was acquired * by the current thread, or the write lock was already held * by the current thread; and {@code false} otherwise. */ public boolean tryLock( ) { return sync.tryWriteLock(); } /** * Acquires the write lock if it is not held by another thread * within the given waiting time and the current thread has * not been {@linkplain Thread#interrupt interrupted}. * * <p>Acquires the write lock if neither the read nor write lock * are held by another thread * and returns immediately with the value {@code true}, * setting the write lock hold count to one. If this lock has been * set to use a fair ordering policy then an available lock * <em>will not</em> be acquired if any other threads are * waiting for the write lock. This is in contrast to the {@link * #tryLock()} method. If you want a timed {@code tryLock} * that does permit barging on a fair lock then combine the * timed and un-timed forms together: * * <pre> {@code * if (lock.tryLock() || * lock.tryLock(timeout, unit)) { * ... * }}</pre> * * <p>If the current thread already holds this lock then the * hold count is incremented by one and the method returns * {@code true}. * * <p>If the lock is held by another thread then the current * thread becomes disabled for thread scheduling purposes and * lies dormant until one of three things happens: * * <ul> * * <li>The write lock is acquired by the current thread; or * * <li>Some other thread {@linkplain Thread#interrupt interrupts} * the current thread; or * * <li>The specified waiting time elapses * * </ul> * * <p>If the write lock is acquired then the value {@code true} is * returned and the write lock hold count is set to one. * * <p>If the current thread: * * <ul> * * <li>has its interrupted status set on entry to this method; * or * * <li>is {@linkplain Thread#interrupt interrupted} while * acquiring the write lock, * * </ul> * * then {@link InterruptedException} is thrown and the current * thread's interrupted status is cleared. * * <p>If the specified waiting time elapses then the value * {@code false} is returned. If the time is less than or * equal to zero, the method will not wait at all. * * <p>In this implementation, as this method is an explicit * interruption point, preference is given to responding to * the interrupt over normal or reentrant acquisition of the * lock, and over reporting the elapse of the waiting time. * * @param timeout the time to wait for the write lock * @param unit the time unit of the timeout argument * * @return {@code true} if the lock was free and was acquired * by the current thread, or the write lock was already held by the * current thread; and {@code false} if the waiting time * elapsed before the lock could be acquired. * * @throws InterruptedException if the current thread is interrupted * @throws NullPointerException if the time unit is null */ public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireNanos(1, unit.toNanos(timeout)); } /** * Attempts to release this lock. * * <p>If the current thread is the holder of this lock then * the hold count is decremented. If the hold count is now * zero then the lock is released. If the current thread is * not the holder of this lock then {@link * IllegalMonitorStateException} is thrown. * * @throws IllegalMonitorStateException if the current thread does not * hold this lock */ public void unlock() { sync.release(1); } /** * Returns a {@link Condition} instance for use with this * {@link Lock} instance. * <p>The returned {@link Condition} instance supports the same * usages as do the {@link Object} monitor methods ({@link * Object#wait() wait}, {@link Object#notify notify}, and {@link * Object#notifyAll notifyAll}) when used with the built-in * monitor lock. * * <ul> * * <li>If this write lock is not held when any {@link * Condition} method is called then an {@link * IllegalMonitorStateException} is thrown. (Read locks are * held independently of write locks, so are not checked or * affected. However it is essentially always an error to * invoke a condition waiting method when the current thread * has also acquired read locks, since other threads that * could unblock it will not be able to acquire the write * lock.) * * <li>When the condition {@linkplain Condition#await() waiting} * methods are called the write lock is released and, before * they return, the write lock is reacquired and the lock hold * count restored to what it was when the method was called. * * <li>If a thread is {@linkplain Thread#interrupt interrupted} while * waiting then the wait will terminate, an {@link * InterruptedException} will be thrown, and the thread's * interrupted status will be cleared. * * <li> Waiting threads are signalled in FIFO order. * * <li>The ordering of lock reacquisition for threads returning * from waiting methods is the same as for threads initially * acquiring the lock, which is in the default case not specified, * but for <em>fair</em> locks favors those threads that have been * waiting the longest. * * </ul> * * @return the Condition object */ public Condition newCondition() { return sync.newCondition(); } /** * Returns a string identifying this lock, as well as its lock * state. The state, in brackets includes either the String * {@code "Unlocked"} or the String {@code "Locked by"} * followed by the {@linkplain Thread#getName name} of the owning thread. * * @return a string identifying this lock, as well as its lock state */ public String toString() { Thread o = sync.getOwner(); return super.toString() + ((o == null) ? "[Unlocked]" : "[Locked by thread " + o.getName() + "]"); } /** * Queries if this write lock is held by the current thread. * Identical in effect to {@link * ReentrantReadWriteLock#isWriteLockedByCurrentThread}. * * @return {@code true} if the current thread holds this lock and * {@code false} otherwise * @since 1.6 */ public boolean isHeldByCurrentThread() { return sync.isHeldExclusively(); } /** * Queries the number of holds on this write lock by the current * thread. A thread has a hold on a lock for each lock action * that is not matched by an unlock action. Identical in effect * to {@link ReentrantReadWriteLock#getWriteHoldCount}. * * @return the number of holds on this lock by the current thread, * or zero if this lock is not held by the current thread * @since 1.6 */ public int getHoldCount() { return sync.getWriteHoldCount(); } } // Instrumentation and status /** * Returns {@code true} if this lock has fairness set true. * * @return {@code true} if this lock has fairness set true */ public final boolean isFair() { return sync instanceof FairSync; } /** * Returns the thread that currently owns the write lock, or * {@code null} if not owned. When this method is called by a * thread that is not the owner, the return value reflects a * best-effort approximation of current lock status. For example, * the owner may be momentarily {@code null} even if there are * threads trying to acquire the lock but have not yet done so. * This method is designed to facilitate construction of * subclasses that provide more extensive lock monitoring * facilities. * * @return the owner, or {@code null} if not owned */ protected Thread getOwner() { return sync.getOwner(); } /** * Queries the number of read locks held for this lock. This * method is designed for use in monitoring system state, not for * synchronization control. * @return the number of read locks held */ public int getReadLockCount() { return sync.getReadLockCount(); } /** * Queries if the write lock is held by any thread. This method is * designed for use in monitoring system state, not for * synchronization control. * * @return {@code true} if any thread holds the write lock and * {@code false} otherwise */ public boolean isWriteLocked() { return sync.isWriteLocked(); } /** * Queries if the write lock is held by the current thread. * * @return {@code true} if the current thread holds the write lock and * {@code false} otherwise */ public boolean isWriteLockedByCurrentThread() { return sync.isHeldExclusively(); } /** * 獲取當前線程可重入寫鎖的持有數量 */ public int getWriteHoldCount() { return sync.getWriteHoldCount(); } /** * 獲取當前線程可重入讀鎖的持有數量 */ public int getReadHoldCount() { return sync.getReadHoldCount(); } /** * 獲取所有在同步隊列中等待獲取寫鎖的thread數量 */ protected Collection<Thread> getQueuedWriterThreads() { return sync.getExclusiveQueuedThreads(); } /** * 獲取所有在同步隊列中等待獲取讀鎖的thread數量 */ protected Collection<Thread> getQueuedReaderThreads() { return sync.getSharedQueuedThreads(); } /** * 是否有在隊列當中等待獲取鎖的線程 */ public final boolean hasQueuedThreads() { return sync.hasQueuedThreads(); } /** * Queries whether the given thread is waiting to acquire either * the read or write lock. Note that because cancellations may * occur at any time, a {@code true} return does not guarantee * that this thread will ever acquire a lock. This method is * designed primarily for use in monitoring of the system state. * * @param thread the thread * @return {@code true} if the given thread is queued waiting for this lock * @throws NullPointerException if the thread is null */ public final boolean hasQueuedThread(Thread thread) { return sync.isQueued(thread); } /** * Returns an estimate of the number of threads waiting to acquire * either the read or write lock. The value is only an estimate * because the number of threads may change dynamically while this * method traverses internal data structures. This method is * designed for use in monitoring of the system state, not for * synchronization control. * * @return the estimated number of threads waiting for this lock */ public final int getQueueLength() { return sync.getQueueLength(); } /** * Returns a collection containing threads that may be waiting to * acquire either the read or write lock. Because the actual set * of threads may change dynamically while constructing this * result, the returned collection is only a best-effort estimate. * The elements of the returned collection are in no particular * order. This method is designed to facilitate construction of * subclasses that provide more extensive monitoring facilities. * * @return the collection of threads */ protected Collection<Thread> getQueuedThreads() { return sync.getQueuedThreads(); } /** * Queries whether any threads are waiting on the given condition * associated with the write lock. Note that because timeouts and * interrupts may occur at any time, a {@code true} return does * not guarantee that a future {@code signal} will awaken any * threads. This method is designed primarily for use in * monitoring of the system state. * * @param condition the condition * @return {@code true} if there are any waiting threads * @throws IllegalMonitorStateException if this lock is not held * @throws IllegalArgumentException if the given condition is * not associated with this lock * @throws NullPointerException if the condition is null */ public boolean hasWaiters(Condition condition) { if (condition == null) throw new NullPointerException(); if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject)) throw new IllegalArgumentException("not owner"); return sync.hasWaiters((AbstractQueuedSynchronizer.ConditionObject)condition); } /** * Returns an estimate of the number of threads waiting on the * given condition associated with the write lock. Note that because * timeouts and interrupts may occur at any time, the estimate * serves only as an upper bound on the actual number of waiters. * This method is designed for use in monitoring of the system * state, not for synchronization control. * * @param condition the condition * @return the estimated number of waiting threads * @throws IllegalMonitorStateException if this lock is not held * @throws IllegalArgumentException if the given condition is * not associated with this lock * @throws NullPointerException if the condition is null */ public int getWaitQueueLength(Condition condition) { if (condition == null) throw new NullPointerException(); if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject)) throw new IllegalArgumentException("not owner"); return sync.getWaitQueueLength((AbstractQueuedSynchronizer.ConditionObject)condition); } /** * Returns a collection containing those threads that may be * waiting on the given condition associated with the write lock. * Because the actual set of threads may change dynamically while * constructing this result, the returned collection is only a * best-effort estimate. The elements of the returned collection * are in no particular order. This method is designed to * facilitate construction of subclasses that provide more * extensive condition monitoring facilities. * * @param condition the condition * @return the collection of threads * @throws IllegalMonitorStateException if this lock is not held * @throws IllegalArgumentException if the given condition is * not associated with this lock * @throws NullPointerException if the condition is null */ protected Collection<Thread> getWaitingThreads(Condition condition) { if (condition == null) throw new NullPointerException(); if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject)) throw new IllegalArgumentException("not owner"); return sync.getWaitingThreads((AbstractQueuedSynchronizer.ConditionObject)condition); } /** * Returns a string identifying this lock, as well as its lock state. * The state, in brackets, includes the String {@code "Write locks ="} * followed by the number of reentrantly held write locks, and the * String {@code "Read locks ="} followed by the number of held * read locks. * * @return a string identifying this lock, as well as its lock state */ public String toString() { int c = sync.getCount(); int w = Sync.exclusiveCount(c); int r = Sync.sharedCount(c); return super.toString() + "[Write locks = " + w + ", Read locks = " + r + "]"; } /** * Returns the thread id for the given thread. We must access * this directly rather than via method Thread.getId() because * getId() is not final, and has been known to be overridden in * ways that do not preserve unique mappings. */ static final long getThreadId(Thread thread) { return UNSAFE.getLongVolatile(thread, TID_OFFSET); } // Unsafe mechanics private static final sun.misc.Unsafe UNSAFE; private static final long TID_OFFSET; static { try { UNSAFE = sun.misc.Unsafe.getUnsafe(); Class<?> tk = Thread.class; TID_OFFSET = UNSAFE.objectFieldOffset (tk.getDeclaredField("tid")); } catch (Exception e) { throw new Error(e); } } }
AbstractQueuedSynchronizer.java

import sun.misc.Unsafe; import java.util.ArrayList; import java.util.Collection; import java.util.Date; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.LockSupport; /** * description:AQS同步器框架源碼 */ public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { private static final long serialVersionUID = 7373984972572414691L; /** * Creates a new {@code AbstractQueuedSynchronizer} instance * with initial synchronization state of zero. */ protected AbstractQueuedSynchronizer() { } /** * Wait queue node class. * * <p>The wait queue is a variant of a "CLH" (Craig, Landin, and * Hagersten) lock queue. CLH locks are normally used for * spinlocks. We instead use them for blocking synchronizers, but * use the same basic tactic of holding some of the control * information about a thread in the predecessor of its node. A * "status" field in each node keeps track of whether a thread * should block. A node is signalled when its predecessor * releases. Each node of the queue otherwise serves as a * specific-notification-style monitor holding a single waiting * thread. The status field does NOT control whether threads are * granted locks etc though. A thread may try to acquire if it is * first in the queue. But being first does not guarantee success; * it only gives the right to contend. So the currently released * contender thread may need to rewait. * * <p>To enqueue into a CLH lock, you atomically splice it in as new * tail. To dequeue, you just set the head field. * <pre> * +------+ prev +-----+ +-----+ * head | | <---- | | <---- | | tail * +------+ +-----+ +-----+ * </pre> * * <p>Insertion into a CLH queue requires only a single atomic * operation on "tail", so there is a simple atomic point of * demarcation from unqueued to queued. Similarly, dequeuing * involves only updating the "head". However, it takes a bit * more work for nodes to determine who their successors are, * in part to deal with possible cancellation due to timeouts * and interrupts. * * <p>The "prev" links (not used in original CLH locks), are mainly * needed to handle cancellation. If a node is cancelled, its * successor is (normally) relinked to a non-cancelled * predecessor. For explanation of similar mechanics in the case * of spin locks, see the papers by Scott and Scherer at * http://www.cs.rochester.edu/u/scott/synchronization/ * * <p>We also use "next" links to implement blocking mechanics. * The thread id for each node is kept in its own node, so a * predecessor signals the next node to wake up by traversing * next link to determine which thread it is. Determination of * successor must avoid races with newly queued nodes to set * the "next" fields of their predecessors. This is solved * when necessary by checking backwards from the atomically * updated "tail" when a node's successor appears to be null. * (Or, said differently, the next-links are an optimization * so that we don't usually need a backward scan.) * * <p>Cancellation introduces some conservatism to the basic * algorithms. Since we must poll for cancellation of other * nodes, we can miss noticing whether a cancelled node is * ahead or behind us. This is dealt with by always unparking * successors upon cancellation, allowing them to stabilize on * a new predecessor, unless we can identify an uncancelled * predecessor who will carry this responsibility. * * <p>CLH queues need a dummy header node to get started. But * we don't create them on construction, because it would be wasted * effort if there is never contention. Instead, the node * is constructed and head and tail pointers are set upon first * contention. * * <p>Threads waiting on Conditions use the same nodes, but * use an additional link. Conditions only need to link nodes * in simple (non-concurrent) linked queues because they are * only accessed when exclusively held. Upon await, a node is * inserted into a condition queue. Upon signal, the node is * transferred to the main queue. A special value of status * field is used to mark which queue a node is on. * * <p>Thanks go to Dave Dice, Mark Moir, Victor Luchangco, Bill * Scherer and Michael Scott, along with members of JSR-166 * expert group, for helpful ideas, discussions, and critiques * on the design of this class. */ static final class Node { /** * 標記節點為共享模式 * */ static final Node SHARED = new Node(); /** * 標記節點為獨占模式 */ static final Node EXCLUSIVE = null; /** * 在同步隊列中等待的線程等待超時或者被中斷,需要從同步隊列中取消等待 * */ static final int CANCELLED = 1; /** * 后繼節點的線程處於等待狀態,而當前的節點如果釋放了同步狀態或者被取消, * 將會通知后繼節點,使后繼節點的線程得以運行。 */ static final int SIGNAL = -1; /** * 節點在等待隊列中,節點的線程等待在Condition上,當其他線程對Condition調用了signal()方法后, * 該節點會從等待隊列中轉移到同步隊列中,加入到同步狀態的獲取中 */ static final int CONDITION = -2; /** * 表示下一次共享式同步狀態獲取將會被無條件地傳播下去 */ static final int PROPAGATE = -3; /** * 標記當前節點的信號量狀態 (1,0,-1,-2,-3)5種狀態 * 使用CAS更改狀態,volatile保證線程可見性,高並發場景下, * 即被一個線程修改后,狀態會立馬讓其他線程可見。 */ volatile int waitStatus; /** * 前驅節點,當前節點加入到同步隊列中被設置 */ volatile Node prev; /** * 后繼節點 */ volatile Node next; /** * 節點同步狀態的線程 */ volatile Thread thread; /** * 等待隊列中的后繼節點,如果當前節點是共享的,那么這個字段是一個SHARED常量, * 也就是說節點類型(獨占和共享)和等待隊列中的后繼節點共用同一個字段。 */ Node nextWaiter; /** * Returns true if node is waiting in shared mode. */ final boolean isShared() { return nextWaiter == SHARED; } /** * 返回前驅節點 */ final Node predecessor() throws NullPointerException { Node p = prev; if (p == null) throw new NullPointerException(); else return p; } Node() { // Used to establish initial head or SHARED marker } Node(Thread thread, Node mode) { // Used by addWaiter this.nextWaiter = mode; this.thread = thread; } Node(Thread thread, int waitStatus) { // Used by Condition this.waitStatus = waitStatus; this.thread = thread; } } /** * 指向同步等待隊列的頭節點 */ private transient volatile Node head; /** * 指向同步等待隊列的尾節點 */ private transient volatile Node tail; /** * 同步資源狀態 */ private volatile int state; /** * Returns the current value of synchronization state. * This operation has memory semantics of a {@code volatile} read. * @return current state value */ protected final int getState() { return state; } /** * Sets the value of synchronization state. * This operation has memory semantics of a {@code volatile} write. * @param newState the new state value */ protected final void setState(int newState) { state = newState; } /** * Atomically sets synchronization state to the given updated * value if the current state value equals the expected value. * This operation has memory semantics of a {@code volatile} read * and write. * * @param expect the expected value * @param update the new value * @return {@code true} if successful. False return indicates that the actual * value was not equal to the expected value. */ protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); } // Queuing utilities /** * The number of nanoseconds for which it is faster to spin * rather than to use timed park. A rough estimate suffices * to improve responsiveness with very short timeouts. */ static final long spinForTimeoutThreshold = 1000L; /** * 節點加入CLH同步隊列 */ private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize //隊列為空需要初始化,創建空的頭節點 if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; //set尾部節點 if (compareAndSetTail(t, node)) {//當前節點置為尾部 t.next = node; //前驅節點的next指針指向當前節點 return t; } } } } /** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared * @return the new node */ private Node addWaiter(Node mode) { // 1. 將當前線程構建成Node類型 Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; // 2. 1當前尾節點是否為null? if (pred != null) { // 2.2 將當前節點尾插入的方式 node.prev = pred; // 2.3 CAS將節點插入同步隊列的尾部 if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; } /** * Sets head of queue to be node, thus dequeuing. Called only by * acquire methods. Also nulls out unused fields for sake of GC * and to suppress unnecessary signals and traversals. * * @param node the node */ private void setHead(Node node) { head = node; node.thread = null; node.prev = null; } /** * */ private void unparkSuccessor(Node node) { //獲取wait狀態 int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0);// 將等待狀態waitStatus設置為初始值0 /** * 若后繼結點為空,或狀態為CANCEL(已失效),則從后尾部往前遍歷找到最前的一個處於正常阻塞狀態的結點 * 進行喚醒 */ 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);//喚醒線程 } /** * 把當前結點設置為SIGNAL或者PROPAGATE * 喚醒head.next(B節點),B節點喚醒后可以競爭鎖,成功后head->B,然后又會喚醒B.next,一直重復直到共享節點都喚醒 * head節點狀態為SIGNAL,重置head.waitStatus->0,喚醒head節點線程,喚醒后線程去競爭共享鎖 * head節點狀態為0,將head.waitStatus->Node.PROPAGATE傳播狀態,表示需要將狀態向后繼節點傳播 */ private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) {//head是SIGNAL狀態 /* head狀態是SIGNAL,重置head節點waitStatus為0,這里不直接設為Node.PROPAGATE, * 是因為unparkSuccessor(h)中,如果ws < 0會設置為0,所以ws先設置為0,再設置為PROPAGATE * 這里需要控制並發,因為入口有setHeadAndPropagate跟release兩個,避免兩次unpark */ if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; //設置失敗,重新循環 /* head狀態為SIGNAL,且成功設置為0之后,喚醒head.next節點線程 * 此時head、head.next的線程都喚醒了,head.next會去競爭鎖,成功后head會指向獲取鎖的節點, * 也就是head發生了變化。看最底下一行代碼可知,head發生變化后會重新循環,繼續喚醒head的下一個節點 */ unparkSuccessor(h); /* * 如果本身頭節點的waitStatus是出於重置狀態(waitStatus==0)的,將其設置為“傳播”狀態。 * 意味着需要將狀態向后一個節點傳播 */ } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) //如果head變了,重新循環 break; } } /** * 把node節點設置成head節點,且Node.waitStatus->Node.PROPAGATE */ private void setHeadAndPropagate(Node node, int propagate) { Node h = head; //h用來保存舊的head節點 setHead(node);//head引用指向node節點 /* 這里意思有兩種情況是需要執行喚醒操作 * 1.propagate > 0 表示調用方指明了后繼節點需要被喚醒 * 2.頭節點后面的節點需要被喚醒(waitStatus<0),不論是老的頭結點還是新的頭結點 */ if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { Node s = node.next; if (s == null || s.isShared())//node是最后一個節點或者 node的后繼節點是共享節點 /* 如果head節點狀態為SIGNAL,喚醒head節點線程,重置head.waitStatus->0 * head節點狀態為0(第一次添加時是0),設置head.waitStatus->Node.PROPAGATE表示狀態需要向后繼節點傳播 */ doReleaseShared(); } } // Utilities for various versions of acquire /** * Cancels an ongoing attempt to acquire. * * @param node the node */ private void cancelAcquire(Node node) { // Ignore if node doesn't exist if (node == null) return; node.thread = null; // Skip cancelled predecessors Node pred = node.prev; while (pred.waitStatus > 0) node.prev = pred = pred.prev; // predNext is the apparent node to unsplice. CASes below will // fail if not, in which case, we lost race vs another cancel // or signal, so no further action is necessary. Node predNext = pred.next; // Can use unconditional write instead of CAS here. // After this atomic step, other Nodes can skip past us. // Before, we are free of interference from other threads. node.waitStatus = Node.CANCELLED; // If we are the tail, remove ourselves. if (node == tail && compareAndSetTail(node, pred)) { compareAndSetNext(pred, predNext, null); } else { // If successor needs signal, try to set pred's next-link // so it will get one. Otherwise wake it up to propagate. int ws; if (pred != head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread != null) { Node next = node.next; if (next != null && next.waitStatus <= 0) compareAndSetNext(pred, predNext, next); } else { unparkSuccessor(node); } node.next = node; // help GC } } /** * Checks and updates status for a node that failed to acquire. * Returns true if thread should block. This is the main signal * control in all acquire loops. Requires that pred == node.prev. * * @param pred node's predecessor holding status * @param node the node * @return {@code true} if thread should block */ private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) /* * 若前驅結點的狀態是SIGNAL,意味着當前結點可以被安全地park */ return true; if (ws > 0) { /* * 前驅節點狀態如果被取消狀態,將被移除出隊列 */ do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { /* * 當前驅節點waitStatus為 0 or PROPAGATE狀態時 * 將其設置為SIGNAL狀態,然后當前結點才可以可以被安全地park */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } /** * Convenience method to interrupt current thread. */ static void selfInterrupt() { Thread.currentThread().interrupt(); } /** * 阻塞當前節點,返回當前Thread的中斷狀態 * LockSupport.park 底層實現邏輯調用系統內核功能 pthread_mutex_lock 阻塞線程 */ private final boolean parkAndCheckInterrupt() { LockSupport.park(this);//阻塞 return Thread.interrupted(); } /* * Various flavors of acquire, varying in exclusive/shared and * control modes. Each is mostly the same, but annoyingly * different. Only a little bit of factoring is possible due to * interactions of exception mechanics (including ensuring that we * cancel if tryAcquire throws exception) and other control, at * least not without hurting performance too much. */ /** * 已經在隊列當中的Thread節點,准備阻塞等待獲取鎖 */ final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (;;) {//死循環 final Node p = node.predecessor();//找到當前結點的前驅結點 if (p == head && tryAcquire(arg)) {//如果前驅結點是頭結點,才tryAcquire,其他結點是沒有機會tryAcquire的。 setHead(node);//獲取同步狀態成功,將當前結點設置為頭結點。 p.next = null; // help GC failed = false; return interrupted; } /** * 如果前驅節點不是Head,通過shouldParkAfterFailedAcquire判斷是否應該阻塞 * 前驅節點信號量為-1,當前線程可以安全被parkAndCheckInterrupt用來阻塞線程 */ if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } /** * 與acquireQueued邏輯相似,唯一區別節點還不在隊列當中需要先進行入隊操作 */ private void doAcquireInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.EXCLUSIVE);//以獨占模式放入隊列尾部 boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } /** * 獨占模式定時獲取 */ private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { if (nanosTimeout <= 0L) return false; final long deadline = System.nanoTime() + nanosTimeout; final Node node = addWaiter(Node.EXCLUSIVE);//加入隊列 boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return true; } nanosTimeout = deadline - System.nanoTime(); if (nanosTimeout <= 0L) return false;//超時直接返回獲取失敗 if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) //阻塞指定時長,超時則線程自動被喚醒 LockSupport.parkNanos(this, nanosTimeout); if (Thread.interrupted())//當前線程中斷狀態 throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } /** * 嘗試獲取共享鎖 */ private void doAcquireShared(int arg) { final Node node = addWaiter(Node.SHARED);//入隊 boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor();//前驅節點 if (p == head) { int r = tryAcquireShared(arg); //非公平鎖實現,再嘗試獲取鎖 //state==0時tryAcquireShared會返回>=0(CountDownLatch中返回的是1)。 // state為0說明共享次數已經到了,可以獲取鎖了 if (r >= 0) {//r>0表示state==0,前繼節點已經釋放鎖,鎖的狀態為可被獲取 //這一步設置node為head節點設置node.waitStatus->Node.PROPAGATE,然后喚醒node.thread setHeadAndPropagate(node, r); p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; } } //前繼節點非head節點,將前繼節點狀態設置為SIGNAL,通過park掛起node節點的線程 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } /** * Acquires in shared interruptible mode. * @param arg the acquire argument */ 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) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } /** * Acquires in shared timed mode. * * @param arg the acquire argument * @param nanosTimeout max wait time * @return {@code true} if acquired */ private boolean doAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { if (nanosTimeout <= 0L) return false; final long deadline = System.nanoTime() + nanosTimeout; 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) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return true; } } nanosTimeout = deadline - System.nanoTime(); if (nanosTimeout <= 0L) return false; if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); if (Thread.interrupted()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } // Main exported methods /** * Attempts to acquire in exclusive mode. This method should query * if the state of the object permits it to be acquired in the * exclusive mode, and if so to acquire it. * * <p>This method is always invoked by the thread performing * acquire. If this method reports failure, the acquire method * may queue the thread, if it is not already queued, until it is * signalled by a release from some other thread. This can be used * to implement method {@link Lock#tryLock()}. * * <p>The default * implementation throws {@link UnsupportedOperationException}. * * @param arg the acquire argument. This value is always the one * passed to an acquire method, or is the value saved on entry * to a condition wait. The value is otherwise uninterpreted * and can represent anything you like. * @return {@code true} if successful. Upon success, this object has * been acquired. * @throws IllegalMonitorStateException if acquiring would place this * synchronizer in an illegal state. This exception must be * thrown in a consistent fashion for synchronization to work * correctly. * @throws UnsupportedOperationException if exclusive mode is not supported */ protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); } /** * Attempts to set the state to reflect a release in exclusive * mode. * * <p>This method is always invoked by the thread performing release. * * <p>The default implementation throws * {@link UnsupportedOperationException}. * * @param arg the release argument. This value is always the one * passed to a release method, or the current state value upon * entry to a condition wait. The value is otherwise * uninterpreted and can represent anything you like. * @return {@code true} if this object is now in a fully released * state, so that any waiting threads may attempt to acquire; * and {@code false} otherwise. * @throws IllegalMonitorStateException if releasing would place this * synchronizer in an illegal state. This exception must be * thrown in a consistent fashion for synchronization to work * correctly. * @throws UnsupportedOperationException if exclusive mode is not supported */ protected boolean tryRelease(int arg) { throw new UnsupportedOperationException(); } /** * 共享式:共享式地獲取同步狀態。對於獨占式同步組件來講,同一時刻只有一個線程能獲取到同步狀態, * 其他線程都得去排隊等待,其待重寫的嘗試獲取同步狀態的方法tryAcquire返回值為boolean,這很容易理解; * 對於共享式同步組件來講,同一時刻可以有多個線程同時獲取到同步狀態,這也是“共享”的意義所在。 * 本方法待被之類覆蓋實現具體邏輯 * 1.當返回值大於0時,表示獲取同步狀態成功,同時還有剩余同步狀態可供其他線程獲取; * * 2.當返回值等於0時,表示獲取同步狀態成功,但沒有可用同步狀態了; * 3.當返回值小於0時,表示獲取同步狀態失敗。 */ protected int tryAcquireShared(int arg) { throw new UnsupportedOperationException(); } /** * Attempts to set the state to reflect a release in shared mode. * * <p>This method is always invoked by the thread performing release. * * <p>The default implementation throws * {@link UnsupportedOperationException}. * * @param arg the release argument. This value is always the one * passed to a release method, or the current state value upon * entry to a condition wait. The value is otherwise * uninterpreted and can represent anything you like. * @return {@code true} if this release of shared mode may permit a * waiting acquire (shared or exclusive) to succeed; and * {@code false} otherwise * @throws IllegalMonitorStateException if releasing would place this * synchronizer in an illegal state. This exception must be * thrown in a consistent fashion for synchronization to work * correctly. * @throws UnsupportedOperationException if shared mode is not supported */ protected boolean tryReleaseShared(int arg) { throw new UnsupportedOperationException(); } /** * Returns {@code true} if synchronization is held exclusively with * respect to the current (calling) thread. This method is invoked * upon each call to a non-waiting {@link ConditionObject} method. * (Waiting methods instead invoke {@link #release}.) * * <p>The default implementation throws {@link * UnsupportedOperationException}. This method is invoked * internally only within {@link ConditionObject} methods, so need * not be defined if conditions are not used. * * @return {@code true} if synchronization is held exclusively; * {@code false} otherwise * @throws UnsupportedOperationException if conditions are not supported */ protected boolean isHeldExclusively() { throw new UnsupportedOperationException(); } /** * Acquires in exclusive mode, ignoring interrupts. Implemented * by invoking at least once {@link #tryAcquire}, * returning on success. Otherwise the thread is queued, possibly * repeatedly blocking and unblocking, invoking {@link * #tryAcquire} until success. This method can be used * to implement method {@link Lock#lock}. * * @param arg the acquire argument. This value is conveyed to * {@link #tryAcquire} but is otherwise uninterpreted and * can represent anything you like. */ public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } /** * Acquires in exclusive mode, aborting if interrupted. * Implemented by first checking interrupt status, then invoking * at least once {@link #tryAcquire}, returning on * success. Otherwise the thread is queued, possibly repeatedly * blocking and unblocking, invoking {@link #tryAcquire} * until success or the thread is interrupted. This method can be * used to implement method {@link Lock#lockInterruptibly}. * * @param arg the acquire argument. This value is conveyed to * {@link #tryAcquire} but is otherwise uninterpreted and * can represent anything you like. * @throws InterruptedException if the current thread is interrupted */ public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (!tryAcquire(arg)) doAcquireInterruptibly(arg); } /** * Attempts to acquire in exclusive mode, aborting if interrupted, * and failing if the given timeout elapses. Implemented by first * checking interrupt status, then invoking at least once {@link * #tryAcquire}, returning on success. Otherwise, the thread is * queued, possibly repeatedly blocking and unblocking, invoking * {@link #tryAcquire} until success or the thread is interrupted * or the timeout elapses. This method can be used to implement * method {@link Lock#tryLock(long, TimeUnit)}. * * @param arg the acquire argument. This value is conveyed to * {@link #tryAcquire} but is otherwise uninterpreted and * can represent anything you like. * @param nanosTimeout the maximum number of nanoseconds to wait * @return {@code true} if acquired; {@code false} if timed out * @throws InterruptedException if the current thread is interrupted */ public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); return tryAcquire(arg) || doAcquireNanos(arg, nanosTimeout); } /** * 釋放獨占模式持有的鎖 */ public final boolean release(int arg) { if (tryRelease(arg)) {//釋放一次鎖 Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h);//喚醒后繼結點 return true; } return false; } /** * 請求獲取共享鎖 */ public final void acquireShared(int arg) { if (tryAcquireShared(arg) < 0)//返回值小於0,獲取同步狀態失敗,排隊去;獲取同步狀態成功,直接返回去干自己的事兒。 doAcquireShared(arg); } /** * 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. * @param arg the acquire argument. * This value is conveyed to {@link #tryAcquireShared} but is * otherwise uninterpreted and can represent anything * you like. * @throws InterruptedException if the current thread is interrupted */ public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); } /** * Attempts to acquire in shared mode, aborting if interrupted, and * failing if the given timeout elapses. 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 or the timeout elapses. * * @param arg the acquire argument. This value is conveyed to * {@link #tryAcquireShared} but is otherwise uninterpreted * and can represent anything you like. * @param nanosTimeout the maximum number of nanoseconds to wait * @return {@code true} if acquired; {@code false} if timed out * @throws InterruptedException if the current thread is interrupted */ public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); return tryAcquireShared(arg) >= 0 || doAcquireSharedNanos(arg, nanosTimeout); } /** * Releases in shared mode. Implemented by unblocking one or more * threads if {@link #tryReleaseShared} returns true. * * @param arg the release argument. This value is conveyed to * {@link #tryReleaseShared} but is otherwise uninterpreted * and can represent anything you like. * @return the value returned from {@link #tryReleaseShared} */ public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; } // Queue inspection methods /** * Queries whether any threads are waiting to acquire. Note that * because cancellations due to interrupts and timeouts may occur * at any time, a {@code true} return does not guarantee that any * other thread will ever acquire. * * <p>In this implementation, this operation returns in * constant time. * * @return {@code true} if there may be other threads waiting to acquire */ public final boolean hasQueuedThreads() { return head != tail; } /** * Queries whether any threads have ever contended to acquire this * synchronizer; that is if an acquire method has ever blocked. * * <p>In this implementation, this operation returns in * constant time. * * @return {@code true} if there has ever been contention */ public final boolean hasContended() { return head != null; } /** * Returns the first (longest-waiting) thread in the queue, or * {@code null} if no threads are currently queued. * * <p>In this implementation, this operation normally returns in * constant time, but may iterate upon contention if other threads are * concurrently modifying the queue. * * @return the first (longest-waiting) thread in the queue, or * {@code null} if no threads are currently queued */ public final Thread getFirstQueuedThread() { // handle only fast path, else relay return (head == tail) ? null : fullGetFirstQueuedThread(); } /** * Version of getFirstQueuedThread called when fastpath fails */ private Thread fullGetFirstQueuedThread() { /* * The first node is normally head.next. Try to get its * thread field, ensuring consistent reads: If thread * field is nulled out or s.prev is no longer head, then * some other thread(s) concurrently performed setHead in * between some of our reads. We try this twice before * resorting to traversal. */ Node h, s; Thread st; if (((h = head) != null && (s = h.next) != null && s.prev == head && (st = s.thread) != null) || ((h = head) != null && (s = h.next) != null && s.prev == head && (st = s.thread) != null)) return st; /* * Head's next field might not have been set yet, or may have * been unset after setHead. So we must check to see if tail * is actually first node. If not, we continue on, safely * traversing from tail back to head to find first, * guaranteeing termination. */ Node t = tail; Thread firstThread = null; while (t != null && t != head) { Thread tt = t.thread; if (tt != null) firstThread = tt; t = t.prev; } return firstThread; } /** * 判斷當前線程是否在隊列當中 */ public final boolean isQueued(Thread thread) { if (thread == null) throw new NullPointerException(); for (Node p = tail; p != null; p = p.prev) if (p.thread == thread) return true; return false; } /** * Returns {@code true} if the apparent first queued thread, if one * exists, is waiting in exclusive mode. If this method returns * {@code true}, and the current thread is attempting to acquire in * shared mode (that is, this method is invoked from {@link * #tryAcquireShared}) then it is guaranteed that the current thread * is not the first queued thread. Used only as a heuristic in * ReentrantReadWriteLock. */ final boolean apparentlyFirstQueuedIsExclusive() { Node h, s; return (h = head) != null && (s = h.next) != null && !s.isShared() && s.thread != null; } /** * 判斷當前節點是否有前驅節點 */ public final boolean hasQueuedPredecessors() { Node t = tail; // Read fields in reverse initialization order Node h = head; Node s; return h != t && ((s = h.next) == null || s.thread != Thread.currentThread()); } // Instrumentation and monitoring methods /** * 同步隊列長度 */ public final int getQueueLength() { int n = 0; for (Node p = tail; p != null; p = p.prev) { if (p.thread != null) ++n; } return n; } /** * 獲取隊列等待thread集合 */ public final Collection<Thread> getQueuedThreads() { ArrayList<Thread> list = new ArrayList<Thread>(); for (Node p = tail; p != null; p = p.prev) { Thread t = p.thread; if (t != null) list.add(t); } return list; } /** * 獲取獨占模式等待thread線程集合 */ public final Collection<Thread> getExclusiveQueuedThreads() { ArrayList<Thread> list = new ArrayList<Thread>(); for (Node p = tail; p != null; p = p.prev) { if (!p.isShared()) { Thread t = p.thread; if (t != null) list.add(t); } } return list; } /** * 獲取共享模式等待thread集合 */ public final Collection<Thread> getSharedQueuedThreads() { ArrayList<Thread> list = new ArrayList<Thread>(); for (Node p = tail; p != null; p = p.prev) { if (p.isShared()) { Thread t = p.thread; if (t != null) list.add(t); } } return list; } /** * Returns a string identifying this synchronizer, as well as its state. * The state, in brackets, includes the String {@code "State ="} * followed by the current value of {@link #getState}, and either * {@code "nonempty"} or {@code "empty"} depending on whether the * queue is empty. * * @return a string identifying this synchronizer, as well as its state */ public String toString() { int s = getState(); String q = hasQueuedThreads() ? "non" : ""; return super.toString() + "[State = " + s + ", " + q + "empty queue]"; } // Internal support methods for Conditions /** * 判斷節點是否在同步隊列中 */ final boolean isOnSyncQueue(Node node) { //快速判斷1:節點狀態或者節點沒有前置節點 //注:同步隊列是有頭節點的,而條件隊列沒有 if (node.waitStatus == Node.CONDITION || node.prev == null) return false; //快速判斷2:next字段只有同步隊列才會使用,條件隊列中使用的是nextWaiter字段 if (node.next != null) // If has successor, it must be on queue return true; //上面如果無法判斷則進入復雜判斷 return findNodeFromTail(node); } /** * Returns true if node is on sync queue by searching backwards from tail. * Called only when needed by isOnSyncQueue. * @return true if present */ private boolean findNodeFromTail(Node node) { Node t = tail; for (;;) { if (t == node) return true; if (t == null) return false; t = t.prev; } } /** * 將節點從條件隊列當中移動到同步隊列當中,等待獲取鎖 */ final boolean transferForSignal(Node node) { /* * 修改節點信號量狀態為0,失敗直接返回false */ if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) return false; /* * 加入同步隊列尾部當中,返回前驅節點 */ Node p = enq(node); int ws = p.waitStatus; //前驅節點不可用 或者 修改信號量狀態失敗 if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) LockSupport.unpark(node.thread); //喚醒當前節點 return true; } /** * Transfers node, if necessary, to sync queue after a cancelled wait. * Returns true if thread was cancelled before being signalled. * * @param node the node * @return true if cancelled before the node was signalled */ final boolean transferAfterCancelledWait(Node node) { if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) { enq(node); return true; } /* * If we lost out to a signal(), then we can't proceed * until it finishes its enq(). Cancelling during an * incomplete transfer is both rare and transient, so just * spin. */ while (!isOnSyncQueue(node)) Thread.yield(); return false; } /** * 入參就是新創建的節點,即當前節點 */ final int fullyRelease(Node node) { boolean failed = true; try { //這里這個取值要注意,獲取當前的state並釋放,這從另一個角度說明必須是獨占鎖 //可以考慮下這個邏輯放在共享鎖下面會發生什么? int savedState = getState(); if (release(savedState)) { failed = false; return savedState; } else { //如果這里釋放失敗,則拋出異常 throw new IllegalMonitorStateException(); } } finally { /** * 如果釋放鎖失敗,則把節點取消,由這里就能看出來上面添加節點的邏輯中 * 只需要判斷最后一個節點是否被取消就可以了 */ if (failed) node.waitStatus = Node.CANCELLED; } } // Instrumentation methods for conditions /** * Queries whether the given ConditionObject * uses this synchronizer as its lock. * * @param condition the condition * @return {@code true} if owned * @throws NullPointerException if the condition is null */ public final boolean owns(ConditionObject condition) { return condition.isOwnedBy(this); } /** * Queries whether any threads are waiting on the given condition * associated with this synchronizer. Note that because timeouts * and interrupts may occur at any time, a {@code true} return * does not guarantee that a future {@code signal} will awaken * any threads. This method is designed primarily for use in * monitoring of the system state. * * @param condition the condition * @return {@code true} if there are any waiting threads * @throws IllegalMonitorStateException if exclusive synchronization * is not held * @throws IllegalArgumentException if the given condition is * not associated with this synchronizer * @throws NullPointerException if the condition is null */ public final boolean hasWaiters(ConditionObject condition) { if (!owns(condition)) throw new IllegalArgumentException("Not owner"); return condition.hasWaiters(); } /** * 獲取條件隊列長度 */ public final int getWaitQueueLength(ConditionObject condition) { if (!owns(condition)) throw new IllegalArgumentException("Not owner"); return condition.getWaitQueueLength(); } /** * 獲取條件隊列當中所有等待的thread集合 */ public final Collection<Thread> getWaitingThreads(ConditionObject condition) { if (!owns(condition)) throw new IllegalArgumentException("Not owner"); return condition.getWaitingThreads(); } /** * 條件對象,實現基於條件的具體行為 */ public class ConditionObject implements Condition, java.io.Serializable { private static final long serialVersionUID = 1173984872572414699L; /** First node of condition queue. */ private transient Node firstWaiter; /** Last node of condition queue. */ private transient Node lastWaiter; /** * Creates a new {@code ConditionObject} instance. */ public ConditionObject() { } // Internal methods /** * 1.與同步隊列不同,條件隊列頭尾指針是firstWaiter跟lastWaiter * 2.條件隊列是在獲取鎖之后,也就是臨界區進行操作,因此很多地方不用考慮並發 */ private Node addConditionWaiter() { Node t = lastWaiter; //如果最后一個節點被取消,則刪除隊列中被取消的節點 //至於為啥是最后一個節點后面會分析 if (t != null && t.waitStatus != Node.CONDITION) { //刪除所有被取消的節點 unlinkCancelledWaiters(); t = lastWaiter; } //創建一個類型為CONDITION的節點並加入隊列,由於在臨界區,所以這里不用並發控制 Node node = new Node(Thread.currentThread(), Node.CONDITION); if (t == null) firstWaiter = node; else t.nextWaiter = node; lastWaiter = node; return node; } /** * 發信號,通知遍歷條件隊列當中的節點轉移到同步隊列當中,准備排隊獲取鎖 */ private void doSignal(Node first) { do { if ( (firstWaiter = first.nextWaiter) == null) lastWaiter = null; first.nextWaiter = null; } while (!transferForSignal(first) && //轉移節點 (first = firstWaiter) != null); } /** * 通知所有節點移動到同步隊列當中,並將節點從條件隊列刪除 */ private void doSignalAll(Node first) { lastWaiter = firstWaiter = null; do { Node next = first.nextWaiter; first.nextWaiter = null; transferForSignal(first); first = next; } while (first != null); } /** * 刪除條件隊列當中被取消的節點 */ private void unlinkCancelledWaiters() { Node t = firstWaiter; Node trail = null; while (t != null) { Node next = t.nextWaiter; if (t.waitStatus != Node.CONDITION) { t.nextWaiter = null; if (trail == null) firstWaiter = next; else trail.nextWaiter = next; if (next == null) lastWaiter = trail; } else trail = t; t = next; } } // public methods /** * 發新號,通知條件隊列當中節點到同步隊列當中去排隊 * */ public final void signal() { if (!isHeldExclusively())//節點不能已經持有獨占鎖 throw new IllegalMonitorStateException(); Node first = firstWaiter; if (first != null) /** * 發信號通知條件隊列的節點准備到同步隊列當中去排隊 */ doSignal(first); } /** * 喚醒所有條件隊列的節點轉移到同步隊列當中 */ public final void signalAll() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); Node first = firstWaiter; if (first != null) doSignalAll(first); } /** * Implements uninterruptible condition wait. * <ol> * <li> Save lock state returned by {@link #getState}. * <li> Invoke {@link #release} with saved state as argument, * throwing IllegalMonitorStateException if it fails. * <li> Block until signalled. * <li> Reacquire by invoking specialized version of * {@link #acquire} with saved state as argument. * </ol> */ public final void awaitUninterruptibly() { Node node = addConditionWaiter(); int savedState = fullyRelease(node); boolean interrupted = false; while (!isOnSyncQueue(node)) { LockSupport.park(this); if (Thread.interrupted()) interrupted = true; } if (acquireQueued(node, savedState) || interrupted) selfInterrupt(); } /* * For interruptible waits, we need to track whether to throw * InterruptedException, if interrupted while blocked on * condition, versus reinterrupt current thread, if * interrupted while blocked waiting to re-acquire. */ /** 該模式表示在退出等待時重新中斷 */ private static final int REINTERRUPT = 1; /** 異常中斷 */ private static final int THROW_IE = -1; /** * 這里的判斷邏輯是: * 1.如果現在不是中斷的,即正常被signal喚醒則返回0 * 2.如果節點由中斷加入同步隊列則返回THROW_IE,由signal加入同步隊列則返回REINTERRUPT */ private int checkInterruptWhileWaiting(Node node) { return Thread.interrupted() ? (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0; } /** * 根據中斷時機選擇拋出異常或者設置線程中斷狀態 */ private void reportInterruptAfterWait(int interruptMode) throws InterruptedException { if (interruptMode == THROW_IE) throw new InterruptedException(); else if (interruptMode == REINTERRUPT) selfInterrupt(); } /** * 加入條件隊列等待,條件隊列入口 */ public final void await() throws InterruptedException { //如果當前線程被中斷則直接拋出異常 if (Thread.interrupted()) throw new InterruptedException(); //把當前節點加入條件隊列 Node node = addConditionWaiter(); //釋放掉已經獲取的獨占鎖資源 int savedState = fullyRelease(node); int interruptMode = 0; //如果不在同步隊列中則不斷掛起 while (!isOnSyncQueue(node)) { LockSupport.park(this); //這里被喚醒可能是正常的signal操作也可能是中斷 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; } /** * 走到這里說明節點已經條件滿足被加入到了同步隊列中或者中斷了 * 這個方法很熟悉吧?就跟獨占鎖調用同樣的獲取鎖方法,從這里可以看出條件隊列只能用於獨占鎖 * 在處理中斷之前首先要做的是從同步隊列中成功獲取鎖資源 */ if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; //走到這里說明已經成功獲取到了獨占鎖,接下來就做些收尾工作 //刪除條件隊列中被取消的節點 if (node.nextWaiter != null) // clean up if cancelled unlinkCancelledWaiters(); //根據不同模式處理中斷 if (interruptMode != 0) reportInterruptAfterWait(interruptMode); } /** * Implements timed condition wait. * <ol> * <li> If current thread is interrupted, throw InterruptedException. * <li> Save lock state returned by {@link #getState}. * <li> Invoke {@link #release} with saved state as argument, * throwing IllegalMonitorStateException if it fails. * <li> Block until signalled, interrupted, or timed out. * <li> Reacquire by invoking specialized version of * {@link #acquire} with saved state as argument. * <li> If interrupted while blocked in step 4, throw InterruptedException. * </ol> */ public final long awaitNanos(long nanosTimeout) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); Node node = addConditionWaiter(); int savedState = fullyRelease(node); final long deadline = System.nanoTime() + nanosTimeout; int interruptMode = 0; while (!isOnSyncQueue(node)) { if (nanosTimeout <= 0L) { transferAfterCancelledWait(node); break; } if (nanosTimeout >= spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; nanosTimeout = deadline - System.nanoTime(); } if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; if (node.nextWaiter != null) unlinkCancelledWaiters(); if (interruptMode != 0) reportInterruptAfterWait(interruptMode); return deadline - System.nanoTime(); } /** * Implements absolute timed condition wait. * <ol> * <li> If current thread is interrupted, throw InterruptedException. * <li> Save lock state returned by {@link #getState}. * <li> Invoke {@link #release} with saved state as argument, * throwing IllegalMonitorStateException if it fails. * <li> Block until signalled, interrupted, or timed out. * <li> Reacquire by invoking specialized version of * {@link #acquire} with saved state as argument. * <li> If interrupted while blocked in step 4, throw InterruptedException. * <li> If timed out while blocked in step 4, return false, else true. * </ol> */ public final boolean awaitUntil(Date deadline) throws InterruptedException { long abstime = deadline.getTime(); if (Thread.interrupted()) throw new InterruptedException(); Node node = addConditionWaiter(); int savedState = fullyRelease(node); boolean timedout = false; int interruptMode = 0; while (!isOnSyncQueue(node)) { if (System.currentTimeMillis() > abstime) { timedout = transferAfterCancelledWait(node); break; } LockSupport.parkUntil(this, abstime); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; } if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; if (node.nextWaiter != null) unlinkCancelledWaiters(); if (interruptMode != 0) reportInterruptAfterWait(interruptMode); return !timedout; } /** * Implements timed condition wait. * <ol> * <li> If current thread is interrupted, throw InterruptedException. * <li> Save lock state returned by {@link #getState}. * <li> Invoke {@link #release} with saved state as argument, * throwing IllegalMonitorStateException if it fails. * <li> Block until signalled, interrupted, or timed out. * <li> Reacquire by invoking specialized version of * {@link #acquire} with saved state as argument. * <li> If interrupted while blocked in step 4, throw InterruptedException. * <li> If timed out while blocked in step 4, return false, else true. * </ol> */ public final boolean await(long time, TimeUnit unit) throws InterruptedException { long nanosTimeout = unit.toNanos(time); if (Thread.interrupted()) throw new InterruptedException(); Node node = addConditionWaiter(); int savedState = fullyRelease(node); final long deadline = System.nanoTime() + nanosTimeout; boolean timedout = false; int interruptMode = 0; while (!isOnSyncQueue(node)) { if (nanosTimeout <= 0L) { timedout = transferAfterCancelledWait(node); break; } if (nanosTimeout >= spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; nanosTimeout = deadline - System.nanoTime(); } if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; if (node.nextWaiter != null) unlinkCancelledWaiters(); if (interruptMode != 0) reportInterruptAfterWait(interruptMode); return !timedout; } // support for instrumentation /** * Returns true if this condition was created by the given * synchronization object. * * @return {@code true} if owned */ final boolean isOwnedBy(AbstractQueuedSynchronizer sync) { return sync == AbstractQueuedSynchronizer.this; } /** * Queries whether any threads are waiting on this condition. * Implements {@link AbstractQueuedSynchronizer#hasWaiters(ConditionObject)}. * * @return {@code true} if there are any waiting threads * @throws IllegalMonitorStateException if {@link #isHeldExclusively} * returns {@code false} */ protected final boolean hasWaiters() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); for (Node w = firstWaiter; w != null; w = w.nextWaiter) { if (w.waitStatus == Node.CONDITION) return true; } return false; } /** * Returns an estimate of the number of threads waiting on * this condition. * Implements {@link AbstractQueuedSynchronizer#getWaitQueueLength(ConditionObject)}. * * @return the estimated number of waiting threads * @throws IllegalMonitorStateException if {@link #isHeldExclusively} * returns {@code false} */ protected final int getWaitQueueLength() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); int n = 0; for (Node w = firstWaiter; w != null; w = w.nextWaiter) { if (w.waitStatus == Node.CONDITION) ++n; } return n; } /** * 得到同步隊列當中所有在等待的Thread集合 */ protected final Collection<Thread> getWaitingThreads() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); ArrayList<Thread> list = new ArrayList<Thread>(); for (Node w = firstWaiter; w != null; w = w.nextWaiter) { if (w.waitStatus == Node.CONDITION) { Thread t = w.thread; if (t != null) list.add(t); } } return list; } } /** * Setup to support compareAndSet. We need to natively implement * this here: For the sake of permitting future enhancements, we * cannot explicitly subclass AtomicInteger, which would be * efficient and useful otherwise. So, as the lesser of evils, we * natively implement using hotspot intrinsics API. And while we * are at it, we do the same for other CASable fields (which could * otherwise be done with atomic field updaters). * unsafe魔法類,直接繞過虛擬機內存管理機制,修改內存 */ private static final Unsafe unsafe = Unsafe.getUnsafe(); private static final long stateOffset; private static final long headOffset; private static final long tailOffset; private static final long waitStatusOffset; private static final long nextOffset; static { try { stateOffset = unsafe.objectFieldOffset (AbstractQueuedSynchronizer.class.getDeclaredField("state")); headOffset = unsafe.objectFieldOffset (AbstractQueuedSynchronizer.class.getDeclaredField("head")); tailOffset = unsafe.objectFieldOffset (AbstractQueuedSynchronizer.class.getDeclaredField("tail")); waitStatusOffset = unsafe.objectFieldOffset (Node.class.getDeclaredField("waitStatus")); nextOffset = unsafe.objectFieldOffset (Node.class.getDeclaredField("next")); } catch (Exception ex) { throw new Error(ex); } } /** * CAS 修改頭部節點指向. 並發入隊時使用. */ private final boolean compareAndSetHead(Node update) { return unsafe.compareAndSwapObject(this, headOffset, null, update); } /** * CAS 修改尾部節點指向. 並發入隊時使用. */ private final boolean compareAndSetTail(Node expect, Node update) { return unsafe.compareAndSwapObject(this, tailOffset, expect, update); } /** * CAS 修改信號量狀態. */ private static final boolean compareAndSetWaitStatus(Node node, int expect, int update) { return unsafe.compareAndSwapInt(node, waitStatusOffset, expect, update); } /** * 修改節點的后繼指針. */ private static final boolean compareAndSetNext(Node node, Node expect, Node update) { return unsafe.compareAndSwapObject(node, nextOffset, expect, update); } }
AbstractOwnableSynchronizer.java

public abstract class AbstractOwnableSynchronizer implements java.io.Serializable { private static final long serialVersionUID = 3737899427754241961L; protected AbstractOwnableSynchronizer() { } /** * 獨占模式同步器的當前持有線程. * transient關鍵字表示屬性不參與序列化 */ private transient Thread exclusiveOwnerThread; protected final void setExclusiveOwnerThread(Thread thread) { exclusiveOwnerThread = thread; } protected final Thread getExclusiveOwnerThread() { return exclusiveOwnerThread; } }