Java并發: 鎖和同步

在Java并發: 面臨的挑戰那一篇中我們提到鎖和同步是實現并發安全(可見性/原子性)的方法之一。這一章我們來講講Java中的鎖和同步的各種工具，包括:

LockSupport
AbstractQueuedSynchronizer
Java內置的鎖實現

1. LockSupport

LockSupport是基于Unsafe的park/unpark實現的，用來支持線程的掛起和喚醒。

1.1 工作原理

可以理解為線程上有一個0/1標志位，park/unpark基于這個標志位工作的，使用這個模型我們能比較容易理解它的工作模式

park()調用，檢查標志位，標志位=0掛起當前線程，直到標志位被置1，或被中斷/超時；如果標志位=1，將標志位置0，從park方法返回，執行后續代碼
unpark()調用，作用是將標志位置1

unpark()可以在park()之前被調用，已經被unpark()調用過的線程，調用park()時標志位=1，會直接返回而不阻塞。工具方法，sleep休眠指定毫秒數，println打印小時時間戳。

Thread t = new Thread(() -> {println("before sleep");sleep(2000);println("after sleep, going to park");LockSupport.park();println("after park");
});
t.start();
println("before unpark");
LockSupport.unpark(t);
println("after unpark");
t.join();

我們在線程t啟動后立刻進行了unpark，而此時線程t應該還在sleep中，sleep結束后的park調用是瞬時返回的

關于unpark還有兩個點是需要特別注意的

線程Thread t在t.start()調用之前，調用LockSupport.unpark(t)不會做標志位置位，相當于是無效調用
對同一個線程t連續兩次調用LockSupport.unpark(t)，標志位仍然只是置1，只能喚醒一個LockSupport.park()調用

1.2 虛假喚醒

LockSupport.park()的喚醒可能是因為調用了LockSupport.unpark()，也可能是因為線程中斷、park超時，一般的做法是在檢查park條件時做一個循環。我們來看個常見的示例

public void lock() {while (condition) {LockSupport.park(this);}}

即使park()是因為中斷而退出的，程序也能重新進入條件校驗，重新掛起，從而避免虛假喚醒導致問題。想想鎖和條件wait的寫法，是不是和這個如出一轍呢？

1.3 應用案例

LockSupport的文檔上提供了一個最簡單的鎖的案例，FIFOMutex，按調用順序依次把加鎖的機會給每一個調用者，代碼如下

