共享鎖 和 AQS

1. 基于 AQS 實現的鎖有哪些？

 

AQS（AbstractQueuedSynchronizer），是 Java 并發包中非常重要的一個類，大部分鎖的實現也是基于 AQS 實現的，包括：

• ReentrantLock，可重入鎖。這個是我們最開始介紹的鎖，也是最常用的鎖。通常會與 synchronized 做比較使用。
• ReentrantReadWriteLock，讀寫鎖。讀鎖是共享鎖、寫鎖是獨占鎖。
• Semaphore，信號量鎖。主要用于控制流量，比如：數據庫連接池給你分配10個鏈接，那么讓你來一個連一個，連到10個還沒有人釋放，那你就等等。
• CountDownLatch，閉鎖。Latch 門閂的意思，比如：說四個人一個漂流艇，坐滿了就推下水。

這一章節我們主要來介紹 Semaphore ，信號量鎖的實現，其實也就是介紹一個關于共享鎖的使用和源碼分析。

2. Semaphore 共享鎖使用

Semaphore semaphore = new Semaphore(2, false); // 構造函數入參，permits：信號量、fair：公平鎖/非公平鎖
for (int i = 0; i < 8; i++) {
    new Thread(() -> {
        try {
            semaphore.acquire();
            System.out.println(Thread.currentThread().getName() + "蹲坑");
            Thread.sleep(1000L);
        } catch (InterruptedException ignore) {
        } finally {
            semaphore.release();
        }
    }, "蹲坑編號：" + i).start();
}

這里我們模擬了一個在高速服務區，廁所排隊蹲坑的場景。由于坑位有限，為了避免造成擁擠和踩踏，保安人員在門口攔著，感覺差不多，一次釋放兩個進去，一直到都釋放。你也可以想成早上坐地鐵上班，或者旺季去公園，都是一批一批的放行

「測試結果」

蹲坑編號：0蹲坑
蹲坑編號：1蹲坑

蹲坑編號：2蹲坑
蹲坑編號：3蹲坑

蹲坑編號：4蹲坑
蹲坑編號：5蹲坑

蹲坑編號：6蹲坑
蹲坑編號：7蹲坑

Process finished with exit code 0

• Semaphore 的構造函數可以傳遞是公平鎖還是非公平鎖，最終的測試結果也不同，可以自行嘗試。
• 測試運行時，會先輸出     0坑、1坑，     之后2坑、3坑...，每次都是這樣兩個，兩個的釋放。這就是 Semaphore 信號量鎖的作用。 

3. Semaphore 源碼分析

3.1 構造函數

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

permits：n. 許可證，特許證(尤指限期的)

默認情況下只需要傳入 permits 許可證數量即可，也就是一次允許放行幾個線程。構造函數會創建非公平鎖。如果你需要使用 Semaphore 共享鎖中的公平鎖，那么可以傳入第二個構造函數的參數 fair = false/true。true：FairSync，公平鎖。在我們前面的章節已經介紹了公平鎖相關內容和實現，以及CLH、MCS 《公平鎖介紹》

「初始許可證數量」

FairSync/NonfairSync(int permits) {
    super(permits);
}

Sync(int permits) {
    setState(permits);
}

protected final void setState(int newState) {
    state = newState;
} 

在構造函數初始化的時候，無論是公平鎖還是非公平鎖，都會設置 AQS 中 state 數量值。這個值也就是為了下文中可以獲取的信號量扣減和增加的值。 

3.2 acquire 獲取信號量

• 其實獲取信號量的這四個方法，主要就是，一次獲取幾個和是否響應中斷的組合。
• semaphore.acquire()，源碼中實際調用的方法是，     sync.acquireSharedInterruptibly(1)。也就是相應中斷，一次只占一個坑。
• semaphore.acquire(2)，同理這個就是一次要占兩個名額，也就是許可證。     生活中的場景就是我給我朋友排的對，她來了，進來吧。 

3.3 acquire 釋放信號量

有獲取就得有釋放，獲取了幾個信號量就要釋放幾個信號量。當然你可以嘗試一下，獲取信號量 semaphore.acquire(2) 兩個，釋放信號量 semaphore.release(1)，看看運行效果 

3.4 公平鎖實現

「信號量獲取過程」，一直到公平鎖實現。semaphore.acquire -> sync.acquireSharedInterruptibly(permits) -> tryAcquireShared(arg)

semaphore.acquire(1);

public void acquire(int permits) throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    sync.acquireSharedInterruptibly(permits);
}

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

「FairSync.tryAcquireShared」

