本文部分摘自《Java 并發編程的藝術》
概述
佇列同步器 AbstractQueuedSynchronize（以下簡稱同步器），是用來構建鎖（Lock）或者其他同步組件（JUC 并發包）的基礎框架，它使用了一個 int 成員變數表示同步狀態，通過內置的 FIFO 佇列來完成資源獲取執行緒的排隊作業
同步器的主要使用方式是繼承，子類通過繼承同步器并實作它的抽象方法來管理同步狀態，子類推薦被定義為自定義同步組件的靜態內部類，同步器自身沒有實作任何同步介面，它僅僅是定義了若干同步狀態的獲取和釋放方法來供自定義組件使用
一言以蔽之，同步器是實作鎖（也可以是任意同步組件）的一種方式，它屏蔽了更加底層的一些機制，使開發者更易于理解和使用
佇列同步器的介面
同步器的設計是基于模板方法模式的，使用者需要繼承佇列同步器并重寫指定的方法，隨后將同步器組合在自定義同步組件的實作中，并呼叫同步器提供的模板方法，而這些模板方法將會呼叫使用者重寫的方法
1. 訪問或修改同步狀態
重寫同步器指定的方法時，需要使用同步器提供的如下三個方法來訪問或修改同步狀態：
-
getState()
獲取當前同步狀態
-
setState(int newState)
設定當前同步狀態
-
compareAndSetState(int expect, int update)
使用 CAS 設定當前狀態，該方法能保證狀態設定的原子性
2. 同步器可重寫的方法
| 方法名稱 | 描述 |
|---|---|
| protected boolean tryAcquire(int arg) | 獨占式獲取同步狀態，實作該方法需要查詢當前狀態，并判斷同步狀態是否符合預期，然后再進行 CAS 設定同步狀態 |
| protected boolean tryRelease(int arg) | 獨占式地釋放同步狀態，等待獲取同步狀態的執行緒將有機會獲取同步狀態 |
| protected int tryAcquireShared(int arg) | 共享式獲取同步狀態，回傳大于等于 0 的值，表示獲取成功，否則獲取失敗 |
| protected boolean tryReleaseShared(int arg) | 共享式釋放同步狀態 |
| protected boolean isHeldExclusively() | 當前同步器是否在獨占模式下被執行緒占有，一般該方法表示是否被當前執行緒所獨占 |
3. 同步器提供的模板方法
| 方法名稱 | 描述 |
|---|---|
| void acquire(int arg) | 獨占式獲取同步狀態，如果當前執行緒獲取同步狀態成功，則由該方法回傳，否則，將會進入同步佇列等待，該方法將會呼叫重寫的 tryAcquire(int arg) 方法 |
| void acquireInterruptibly(int arg) | 與 acquire(int arg) 相同，但該方法回應中斷，當前執行緒未獲取到同步狀態而進入同步佇列中，如果當前執行緒被中斷，則該方法會拋出 InterruptedException 并回傳 |
| boolean tryAcquireNanos(int arg, long nanos) | 在 acquireInterruptibly(int arg) 的基礎上增加了超時限制 |
| void acquireShared(int arg) | 共享式的獲取同步狀態，與獨占式獲取的主要區別是在同一時刻可以有多個執行緒獲取到同步狀態 |
| void acquireSharedInterruptibly(int arg) | 與 acquireShared(int arg) 相同，該方法回應中斷 |
| boolean tryAcquireSharedNanos(int arg, long nanos) | 在 acquireSharedInterruptibly 的基礎上增加了超時限制 |
| boolean release(int arg) | 獨占式的釋放同步狀態，該方法會在釋放同步狀態之后，將同步佇列中第一個節點包含的執行緒喚醒 |
| boolean releaseShared(int arg) | 共享式的釋放同步狀態 |
| Collection<Thread> getQueuedThreads() | 獲取等待在同步佇列上的執行緒集合 |
4. 示例
下面通過一個獨占鎖的示例來深入了解一下同步器的作業原理，顧名思義，獨占鎖就是在同一時刻只能有一個執行緒獲取到鎖，其他獲取鎖的執行緒只能處于同步佇列中等待，只有獲取鎖的執行緒釋放了鎖，后繼的執行緒才能獲取鎖
public class Mutex implements Lock {
/**
* 自定義同步器
*/
private static class Sync extends AbstractQueuedSynchronizer {
@Override
protected boolean isHeldExclusively() {
// 是否處于占用狀態
return getState() == 1;
}
@Override
public boolean tryAcquire(int acquires) {
// 當狀態為 0 時獲取鎖
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
@Override
protected boolean tryRelease(int releases) {
// 釋放鎖，將狀態設定為 0
if (getState() == 0) {
throw new IllegalMonitorStateException();
}
setExclusiveOwnerThread(null);
setState(0);
return true;
}
/**
* 回傳一個 Condition, 每個 condition 都包含一個 condition 佇列
*/
Condition newCondition() {
return new ConditionObject();
}
}
private final Sync sync = new Sync();
@Override
public void lock() {
sync.acquire(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
@Override
public boolean tryLock() {
return sync.tryAcquire(1);
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(time));
}
@Override
public void unlock() {
sync.release(1);
}
@Override
public Condition newCondition() {
return sync.newCondition();
}
}
Mutex 中定義了一個靜態內部類，該內部類繼承了同步器并實作了獨占式獲取和釋放同步狀態，用戶使用 Mutex 時并不會直接和內部同步器實作打交道，而是呼叫 Mutex 提供的方法，大大降低了實作一個可靠自定義組件的門檻
佇列同步器的實作
1. 同步佇列
同步器依賴內部的同步雙向佇列來完成同步狀態的管理，當前執行緒獲取同步狀態失敗后，同步器會將當前執行緒及其等待狀態等資訊構造成一個節點，并加入同步佇列，同時阻塞當前執行緒，當同步狀態釋放后，會把首節點中的執行緒喚醒，使其再次嘗試獲取同步狀態
節點是構成同步佇列的基礎，同步器擁有首節點（head）和尾結點（tail），沒有成功獲取同步狀態的執行緒將會成為節點并加入該佇列的尾部
同步佇列的基本結構如下：
同步器將節點加入到同步佇列的程序如圖所示：
首節點是獲取同步狀態成功的節點，首節點執行緒在釋放同步狀態時，會喚醒后繼節點，而后繼節點將會在獲取同步狀態成功時將自己設定為首節點，程序如下：
設定首節點是通過獲取同步狀態成功的執行緒來完成的，由于只有一個執行緒能夠成功獲取同步狀態，因此設定頭節點的方法并不需要使用 CAS 來保證，只需要將首節點設定成原首節點的后繼節點并斷開原首節點的 next 參考即可
2. 獨占式同步狀態獲取與釋放
通過呼叫同步器的 acquire(int arg) 方法可以獲取同步狀態，該方法對中斷不敏感，執行緒獲取同步狀態失敗則進入同步佇列中，后續對執行緒進行中斷操作，執行緒不會從同步佇列中移出
獨占式同步狀態獲取流程，也就是 acquire(int arg) 方法呼叫流程如圖所示：
如果當前執行緒獲取同步狀態失敗，就會生成一個節點（獨占式 Node.EXCLUSIVE，同一時刻只能有一個執行緒成功獲取同步狀態），并加入到佇列尾部，一個佇列里有很多節點，而只有前驅節點是頭節點的節點才能嘗試獲取同步狀態，原因有兩個：
- 頭節點是成功獲取到同步狀態的節點，而頭節點的執行緒釋放了同步狀態之后，將會喚醒其后繼節點，后繼節點的執行緒被喚醒后需要檢查自己的前驅節點是否是頭節點
- 維護同步佇列的 FIFO 原則
因此，如果佇列中的非頭節點執行緒的前驅節點出隊或者被中斷而從等待狀態回傳，那么其隨后會檢查自己的前驅是否為頭節點，如果是則嘗試獲取同步狀態
當前執行緒獲取同步狀態并執行了相應邏輯之后，就需要釋放同步狀態，使得后繼節點能夠繼續獲取同步狀態，通過呼叫同步器的 release(int arg) 方法可以釋放同步狀態，該方法執行時，會喚醒頭節點的后繼節點執行緒
3. 共享式同步狀態獲取與釋放
共享式獲取與獨占式獲取最主要的區別在于同一時刻能否有多個執行緒同時獲取到同步狀態，以檔案的讀寫為例，若一個程式在對檔案進行讀操作，那么這一時刻對于該檔案的寫操作均被阻塞，而讀操作能夠同時進行，寫操作要求對資源的獨占式訪問，而讀操作可以是共享式訪問，兩種不同的訪問模式在同一時刻對檔案或資源的訪問情況，如下圖所示：
通過呼叫同步器的 acquireShared(int arg) 方法可以共享式地獲取同步狀態，其代碼核心邏輯和 acquire() 差不多，也是判斷當前節點的前驅是否為頭節點，如果是就嘗試獲取同步狀態，頭節點在釋放同步狀態之后，也會喚醒后續處于等待狀態的節點
問題的關鍵在于如何做到多個執行緒訪問同步狀態，因為按照上面所講的程序，和獨占式幾乎沒有任何區別，獨占式與共享式在實作上的差別其實僅僅在于：每次頭節點釋放同步狀態之后，獨占式只是把其后繼節點設定為頭節點，而共享式還多了一個傳播的程序（筆者能力有限，這一塊沒搞明白，就不瞎寫了，，）
與獨占式一樣，共享式獲取也需要釋放同步狀態，通過呼叫 releaseShared(int arg) 方法可以釋放同步狀態，并喚醒后續處于等待狀態的節點
4. 獨占式超時獲取同步狀態
通過呼叫同步器的 doAcquireNanos(int arg, long nanosTimeout) 方法可以超時獲取同步狀態，即在指定的時間段內獲取同步狀態
在介紹這個方法之前，先介紹一下回應中斷的同步狀態獲取程序，Java5 以后，同步器提供了 acquireInterruptibly(int arg) 方法，這個方法在等待獲取同步狀態時，如果當前執行緒被中斷，會立刻回傳，并拋出 InterruptedException
超時獲取同步狀態可以視為回應中斷獲取同步狀態的增強版，獨占式超時和非獨占式獲取在流程上非常相似，其主要區別在于未獲取到同步狀態時的處理邏輯，acquire(int arg) 在未獲取到同步狀態時，會使當前執行緒一致處于等待狀態，而 doAcquireNanos(int arg, long nanosTimeout) 會使當前執行緒等待 nanosTimeout 納秒，如果當前執行緒在 nanosTimeout 納秒內沒有獲取同步狀態，將會從等待邏輯中自動回傳
自定義同步組件
設計一個同步工具：同一時刻，只能允許至多兩個執行緒同時訪問，超過兩個執行緒的訪問將被阻塞，顯然這是共享式訪問，主要設計思路如下：
- 重寫 tryAcquireShared(int args) 方法和 tryReleaseShared(int args) 方法
- 定義初始狀態 status 為 2，當一個執行緒進行獲取，status 減 1，該執行緒釋放，status 加 1，為 0 時再有其他執行緒進行獲取，則阻塞
示例代碼如下：
public class TwinsLock implements Lock {
private final Sync sync = new Sync(2);
private static final class Sync extends AbstractQueuedSynchronizer {
Sync(int count) {
if (count <= 0) {
throw new IllegalArgumentException("count must large than zero");
}
setState(count);
}
@Override
public int tryAcquireShared(int reduceCount) {
while (true) {
int current = getState();
int newCount = current - reduceCount;
if (newCount < 0 || compareAndSetState(current, newCount)) {
return newCount;
}
}
}
@Override
protected boolean tryReleaseShared(int reduceCount) {
while (true) {
int current = getState();
int newCount = current + reduceCount;
if (compareAndSetState(current, newCount)) {
return true;
}
}
}
Condition newCondition() {
return new ConditionObject();
}
}
@Override
public void lock() {
sync.acquireShared(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
@Override
public boolean tryLock() {
return sync.tryAcquireShared(1) > 0;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(time));
}
@Override
public void unlock() {
sync.releaseShared(1);
}
@Override
public Condition newCondition() {
return sync.newCondition();
}
}
再撰寫一個測驗來驗證 TwinsLock 是否按預期作業
public class TwinsLockTest {
public static void main(String[] args) {
final Lock lock = new TwinsLock();
class Worker extends Thread {
@Override
public void run() {
while (true) {
lock.lock();
try {
SleepUtils.second(1);
System.out.println(Thread.currentThread().getName());
SleepUtils.second(1);
} finally {
lock.unlock();
}
}
}
}
for (int i = 0; i < 10; i++) {
Worker worker = new Worker();
worker.setDaemon(true);
worker.start();
}
for (int i = 0; i < 10; i++) {
SleepUtils.second(1);
System.out.println();
}
}
}
運行該測驗用例，發現執行緒名稱成對輸出，說明同一時刻只有兩個執行緒能夠獲取到鎖
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/265055.html
標籤：其他