class FIFOMutex {private final AtomicBoolean locked = new AtomicBoolean(false);private final Queue<Thread> waiters = new ConcurrentLinkedQueue<>();public void lock() {boolean wasInterrupted = false;// 將想要加鎖的線程進隊列waiters.add(Thread.currentThread());// 出隊列的第一個線程外，全部掛起；第一個線程，嘗試加鎖，CAS設置locked=truewhile (waiters.peek() != Thread.currentThread() || !locked.compareAndSet(false, true)) {LockSupport.park(this);if (Thread.interrupted()) // 如果線程被中斷了，用wasInterrupted保留中斷的狀態wasInterrupted = true;}waiters.remove(); // 加鎖成功的線程從隊列移除if (wasInterrupted)Thread.currentThread().interrupt();}public void unlock() {locked.set(false); // 釋放鎖LockSupport.unpark(waiters.peek()); // 恢復等待鎖的第一個線程}static {// Reduce the risk of "lost unpark" due to classloadingClass<?> ensureLoaded = LockSupport.class;}
}

在3. 使用Unsafe里我們有寫過一個CrashIntegerID在無鎖的情況下生成自增ID，會導致ID重復，限制我們用這個自定義的FIFOMutex進行競態條件保護，修改后代碼如下

public class CrashIntegerID implements ID {private int id;private FIFOMutex mutex;public CrashIntegerID(FIFOMutex lock, int start) {this.id = start;this.mutex = lock;}public int incrementAndGet() {mutex.lock();try {return id++;} finally {mutex.unlock();}}
}

將控制臺輸出用shell命令統計，可以發現生成10w次后，最大ID是10_0000，ID沒有重復的了，說明我們FIFOMutext是生效的。

randy@Randy:~$ cat num | egrep -v '^$' | sort -n | tail -599996
99997
99998
99999
100000
randy@Randy:~$ cat num | egrep -v '^$' | sort -n | uniq -d

2.AbstractQueuedSynchronized

前面我們通過LockSupport實現了一個簡單的獨占鎖FIFOMutex，但是功能比較簡易。Java內部通過了一個類似的實現，只需要覆寫少數方法就能創建一個功能強大的鎖，AbstractQueueSynchronizer

類似于FIFOMutex，AQS也維護了一個內部狀態state，將等待鎖的線程通過一個CLH隊列保存，額外提供ConditionObject對象，支持基于條件的等待還喚醒，同時它還支持共享鎖。JDK內部大量的鎖和同步器都是基于AQS實現的，比如ReentrantLock、Semaphore等等。

2.1 如何使用

要想基于AQS實現同步器和鎖，只需通過AQS提供的getState()、setState(int)、compareAndSetState(int,int)覆寫AQS中的5個方法。根據先要實現的鎖不同state有不同的含義、不同的值，假設要實現一個非可重入鎖，我們可以假定state=0時鎖已經被其他線程持有，state=1表示鎖限制沒有被持有；假設要實現一個類似Semaphore的同步器，state就用來表示可用的信號量。

方法	說明
boolean tryAcquire(int n)	申請n個獨占資源，返回true表示申請成功，false表示申請失敗
boolean tryRelease(int n)	釋放n給獨占資源，返回true表示釋放成功，false表示釋放失敗
int tryAcquireShared(int n)	申請n個共享資源，返回true表示申請成功，false表示申請失敗
boolean tryReleaseShared(int n)	釋放n給共享資源，返回true表示釋放成功，false表示釋放失敗
boolean isHeldExclusively()	根據state判斷是否獨占鎖，如果是獨占式的，鎖持有期間AQS不會調度鎖的等待隊列的節點來嘗試加鎖

要讓AQS正常且高效的工作，覆寫這5個方法必須是線程安全的，且不應該有長時間的阻塞。此外AQS還繼承了AbstractOwnableSynchronizer，支持在同步器上繼續當前持有鎖的線程，這樣我們能做線程的監控和分析工具能查看，方便定位問題。

2.2 源碼解析

鎖的使用中核心的邏輯就4個，鎖的申請和釋放，條件的等待和喚醒，接下來我們重點看一下這4段的邏輯實現。為了方便理解，對源碼做過編輯，核心邏輯是接近的。

1. 申請鎖

首先是鎖的申請，AQS是通過acquire(n)方法申請鎖，調用后會一直初始當前線程，除非加鎖成功。acquire的第一層邏輯很簡單，嘗試通過tryAcquire申請資源，申請成功直接就算加鎖成功

public final void acquire(int arg) {if (!tryAcquire(arg)) {acquire(null, arg, false, false, false, 0L);}
}

如果申請失敗，調用acquire方法，進入一個無限循環，循環的代碼略長，根據代碼的目的，我把它定義為6個操作，分別是

操作1，申請鎖的當前節點不是等待隊列的隊首，清理CLH等待隊列中已經放棄(取消)的節點
操作2，如果是等待隊列對手或沒有前置節點，嘗試加鎖
操作3，如果node是null創建節點
操作4，將node加入到CLH等待隊列
操作5，如果是等待隊列的隊首，還有自旋次數可以用，進行一次自旋
操作6，自旋失敗，升級使用LockSupport掛起線程

我們來看一下acquire(int arg)調用的acquire方法內部的執行過程

一開始node和pred都是null，會先執行操作2，如果加鎖成功直接返回，否則繼續運行
加鎖失敗的話，執行操作3，創建node節點，進入下一輪循環
這個時候node!=null，但是pred依然是null，再次執行操作2，加鎖成功直接返回，否則繼續運行
加鎖失敗的話，執行操作4，將node加入到CLH等待隊列，進入下一輪循環
進入操作1，判斷在等待隊列中的位置
1. 第1個節點，執行操作2
2. 第2個節點，自旋并進入下一輪循環
多次嘗試后，確實無法加鎖的，進入操作6，將線程掛起

在有的多線程編程的文章和書籍中，將這個執行過程描述為鎖升級，把自旋鎖定義為玄而又玄的算法，其實所謂的自旋只是讓CPU執行一個空指令，看是不是能在幾個指令周期后能夠成功加鎖，從而避免因為線程的掛起(park/unpark)導致的線程上下文切換。所謂的鎖升級只是從一開始直接嘗試加鎖，失敗后嘗試自旋，仍然不能成功才進入等待隊列的過程。

final int acquire(Node node, int arg, boolean shared, boolean interruptible, boolean timed, long time) {Thread current = Thread.currentThread();byte spins = 0, postSpins = 0;   // retries upon unpark of first threadboolean interrupted = false, first = false;Node pred = null;                // predecessor of node when enqueuedfor (;;) {// 操作1: 如果node不是第一個節點，有前置節點，前置節點不是head節點，等待前置節點if (!first && (pred = (node == null) ? null : node.prev) != null && !(first = (head == pred))) {if (pred.status < 0) {cleanQueue();           // 如果前置節點是取消狀態的，清除前置節點continue;} else if (pred.prev == null) {Thread.onSpinWait();    // 如果隊列中只有一個前置節點，嘗試自旋等待continue;}}// 操作2: 如果是第一個節點，或沒有前置節點，嘗試加鎖if (first || pred == null) {boolean acquired;try {if (shared)acquired = (tryAcquireShared(arg) >= 0);elseacquired = tryAcquire(arg);} catch (Throwable ex) {cancelAcquire(node, interrupted, false);throw ex;}if (acquired) {if (first) { // 如果第一個節點加鎖成功，刪除waiter對線程的引用，讓head執行第一個節點node.prev = null;head = node;pred.next = null;node.waiter = null;if (shared)signalNextIfShared(node);if (interrupted)current.interrupt();}return 1;}}// 操作3: 如果節點為null，先創建節點if (node == null) {                 // allocate; retry before enqueueif (shared)node = new SharedNode();elsenode = new ExclusiveNode();} // 操作4: 將Node放入到CLH的等待隊列else if (pred == null) {          // try to enqueuenode.waiter = current;Node t = tail;node.setPrevRelaxed(t);         // avoid unnecessary fenceif (t == null)tryInitializeHead();else if (!casTail(t, node))node.setPrevRelaxed(null);  // back outelset.next = node;} // 操作5: 第一個節點，且自旋次數大于0，嘗試自旋else if (first && spins != 0) {--spins;                        // reduce unfairness on rewaitsThread.onSpinWait();} else if (node.status == 0) {node.status = WAITING;          // enable signal and recheck} // 操作6: 自旋失敗，使用LockSupport掛起線程else {long nanos;spins = postSpins = (byte)((postSpins << 1) | 1);if (!timed)LockSupport.park(this);else if ((nanos = time - System.nanoTime()) > 0L)LockSupport.parkNanos(this, nanos);elsebreak;node.clearStatus();if ((interrupted |= Thread.interrupted()) && interruptible)break;}}return cancelAcquire(node, interrupted, interruptible);
}

2. 釋放鎖

相比申請鎖的過程，釋放就極其的簡單了，直接調用tryRelease釋放資源，釋放重構后通過siganalNext通知等待隊列，執行LockSupport.unpark喚醒線程。

public final boolean release(int arg) {if (tryRelease(arg)) {signalNext(head);return true;}return false;
}

3. 條件等待

AQS通過ConditionObject提供條件等待的支持，當我們調用Condition.await()時，程序經歷了4步操作

操作1: 釋放await關聯的鎖對象
操作2: 掛起線程
操作3: 修改節點、線程狀態
操作4: 重新加鎖

之前我們有提到過，一個持有鎖的方法調用，只有在方法執行結束、方法執行異常、或者調用鎖相關的條件等待時才會釋放鎖。這個操作從源碼層面告訴我們為什么條件等待會釋放鎖。

public final void await() throws InterruptedException {ConditionNode node = new ConditionNode();// 操作1: 釋放鎖int savedState = enableWait(node);LockSupport.setCurrentBlocker(this); // for back-compatibility...while (!canReacquire(node)) {...if ((node.status & COND) != 0) { // 操作2: 阻塞線程if (rejected)node.block(); // 內部調用的還是LockSupport.parkelseForkJoinPool.managedBlock(node);} else {Thread.onSpinWait();    // awoke while enqueuing}}// 操作3: 執行到這里，說明線程已經被喚醒LockSupport.setCurrentBlocker(null);node.clearStatus();// 操作4: 重新加鎖acquire(node, savedState, false, false, false, 0L);if (interrupted) {if (cancelled) {unlinkCancelledWaiters(node);throw new InterruptedException();}Thread.currentThread().interrupt();}
}private int enableWait(ConditionNode node) {if (isHeldExclusively()) {node.waiter = Thread.currentThread();...int savedState = getState(); // condition對象上會保存關聯的鎖的資源if (release(savedState))     // await時，會釋放鎖return savedState;}node.status = CANCELLED; // lock not held or inconsistentthrow new IllegalMonitorStateException();
}

2.3 應用案例

如果用AQS重寫1.3中的案例FIFOMutex會比原來簡單的多，我們來看一下重寫后的代碼

public class AQSFIFOMutex {private Sync sync;public AQSFIFOMutex() {sync = new Sync();}public void lock() {sync.acquire(1);}public void unlock() {sync.release(1);}private static class Sync extends AbstractQueuedSynchronizer {@Overrideprotected boolean tryAcquire(int n) {assert getState() == 0;return compareAndSetState(0, 1);}@Overrideprotected boolean tryRelease(int n) {assert getState() == 1;return compareAndSetState(1, 0);}@Overrideprotected boolean isHeldExclusively() {return getState() == 1;}}
}

3. Java自帶的鎖實現

到現在我們已經大概了解鎖的實現原理，后續的章節我們來看看JDK內置的鎖實現類，有什么特點，要如何使用。

3.1 ReentrantLock

首先要看的是ReentranLock，ReentrantLock是一把可重入鎖，它是基于AbstractQueuedSynchronizer實現的。如果一個線程已經持有了鎖，再次調用申請鎖的時候，這個調用不會被阻塞。

1. 接口定義

核心方法定義見下表

方法	說明
void lock()	嘗試加鎖，加鎖成功則返回，否則阻塞等待
void lockInterruptibly()	同lock()方法，但是響應中斷，在lockInterruptibly()執行期間，如果線程被中斷，這個方法拋出InterruptedException
boolean tryLock()	嘗試加鎖但不阻塞，成功返回true，失敗返回false
boolean tryLock(long timeout, TimeUnit unit)	嘗試加鎖，設置超時時間，如果給定時間內沒加鎖成功返回false，否則返回true
void unlock()	釋放鎖

2. 使用案例

ReentrantLock有兩種典型的使用模式，阻塞和非阻塞，不管那種方式都應該把unlock放到finally中以保證unlock會被調用。

ReentrantLock lock = new ReentrantLock();
lock.lock();
try {// 業務代碼
} finally {lock.unlock();
}

如果使用tryLock代碼應該這樣寫

if (lock.tryLock(2000, TimeUnit.MILLISECONDS)) {try {// 業務代碼} finally {lock.unlock();}
}

3.2 ReentrantReadWriteLock

ReentrantReadWriteLock相比ReentrantLock的做了增強，支持讀寫鎖，實現原理是將AQS的state分成了2部分，高16位用于保存共享鎖，低16位用于保存獨占鎖，以這個邏輯實現AQS的tryAcquire、tryAcquireShared。我們來看一個案例，假設有兩個線程，DoRead負責讀數據，DoWrite負責寫數據，我們現在想模擬的是兩類場景

writeLock被持有的時，所有的readLock無法加鎖成功
readLock可以被兩個線程同時持有

為了做到這兩點可觀測，我們定義一個DoWrite，持有writeLock后休眠5s，啟動DoWrite后，等1s在啟動DoRead，為了讓DoWrite先執行并先拿到寫鎖。

public static class DoWrite implements Runnable {private ReentrantReadWriteLock.WriteLock writeLock;public DoWrite(ReentrantReadWriteLock.WriteLock writeLock) {this.writeLock = writeLock;}public void run() {println("before write lock");writeLock.lock();try {println("under write lock , before sleep");sleep(5000);println("under write lock , after sleep");} finally {writeLock.unlock();}println("after write lock");}
}

以下是讀鎖的代碼，以及測試啟動的代碼

public static class DoRead implements Runnable {private ReentrantReadWriteLock.ReadLock readLock;private int identity;public DoRead(int identity, ReentrantReadWriteLock.ReadLock readLock) {this.readLock = readLock;this.identity = identity;}public void run() {println("before read lock , identity: " + identity);readLock.lock();try {println("under read lock, before sleep , identity: " + identity);sleep(3000);println("under read lock, after sleep , identity: " + identity);} finally {readLock.unlock();}println("after read lock , identity: " + identity);}
}
// 測試代碼
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
Thread tw = new Thread(new DoWrite(writeLock));
tw.start();
sleep(1000);
Thread tr1 = new Thread(new DoRead(1, readLock));
tr1.start();
Thread tr2 = new Thread(new DoRead(2, readLock));
tr2.start();tw.join();
tr1.join();
tr2.join();

我們來分析一下輸入的日志，看看程序是按什么順序執行的

3.3 StampedLock

JDK 8開始提供StampedLock，它支持3中鎖模式，比較特別的是它不是可重入鎖，因此在某個線程拿到鎖之后，不能在這個線程內部再次申請鎖

寫鎖writeLock，只在讀寫鎖都沒有被持有的情況下才能申請
讀鎖readLock，只在沒有線程持有寫鎖時才能申請
樂觀讀tryOptimisticRead，讀取鎖的state狀態，假設操作期間不會發生寫鎖

StampedLock的實現思路借鑒了有序讀寫鎖的算法(Ordered RW locks)，感興趣的話可以查看對應的算法描述： Design, verification and applications of a new read-write lock algorithm | Proceedings of the twenty-fourth annual ACM symposium on Parallelism in algorithms and architectures。

按簡化模型來理解的話，調用tryOptimisticRead時會獲取stamp作為版本號，建立本地數據的快照，再驗證版本號，如果版本號未變更則任務數據快照是有效的。我們來看一下下使用流程

獲取stamp版本后，用的是state的值
建立業務數據快照
使用Unsafe.loadFence()建立內存屏障，保證進入第4步之前，業務數據快照已經讀取完成
驗證第1步讀取的stamp版本號，驗證通過說明stamp未被修改，任意的寫鎖會導致stamp被修改，stamp未修改說明期間沒有申請過寫鎖，因此數據未被修改
如果驗證通過，升級為讀鎖，再次執行第2步重新建立數據快照
釋放讀鎖
使用數據快照，執行業務邏輯

通過這個執行步驟，我們可以知道tryOptimisticRead能提升性能的前提是大部分情況下validate(stamp)會成功，即業務是讀多寫少的情況。業務數據快照只是基于內存屏障實現的，執行期間并沒有鎖，所以只能保證快照是某一時刻的數據，但不能保證是當前最新的數據。

下面我們舉個例子來解釋一下StampedLock怎么使用，假設我們有一個Statistic類，用來統計數字的個數、總和，然后提供平均值

public class Statistic {private final StampedLock lock = new StampedLock();private int count;private int total;public void newNum(int num) {long stamp = lock.writeLock(); // 寫鎖try {count++;total += num;} finally {lock.unlock(stamp);}}public double avg() {long stamp = lock.tryOptimisticRead(); // 樂觀讀int tempCount = count, tempTotal = total; // 快照數據if (!lock.validate(stamp)) {stamp = lock.readLock(); // 讀鎖try {tempCount = count;tempTotal = total;} finally {lock.unlock(stamp);}}return tempTotal * 1.0 / tempCount; // 使用快照數據做業務計算}
}