protected int tryAcquireShared(int acquires) {
    for (;;) {
        if (hasQueuedPredecessors())
            return -1;
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

• hasQueuedPredecessors，公平鎖的主要實現邏輯都在于這個方法的使用。它的目的就是判斷有線程排在自己前面沒，以及把線程添加到隊列中的邏輯實現。     在前面我們介紹過CLH等實現，可以往前一章節閱讀
• for (;;)，是一個自旋的過程，通過 CAS 來設置 state 偏移量對應值。這樣就可以避免多線程下競爭獲取信號量沖突。
• getState()，在構造函數中已經初始化 state 值，在這里獲取信號量時就是使用 CAS 不斷的扣減。
• 另外需要注意，共享鎖和獨占鎖在這里是有區別的，獨占鎖直接返回true/false，共享鎖返回的是int值。
• 如果該值小于0，則當前線程獲取共享鎖失敗。
• 如果該值大于0，則當前線程獲取共享鎖成功，并且接下來其他線程嘗試獲取共享鎖的行為很可能成功。
• 如果該值等于0，則當前線程獲取共享鎖成功，但是接下來其他線程嘗試獲取共享鎖的行為會失敗。

3.5 非公平鎖實現

「NonfairSync.nonfairTryAcquireShared」

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

• 有了公平鎖的實現，非公平鎖的理解就比較簡單了，只是拿去了     if (hasQueuedPredecessors()) 的判斷操作。
• 其他的邏輯實現都和公平鎖一致。

3.6 獲取信號量失敗，加入同步等待隊列

在公平鎖和非公平鎖的實現中，我們已經看到正常獲取信號量的邏輯。那么如果此時不能正常獲取信號量呢？其實這部分線程就需要加入到同步隊列。

「doAcquireSharedInterruptibly」

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

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);
    }
}

• 首先     doAcquireSharedInterruptibly 方法來自 AQS 的內部方法，與我們在學習競爭鎖時有部分知識點相同，但也有一些差異。比如：     addWaiter(Node.SHARED)，     tryAcquireShared，我們主要介紹下這內容。
• Node.SHARED，其實沒有特殊含義，它只是一個標記作用，用于判斷是否共享。     final boolean isShared() { return nextWaiter == SHARED; }
• tryAcquireShared，主要是來自     Semaphore 共享鎖中公平鎖和非公平鎖的實現。用來獲取同步狀態。
• setHeadAndPropagate(node, r)，如果r > 0，同步成功后則將當前線程結點設置為頭結點，同時 helpGC，p.next = null，斷鏈操作。
• shouldParkAfterFailedAcquire(p, node)，調整同步隊列中 node 結點的狀態，并判斷是否應該被掛起。這在我們之前關于鎖的文章中已經介紹。
• parkAndCheckInterrupt()，判斷是否需要被中斷，如果中斷直接拋出異常，當前結點請求也就結束。
• cancelAcquire(node)，取消該節點的線程請求。

4. CountDownLatch 共享鎖使用

CountDownLatch 也是共享鎖的一種類型，之所以在這里體現下，是因為它和 Semaphore 共享鎖，既相似有不同。

CountDownLatch 更多體現的組團一波的思想，同樣是控制人數，但是需要夠一窩。比如：我們說過的4個人一起上皮劃艇、兩個人一起上蹺蹺板、2個人一起蹲坑我沒見過，這樣的方式就是門閂 CountDownLatch 鎖的思想。

public static void main(String[] args) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(10);
    ExecutorService exec = Executors.newFixedThreadPool(10);
    for (int i = 0; i < 10; i++) {
        exec.execute(() -> {
            try {
                int millis = new Random().nextInt(10000);
                System.out.println("等待游客上船，耗時：" + millis + "(millis)");
                Thread.sleep(millis);
            } catch (Exception ignore) {
            } finally {
                latch.countDown(); // 完事一個扣減一個名額
            }
        });
    }
    // 等待游客
    latch.await();
    System.out.println("船長急躁了，開船!");
    // 關閉線程池
    exec.shutdown();
}

• 這一個公園游船的場景案例，等待10個乘客上傳，他們比較墨跡。
• 上一個扣減一個     latch.countDown()
• 等待游客都上船     latch.await()
• 最后船長開船！！     急躁了

「測試結果」

等待游客上船，耗時：6689(millis)
等待游客上船，耗時：2303(millis)
等待游客上船，耗時：8208(millis)
等待游客上船，耗時：435(millis)
等待游客上船，耗時：9489(millis)
等待游客上船，耗時：4937(millis)
等待游客上船，耗時：2771(millis)
等待游客上船，耗時：4823(millis)
等待游客上船，耗時：1989(millis)
等待游客上船，耗時：8506(millis)
船長急躁了，開船!

Process finished with exit code 0

• 在你實際的測試中會發現，     船長急躁了，開船!，會需要等待一段時間。
• 這里體現的就是門閂的思想，組隊、一波帶走。
• CountDownLatch 的實現與 Semaphore  基本相同、細節略有差異，就不再做源碼分析了。