簡(jiǎn)介
Java 并發(fā)編程離不開(kāi)鎖， Synchronized 是常用的一種實(shí)現(xiàn)加鎖的方式，使用比較簡(jiǎn)單快捷。在 Java 中還有另一種鎖，即 Lock 鎖。 Lock 是一個(gè)接口，提供了超時(shí)阻塞、可響應(yīng)中斷以及公平非公平鎖等特性，相比于 Synchronized，Lock 功能更強(qiáng)大，可以實(shí)現(xiàn)更靈活的加鎖方式。
Lock 的主要實(shí)現(xiàn)類是 ReentrantLock，而 ReetrantLock 中具體的實(shí)現(xiàn)方式是利用另外一個(gè)類 AbstractQueuedSynchronizer，所有的操作都是委托給這個(gè)類完成。AbstractQueuedSynchronizer 是 Lock 鎖的重要組件，本文從 AbstractQueuedSynchronizer 來(lái)分析 ReetrantLock 的實(shí)現(xiàn)原理。
基本用法
先看一下 Lock 的基本用法：
Lock lock = ...;
lock.lock();
try{
//處理任務(wù)
}catch(Exception ex){
}finally{
lock.unlock(); //釋放鎖
}
lock.lock() 即是加鎖， lock.unolck() 是釋放鎖，為了保證所能夠釋放，unlock() 應(yīng)該放到 finally 中。
下面分別從 lock() 和 unlock() 方法來(lái)分析加鎖和解鎖到底做了什么。
lock
下面是 lock() 的代碼：
public void lock() {
sync.lock();
}
可以看到，只是簡(jiǎn)單調(diào)用了 sync 對(duì)應(yīng)的 lock() 方法。那么這個(gè) sync 是什么呢？其實(shí)這個(gè)就是 AbstractQueuedSynchronizer 的實(shí)現(xiàn)類?？梢钥匆幌?ReentrantLock 的構(gòu)造方法：
/**
* 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 有兩個(gè)方法，主要目的是選擇是公平鎖還是非公平鎖。公平鎖指的是先來(lái)后到，先爭(zhēng)搶鎖的線程先獲得鎖，而非公平鎖則不一定。ReentrantLock 默認(rèn)使用的是非公平鎖，也可以通過(guò)構(gòu)造參數(shù)選擇公平鎖。選擇哪個(gè)鎖其實(shí)是生成了一個(gè)對(duì)象并賦值給變量 sync，下面是涉及到的代碼：
/**
* 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 中有一個(gè)抽象的內(nèi)部類 Sync，繼承于 AbstractQueuedSynchronizer 并實(shí)現(xiàn)了一些方法。另有兩個(gè)類 FairSync 和 NoFairSync 繼承了 Sync，它們自然就是公平鎖以及非公平鎖的實(shí)現(xiàn)。下面分析將從公平鎖出發(fā)，非公平鎖與公平鎖差別并不是很多。
公平鎖 FairSync 加鎖的代碼如下：
final void lock() {
acquire(1);
}
只有一行，調(diào)用了 acquire，這是 AbstractQueuedSynchronizer 中的一個(gè)方法，代碼如下：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
acquire 的實(shí)現(xiàn)也很短，不多在其中卻包含了加鎖的具體實(shí)現(xiàn)，關(guān)鍵就在內(nèi)部調(diào)用的幾個(gè)方法中。
為了理解加鎖和解鎖的過(guò)程，下面具體介紹一下 AbstractQueuedSynchronizer（以下簡(jiǎn)稱 AQS）。
AbstractQueuedSynchronizer
AQS 中使用一個(gè)同步隊(duì)列來(lái)實(shí)現(xiàn)線程同步狀態(tài)的管理，當(dāng)一個(gè)線程獲取鎖失敗的時(shí)候， AQS將此線程構(gòu)造成一個(gè)節(jié)點(diǎn)(Node)并加入同步隊(duì)列并且阻塞線程。當(dāng)鎖釋放時(shí)，會(huì)從同步隊(duì)列中將第一個(gè)節(jié)點(diǎn)喚醒并使其再次獲取鎖。
同步隊(duì)列中的節(jié)點(diǎn)用來(lái)保存獲取鎖失敗的線程的相關(guān)信息，包含如下屬性：
static final class Node {
/** Marker to indicate a node is waiting in shared mode */
// 標(biāo)識(shí)共享模式
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
// 標(biāo)識(shí)獨(dú)占模式
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. */
// 需要喚醒后繼節(jié)點(diǎn)
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;
// 節(jié)點(diǎn)狀態(tài)，為上面的幾個(gè)狀態(tài)之一
volatile int waitStatus;
// 前置節(jié)點(diǎn)
volatile Node prev;
// 后繼節(jié)點(diǎn)
volatile Node next;
// 節(jié)點(diǎn)所表示的線程
volatile Thread thread;
Node nextWaiter;
...
}
Node 是 AQS 的內(nèi)部類，其中包含一些屬性標(biāo)識(shí)一個(gè)阻塞線程的節(jié)點(diǎn)，包括是獨(dú)占模式還是共享模式、節(jié)點(diǎn)的狀態(tài)、前驅(qū)結(jié)點(diǎn)、后繼結(jié)點(diǎn)以及節(jié)點(diǎn)所代表的線程。
同步隊(duì)列是一個(gè)雙向列表，在 AQS 中有這樣幾個(gè)屬性：
/**
* 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.
*/
// 頭結(jié)點(diǎn)
private transient volatile Node head;
/**
* Tail of the wait queue, lazily initialized. Modified only via
* method enq to add new wait node.
*/
// 尾節(jié)點(diǎn)
private transient volatile Node tail;
/**
* The synchronization state.
*/
// 鎖的狀態(tài)
private volatile int state;
其中，head 和 tail 分別指向同步隊(duì)列的頭結(jié)點(diǎn)和尾節(jié)點(diǎn)，state 標(biāo)識(shí)鎖當(dāng)前的狀態(tài)，為 0 時(shí)表示當(dāng)前鎖未被占用，大于 1 表示被占用，之所以是大于 1 是因?yàn)殒i可以重入，每重入一次增加 1。同步隊(duì)列的結(jié)構(gòu)大致如下圖：
了解了同步隊(duì)列后，下面具體看看加鎖和解鎖的過(guò)程。
加鎖
final void lock() {
acquire(1);
}
public final void acquire(int arg) {
// 加鎖的主要代碼
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
主要邏輯其實(shí)就是一行代碼：if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))，tryAcquire 是嘗試獲取一下鎖，為什么說(shuō)是嘗試呢？看代碼：
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState(); // 獲取當(dāng)前狀態(tài)
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 可以分為三條分支：
當(dāng)前鎖未被占用（
getState() == 0），則判斷是否有前驅(qū)結(jié)點(diǎn)，沒(méi)有的話就用 CAS 加鎖(compareAndSetState(0, acquires))，加鎖成功則調(diào)用setExclusiveOwnerThread(current)標(biāo)示一下并返回true。當(dāng)前線程已經(jīng)獲取過(guò)這個(gè)鎖，則此時(shí)是重入，改變
state的計(jì)數(shù)即可，返回true表示加鎖成功。如果不是上面兩種情況，那么說(shuō)明鎖被占用或者 CAS 沒(méi)有搶過(guò)其它線程，則需要進(jìn)入同步隊(duì)列，返回
false表示嘗試加鎖失敗。
回到 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 這一行，如果 tryAcquire(arg) 返回 false 將會(huì)執(zhí)行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)。先看一下 addWaiter(Node.EXCLUSIVE) ，這個(gè)方法的代碼如下所示：
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;
// 尾節(jié)點(diǎn)為空
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
其中主要邏輯是將當(dāng)前線程包裝為一個(gè) Node 節(jié)點(diǎn)并加入同步隊(duì)列。如果尾節(jié)點(diǎn)為空，則用 CAS 設(shè)置尾節(jié)點(diǎn)，如果入隊(duì)失敗則調(diào)用 enq(node)，這個(gè)方法內(nèi)部是一個(gè)循環(huán)，利用自旋 CAS 把節(jié)點(diǎn)加入同步隊(duì)列，具體代碼就不分析了。
在節(jié)點(diǎn)加入隊(duì)列之后，執(zhí)行的是 acquireQueued 方法，代碼如下：
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
// 這是一個(gè)無(wú)限循環(huán)
for (;;) {
final Node p = node.predecessor();
// 如果前驅(qū)節(jié)點(diǎn)是 head，則嘗試獲取鎖
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 獲取鎖失敗則進(jìn)入判斷是否要進(jìn)入睡眠
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
acquireQueued 實(shí)現(xiàn)了線程的睡眠與喚醒。在內(nèi)部是一個(gè)無(wú)限循環(huán)，每次獲取前驅(qū)節(jié)點(diǎn)，如果前驅(qū)結(jié)點(diǎn)是 HEAD，那么嘗試去獲取鎖，獲取成功則將此節(jié)點(diǎn)變?yōu)樾碌念^結(jié)點(diǎn)并將原先的頭結(jié)點(diǎn)出隊(duì)。如果前驅(qū)節(jié)點(diǎn)不是頭結(jié)點(diǎn)或者獲取鎖失敗，那么就會(huì)進(jìn)入 shouldParkAfterFailedAcquire 方法，判斷是否進(jìn)入睡眠，如果這個(gè)方法返回 true，則調(diào)用 parkAndCheckInterrupt 讓線程進(jìn)入睡眠狀態(tài)。下面是 parkAndCheckInterrupt 的代碼：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 線程在這一步進(jìn)入阻塞狀態(tài)
return Thread.interrupted();
}
對(duì)于 shouldParkAfterFailedAcquire 來(lái)說(shuō)，如果前驅(qū)節(jié)點(diǎn)正常，那么會(huì)返回 true，表示當(dāng)前線程應(yīng)該掛起，如果前驅(qū)結(jié)點(diǎn)取消了排隊(duì)，那么當(dāng)前線程有機(jī)會(huì)搶鎖，此時(shí)返回 false，并繼續(xù) acquireQueued 中的循環(huán)。
解鎖
相比于加鎖，解鎖稍微簡(jiǎn)單一點(diǎn)，看一下 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)
// 喚醒后繼節(jié)點(diǎn)
unparkSuccessor(h);
return true;
}
return false;
}
首先調(diào)用 tryRelease 解鎖，如果解鎖成功則喚醒后繼結(jié)點(diǎn)，返回值表示是否成功釋放鎖。那為什么會(huì)解鎖不成功，其實(shí)是因?yàn)橹厝?，看一?tryRelease 的代碼：
protected final boolean tryRelease(int releases) {
// 更新 state 計(jì)數(shù)值
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 減去對(duì)應(yīng)的值，如果 state == 0，那么說(shuō)明鎖已經(jīng)完全釋放。
在 release 中，如果鎖已經(jīng)完全釋放，那么將調(diào)用 unparkSuccessor 喚醒后繼節(jié)點(diǎn)，喚醒的節(jié)點(diǎn)所代表的線程阻塞在 parkAndCheckInterrupt 中：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 線程在這一步進(jìn)入阻塞狀態(tài)
return Thread.interrupted();
}
線程被喚醒后，將繼續(xù) acquireQueued 中的循環(huán)，嘗試獲取鎖。
總結(jié)
本文簡(jiǎn)要分析了 Lock 鎖的原理，主要是利用 AbstractQueuedSynchronizer這個(gè)關(guān)鍵的類。AQS 的核心在于使用 CAS 更新鎖的狀態(tài)，并利用一個(gè)同步隊(duì)列將獲取鎖失敗的線程進(jìn)行排隊(duì)，當(dāng)前驅(qū)節(jié)點(diǎn)解鎖后再喚醒后繼節(jié)點(diǎn)，是一個(gè)幾乎純 Java 實(shí)現(xiàn)的加鎖與解鎖。
