從 AbstractQueuedSynchronizer 理解 ReentrantLock
簡介
Java 併發編程離不開鎖， Synchronized 是經常使用的一種實現加鎖的方式，使用比較簡單快捷。在 Java 中還有另外一種鎖，即 Lock 鎖。 Lock 是一個接口，提供了超時阻塞、可響應中斷以及公平非公平鎖等特性，相比於 Synchronized，Lock 功能更強大，能夠實現更靈活的加鎖方式。java
Lock 的主要實現類是 ReentrantLock，而 ReetrantLock 中具體的實現方式是利用另一個類 AbstractQueuedSynchronizer，全部的操做都是委託給這個類完成。AbstractQueuedSynchronizer 是 Lock 鎖的重要組件，本文從 AbstractQueuedSynchronizer 來分析 ReetrantLock 的實現原理。node
基本用法
先看一下 Lock 的基本用法：編程
Lock lock = ...;
lock.lock();
try{
//處理任務
}catch(Exception ex){
}finally{
lock.unlock(); //釋放鎖
}
lock.lock() 便是加鎖， lock.unolck() 是釋放鎖，爲了保證所可以釋放，unlock() 應該放到 finally 中。併發
下面分別從 lock() 和 unlock() 方法來分析加鎖和解鎖到底作了什麼。less
lock
下面是 lock() 的代碼：ui
public void lock() {
sync.lock();
}
能夠看到，只是簡單調用了 sync 對應的 lock() 方法。那麼這個 sync 是什麼呢？其實這個就是 AbstractQueuedSynchronizer 的實現類。能夠看一下 ReentrantLock 的構造方法：this
/**
* Creates an instance of {@code ReentrantLock}.
* This is equivalent to using {@code ReentrantLock(false)}.
*/
public ReentrantLock() {
sync = new NonfairSync();
}
/**
* Creates an instance of {@code ReentrantLock} with the
* given fairness policy.
*
* @param fair {@code true} if this lock should use a fair ordering policy
*/
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
ReentrantLock 有兩個方法，主要目的是選擇是公平鎖仍是非公平鎖。公平鎖指的是先來後到，先爭搶鎖的線程先得到鎖，而非公平鎖則不必定。ReentrantLock 默認使用的是非公平鎖，也能夠經過構造參數選擇公平鎖。選擇哪一個鎖實際上是生成了一個對象並賦值給變量 sync，下面是涉及到的代碼：spa
/**
* Base of synchronization control for this lock. Subclassed
* into fair and nonfair versions below. Uses AQS state to
* represent the number of holds on the lock.
*/
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L;
/**
* Performs {@link Lock#lock}. The main reason for subclassing
* is to allow fast path for nonfair version.
*/
abstract void lock();
/**
* Performs non-fair tryLock. tryAcquire is implemented in
* subclasses, but both need nonfair try for trylock method.
*/
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
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() {
// 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();
}
final ConditionObject newCondition() {
return new ConditionObject();
}
// Methods relayed from outer class
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;
/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
/**
* Sync object for fair locks
* 公平鎖
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
final void lock() {
acquire(1);
}
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
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;
}
}
在 ReentrantLock 中有一個抽象的內部類 Sync，繼承於 AbstractQueuedSynchronizer 並實現了一些方法。另有兩個類 FairSync 和 NoFairSync 繼承了 Sync，它們天然就是公平鎖以及非公平鎖的實現。下面分析將從公平鎖出發，非公平鎖與公平鎖差異並非不少。線程
公平鎖 FairSync 加鎖的代碼以下：3d
final void lock() {
acquire(1);
}
只有一行，調用了 acquire，這是 AbstractQueuedSynchronizer 中的一個方法，代碼以下：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
acquire 的實現也很短，很少在其中卻包含了加鎖的具體實現，關鍵就在內部調用的幾個方法中。
爲了理解加鎖和解鎖的過程，下面具體介紹一下 AbstractQueuedSynchronizer（如下簡稱 AQS）。
AbstractQueuedSynchronizer
AQS 中使用一個同步隊列來實現線程同步狀態的管理，當一個線程獲取鎖失敗的時候， AQS將此線程構形成一個節點(Node)並加入同步隊列而且阻塞線程。當鎖釋放時，會從同步隊列中將第一個節點喚醒並使其再次獲取鎖。
同步隊列中的節點用來保存獲取鎖失敗的線程的相關信息，包含以下屬性：
static final class Node {
/** Marker to indicate a node is waiting in shared mode */
// 標識共享模式
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
// 標識獨佔模式
static final Node EXCLUSIVE = null;
/** waitStatus value to indicate thread has cancelled. */
// 線程取消
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking. */
// 須要喚醒後繼節點
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition. */
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate.
*/
static final int PROPAGATE = -3;
// 節點狀態，爲上面的幾個狀態之一
volatile int waitStatus;
// 前置節點
volatile Node prev;
// 後繼節點
volatile Node next;
// 節點所表示的線程
volatile Thread thread;
Node nextWaiter;
...
}
Node 是 AQS 的內部類，其中包含一些屬性標識一個阻塞線程的節點，包括是獨佔模式仍是共享模式、節點的狀態、前驅結點、後繼結點以及節點所表明的線程。
同步隊列是一個雙向列表，在 AQS 中有這樣幾個屬性：
/**
* Head of the wait queue, lazily initialized. Except for
* initialization, it is modified only via method setHead. Note:
* If head exists, its waitStatus is guaranteed not to be
* CANCELLED.
*/
// 頭結點
private transient volatile Node head;
/**
* Tail of the wait queue, lazily initialized. Modified only via
* method enq to add new wait node.
*/
// 尾節點
private transient volatile Node tail;
/**
* The synchronization state.
*/
// 鎖的狀態
private volatile int state;
其中，head 和 tail 分別指向同步隊列的頭結點和尾節點，state 標識鎖當前的狀態，爲 0 時表示當前鎖未被佔用，大於 1 表示被佔用，之因此是大於 1 是由於鎖能夠重入，每重入一次增長 1。同步隊列的結構大體以下圖：
瞭解了同步隊列後，下面具體看看加鎖和解鎖的過程。
加鎖
final void lock() {
acquire(1);
}
public final void acquire(int arg) {
// 加鎖的主要代碼
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
主要邏輯其實就是一行代碼：if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))，tryAcquire 是嘗試獲取一下鎖，爲何說是嘗試呢？看代碼：
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState(); // 獲取當前狀態
if (c == 0) {
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;
}
tryAcquire 能夠分爲三條分支：
- 當前鎖未被佔用（
getState() == 0），則判斷是否有前驅結點，沒有的話就用 CAS 加鎖(compareAndSetState(0, acquires))，加鎖成功則調用setExclusiveOwnerThread(current)標示一下並返回true。 - 當前線程已經獲取過這個鎖，則此時是重入，改變
state的計數便可，返回true表示加鎖成功。 - 若是不是上面兩種狀況，那麼說明鎖被佔用或者 CAS 沒有搶過其它線程，則須要進入同步隊列，返回
false表示嘗試加鎖失敗。
回到 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 這一行，若是 tryAcquire(arg) 返回 false 將會執行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)。先看一下 addWaiter(Node.EXCLUSIVE) ，這個方法的代碼以下所示：
private Node addWaiter(Node mode) {
// 將線程包裝成 Node
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
// 尾節點爲空
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
其中主要邏輯是將當前線程包裝爲一個 Node 節點並加入同步隊列。若是尾節點爲空，則用 CAS 設置尾節點，若是入隊失敗則調用 enq(node)，這個方法內部是一個循環，利用自旋 CAS 把節點加入同步隊列，具體代碼就不分析了。
在節點加入隊列以後，執行的是 acquireQueued 方法，代碼以下：
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
// 這是一個無限循環
for (;;) {
final Node p = node.predecessor();
// 若是前驅節點是 head，則嘗試獲取鎖
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 獲取鎖失敗則進入判斷是否要進入睡眠
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
acquireQueued 實現了線程的睡眠與喚醒。在內部是一個無限循環，每次獲取前驅節點，若是前驅結點是 HEAD，那麼嘗試去獲取鎖，獲取成功則將此節點變爲新的頭結點並將原先的頭結點出隊。若是前驅節點不是頭結點或者獲取鎖失敗，那麼就會進入 shouldParkAfterFailedAcquire 方法，判斷是否進入睡眠，若是這個方法返回 true，則調用 parkAndCheckInterrupt 讓線程進入睡眠狀態。下面是 parkAndCheckInterrupt 的代碼：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 線程在這一步進入阻塞狀態
return Thread.interrupted();
}
對於 shouldParkAfterFailedAcquire 來講，若是前驅節點正常，那麼會返回 true，表示當前線程應該掛起，若是前驅結點取消了排隊，那麼當前線程有機會搶鎖，此時返回 false，並繼續 acquireQueued 中的循環。
解鎖
相比於加鎖，解鎖稍微簡單一點，看一下 unlock 的代碼：
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
// 嘗試解鎖
if (tryRelease(arg)) {
// 解鎖成功
Node h = head;
if (h != null && h.waitStatus != 0)
// 喚醒後繼節點
unparkSuccessor(h);
return true;
}
return false;
}
首先調用 tryRelease 解鎖，若是解鎖成功則喚醒後繼結點，返回值表示是否成功釋放鎖。那爲何會解鎖不成功，實際上是由於重入，看一下 tryRelease 的代碼：
protected final boolean tryRelease(int releases) {
// 更新 state 計數值
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;
}
將 state 減去對應的值，若是 state == 0，那麼說明鎖已經徹底釋放。
在 release 中，若是鎖已經徹底釋放，那麼將調用 unparkSuccessor 喚醒後繼節點，喚醒的節點所表明的線程阻塞在 parkAndCheckInterrupt 中：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 線程在這一步進入阻塞狀態
return Thread.interrupted();
}
線程被喚醒後，將繼續 acquireQueued 中的循環，嘗試獲取鎖。
總結
本文簡要分析了 Lock 鎖的原理，主要是利用 AbstractQueuedSynchronizer這個關鍵的類。AQS 的核心在於使用 CAS 更新鎖的狀態，並利用一個同步隊列將獲取鎖失敗的線程進行排隊，當前驅節點解鎖後再喚醒後繼節點，是一個幾乎純 Java 實現的加鎖與解鎖。
若是個人文章對您有幫助，不妨點個贊支持一下(^_^)
- 1. 從ReentrantLock來理解AbstractQueuedSynchronizer
- 2. 從AbstractQueuedSynchronizer到Reentrantlock。
- 3. AbstractQueuedSynchronizer理解之一（ReentrantLock）
- 4. AbstractQueuedSynchronizer理解
- 5. [源碼分析]ReentrantLock & AbstractQueuedSynchronizer & Condition
- 6. JUC源碼解析--ReentrantLock和AbstractQueuedSynchronizer
- 7. 深刻理解ReentrantLock
- 8. 深入理解ReentrantLock
- 9. 深刻理解AbstractQueuedSynchronizer
- 10. 從源碼角度完全理解ReentrantLock
- 更多相關文章...
- • SQLite Explain（解釋） - SQLite教程
- • 從RedisTemplate中獲得Jedis實例 - Redis教程
- • Docker 清理命令
- • ☆技術問答集錦（13）Java Instrument原理
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 從ReentrantLock來理解AbstractQueuedSynchronizer
- 2. 從AbstractQueuedSynchronizer到Reentrantlock。
- 3. AbstractQueuedSynchronizer理解之一（ReentrantLock）
- 4. AbstractQueuedSynchronizer理解
- 5. [源碼分析]ReentrantLock & AbstractQueuedSynchronizer & Condition
- 6. JUC源碼解析--ReentrantLock和AbstractQueuedSynchronizer
- 7. 深刻理解ReentrantLock
- 8. 深入理解ReentrantLock
- 9. 深刻理解AbstractQueuedSynchronizer
- 10. 從源碼角度完全理解ReentrantLock