從 synchronized 到 CAS 和 AQS - 完全弄懂 Java 各類併發鎖

時間 2019-11-07

標籤 synchronized cas aqs 完全弄懂 java 各類併發欄目 Java 简体版

原文原文鏈接

概述

Java 中的併發鎖大體分爲隱式鎖和顯式鎖兩種。隱式鎖就是咱們最常使用的 synchronized 關鍵字，顯式鎖主要包含兩個接口：Lock 和 ReadWriteLock，主要實現類分別爲 ReentrantLock 和 ReentrantReadWriteLock，這兩個類都是基於 AQS(AbstractQueuedSynchronizer) 實現的。還有的地方將 CAS 也稱爲一種鎖，在包括 AQS 在內的不少併發相關類中，CAS 都扮演了很重要的角色。html

咱們只須要弄清楚 synchronized 和 AQS 的原理，再去理解併發鎖的性質和侷限就很簡單了。所以這篇文章重點放在原理上，對於使用和特色不會過多涉及。java

概念辨析

下面是關於鎖的一些概念解釋，這些都是一些關於鎖的性質的描述，並不是具體實現。node

悲觀鎖和樂觀鎖

悲觀鎖和獨佔鎖是一個意思，它假設必定會發生衝突，所以獲取到鎖以後會阻塞其餘等待線程。這麼作的好處是簡單安全，可是掛起線程和恢復線程都須要轉入內核態進行，這樣作會帶來很大的性能開銷。悲觀鎖的表明是 synchronized。然而在真實環境中，大部分時候都不會產生衝突。悲觀鎖會形成很大的浪費。而樂觀鎖不同，它假設不會產生衝突，先去嘗試執行某項操做，失敗了再進行其餘處理（通常都是不斷循環重試）。這種鎖不會阻塞其餘的線程，也不涉及上下文切換，性能開銷小。表明實現是 CAS。segmentfault

公平鎖和非公平鎖

公平鎖是指各個線程在加鎖前先檢查有無排隊的線程，按排隊順序去得到鎖。非公平鎖是指線程加鎖前不考慮排隊問題，直接嘗試獲取鎖，獲取不到再去隊尾排隊。值得注意的是，在 AQS 的實現中，一旦線程進入排隊隊列，即便是非公平鎖，線程也得乖乖排隊。緩存

可重入鎖和不可重入鎖

若是一個線程已經獲取到了一個鎖，那麼它能夠訪問被這個鎖鎖住的全部代碼塊。不可重入鎖與之相反。安全

Synchronized 關鍵字

Synchronized 是一種獨佔鎖。在修飾靜態方法時，鎖的是類對象，如 Object.class。修飾非靜態方法時，鎖的是對象，即 this。修飾方法塊時，鎖的是括號裏的對象。每一個對象有一個鎖和一個等待隊列，鎖只能被一個線程持有，其餘須要鎖的線程須要阻塞等待。鎖被釋放後，對象會從隊列中取出一個並喚醒，喚醒哪一個線程是不肯定的，不保證公平性。bash

類鎖與對象鎖

synchronized 修飾靜態方法時，鎖的是類對象,如 Object.class。修飾非靜態方法時，鎖的是對象，即 this。多個線程是能夠同時執行同一個synchronized實例方法的，只要它們訪問的對象是不一樣的。架構

synchronized 鎖住的是對象而非代碼，只要訪問的是同一個對象的 synchronized 方法，即便是不一樣的代碼，也會被同步順序訪問。併發

此外，須要說明的，synchronized方法不能防止非synchronized方法被同時執行，因此，通常在保護變量時，須要在全部訪問該變量的方法上加上synchronized。app

實現原理

synchronized 是基於 Java 對象頭和 Monitor 機制來實現的。

Java 對象頭

一個對象在內存中包含三部分：對象頭，實例數據和對齊填充。其中 Java 對象頭包含兩部分：

Class Metadata Address （類型指針）。存儲類的元數據的指針。虛擬機經過這個指針找到它是哪一個類的實例。
Mark Word（標記字段）。存出一些對象自身運行時的數據。包括哈希碼，GC 分代年齡，鎖狀態標誌等。

Monitor

Mark Word 有一個字段指向 monitor 對象。monitor 中記錄了鎖的持有線程，等待的線程隊列等信息。前面說的每一個對象都有一個鎖和一個等待隊列，就是在這裏實現的。 monitor 對象由 C++ 實現。其中有三個關鍵字段：

_owner 記錄當前持有鎖的線程
_EntryList 是一個隊列，記錄全部阻塞等待鎖的線程
_WaitSet 也是一個隊列，記錄調用 wait() 方法並還未被通知的線程。

Monitor的操做機制以下：

多個線程競爭鎖時，會先進入 EntryList 隊列。競爭成功的線程被標記爲 Owner。其餘線程繼續在此隊列中阻塞等待。
若是 Owner 線程調用 wait() 方法，則其釋放對象鎖並進入 WaitSet 中等待被喚醒。Owner 被置空，EntryList 中的線程再次競爭鎖。
若是 Owner 線程執行完了，便會釋放鎖，Owner 被置空，EntryList 中的線程再次競爭鎖。

JVM 對 synchronized 的處理

上面瞭解了 monitor 的機制，那虛擬機是如何將 synchronized 和 monitor 關聯起來的呢？分兩種狀況：

若是同步的是代碼塊，編譯時會直接在同步代碼塊前加上 monitorenter 指令，代碼塊後加上 monitorexit 指令。這稱爲顯示同步。
若是同步的是方法，虛擬機會爲方法設置 ACC_SYNCHRONIZED 標誌。調用的時候 JVM 根據這個標誌判斷是不是同步方法。

JVM 對 synchronized 的優化

synchronized 是重量級鎖，因爲消耗太大，虛擬機對其作了一些優化。

自旋鎖與自適應自旋

在許多應用中，鎖定狀態只會持續很短的時間，爲了這麼一點時間去掛起恢復線程，不值得。咱們可讓等待線程執行必定次數的循環，在循環中去獲取鎖。這項技術稱爲自旋鎖，它能夠節省系統切換線程的消耗，但仍然要佔用處理器。在 JDK1.4.2 中，自選的次數能夠經過參數來控制。 JDK 1.6又引入了自適應的自旋鎖，再也不經過次數來限制，而是由前一次在同一個鎖上的自旋時間及鎖的擁有者的狀態來決定。

鎖消除

虛擬機在運行時，若是發現一段被鎖住的代碼中不可能存在共享數據，就會將這個鎖清除。

鎖粗化

當虛擬機檢測到有一串零碎的操做都對同一個對象加鎖時，會把鎖擴展到整個操做序列外部。如 StringBuffer 的 append 操做。

輕量級鎖

對絕大部分的鎖來講，在整個同步週期內都不存在競爭。若是沒有競爭，輕量級鎖可使用 CAS 操做避免使用互斥量的開銷。

偏向鎖

偏向鎖的核心思想是，若是一個線程得到了鎖，那麼鎖就進入偏向模式，當這個線程再次請求鎖時，無需再作任何同步操做，便可獲取鎖。

CAS

操做模型

CAS 是 compare and swap 的簡寫，即比較並交換。它是指一種操做機制，而不是某個具體的類或方法。在 Java 平臺上對這種操做進行了包裝。在 Unsafe 類中，調用代碼以下：

unsafe.compareAndSwapInt(this, valueOffset, expect, update);
複製代碼

它須要三個參數，分別是內存位置 V，舊的預期值 A 和新的值 B。操做時，先從內存位置讀取到值，而後和預期值A比較。若是相等，則將此內存位置的值改成新值 B，返回 true。若是不相等，說明和其餘線程衝突了，則不作任何改變，返回 false。

這種機制在不阻塞其餘線程的狀況下避免了併發衝突，比獨佔鎖的性能高不少。 CAS 在 Java 的原子類和併發包中有大量使用。

重試機制（循環 CAS）

有不少文章說，CAS 操做失敗後會一直重試直到成功，這種說法很不嚴謹。

第一，CAS 自己並未實現失敗後的處理機制，它只負責返回成功或失敗的布爾值，後續由調用者自行處理。只不過咱們最經常使用的處理方式是重試而已。

第二，這句話很容易理解錯，被理解成從新比較並交換。實際上失敗的時候，原值已經被修改，若是不更改指望值，再怎麼比較都會失敗。而新值一樣須要修改。

因此正確的方法是，使用一個死循環進行 CAS 操做，成功了就結束循環返回，失敗了就從新從內存讀取值和計算新值，再調用 CAS。看下 AtomicInteger 的源碼就什麼都懂了：

public final int incrementAndGet () {
    for (;;) {
        int current = get();
        int next = current + 1;
        if (compareAndSet(current, next))
            return next;
    }
}
複製代碼

底層實現

CAS 主要分三步，讀取-比較-修改。其中比較是在檢測是否有衝突，若是檢測到沒有衝突後，其餘線程還能修改這個值，那麼 CAS 仍是沒法保證正確性。因此最關鍵的是要保證比較-修改這兩步操做的原子性。

CAS 底層是靠調用 CPU 指令集的 cmpxchg 完成的，它是 x86 和 Intel 架構中的 compare and exchange 指令。在多核的狀況下，這個指令也不能保證原子性，須要在前面加上 lock 指令。lock 指令能夠保證一個 CPU 核心在操做期間獨佔一片內存區域。那麼這又是如何實現的呢？

在處理器中，通常有兩種方式來實現上述效果：總線鎖和緩存鎖。在多核處理器的結構中，CPU 核心並不能直接訪問內存，而是統一經過一條總線訪問。總線鎖就是鎖住這條總線，使其餘核心沒法訪問內存。這種方式代價太大了，會致使其餘核心中止工做。而緩存鎖並不鎖定總線，只是鎖定某部份內存區域。當一個 CPU 核心將內存區域的數據讀取到本身的緩存區後，它會鎖定緩存對應的內存區域。鎖住期間，其餘核心沒法操做這塊內存區域。

CAS 就是經過這種方式實現比較和交換操做的原子性的。值得注意的是， CAS 只是保證了操做的原子性，並不保證變量的可見性，所以變量須要加上 volatile 關鍵字。

ABA 問題

上面提到，CAS 保證了比較和交換的原子性。可是從讀取到開始比較這段期間，其餘核心仍然是能夠修改這個值的。若是核心將 A 修改成 B，CAS 能夠判斷出來。可是若是核心將 A 修改成 B 再修改回 A。那麼 CAS 會認爲這個值並無被改變，從而繼續操做。這是和實際狀況不符的。解決方案是加一個版本號。

可重入鎖 ReentrantLock

ReentrantLock 使用代碼實現了和 synchronized 同樣的語義，包括可重入，保證內存可見性和解決競態條件問題等。相比 synchronized，它還有以下好處：

支持以非阻塞方式獲取鎖
能夠響應中斷
能夠限時
支持了公平鎖和非公平鎖

基本用法以下：

public class Counter {
    private final Lock lock = new ReentrantLock();
    private volatile int count;

    public void incr() {
        lock.lock();
        try {
            count++;
        } finally {
            lock.unlock();
        }
    }

    public int getCount() {
        return count;
    }
}
複製代碼

ReentrantLock 內部有兩個內部類，分別是 FairSync 和 NoFairSync，對應公平鎖和非公平鎖。他們都繼承自 Sync。Sync 又繼承自AQS。

AQS

AQS 全稱 AbstractQueuedSynchronizer。AQS 中有兩個重要的成員：

成員變量 state。用於表示鎖如今的狀態，用 volatile 修飾，保證內存一致性。同時所用對 state 的操做都是使用 CAS 進行的。state 爲0表示沒有任何線程持有這個鎖，線程持有該鎖後將 state 加1，釋放時減1。屢次持有釋放則屢次加減。
還有一個雙向鏈表，鏈表除了頭結點外，每個節點都記錄了線程的信息，表明一個等待線程。這是一個 FIFO 的鏈表。

下面以 ReentrantLock 非公平鎖的代碼看看 AQS 的原理。

請求鎖

請求鎖時有三種可能：

若是沒有線程持有鎖，則請求成功，當前線程直接獲取到鎖。
若是當前線程已經持有鎖，則使用 CAS 將 state 值加1，表示本身再次申請了鎖，釋放鎖時減1。這就是可重入性的實現。
若是由其餘線程持有鎖，那麼將本身添加進等待隊列。

final void lock() {
    if (compareAndSetState(0, 1))   
        setExclusiveOwnerThread(Thread.currentThread()); //沒有線程持有鎖時，直接獲取鎖，對應狀況1
    else
        acquire(1);
}

public final void acquire(int arg) {
    if (!tryAcquire(arg) && //在此方法中會判斷當前持有線程是否等於本身，對應狀況2
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //將本身加入隊列中，對應狀況3
        selfInterrupt();
}
複製代碼

建立 Node 節點並加入鏈表

若是沒競爭到鎖，這時候就要進入等待隊列。隊列是默認有一個 head 節點的，而且不包含線程信息。上面狀況3中，addWaiter 會建立一個 Node，並添加到鏈表的末尾，Node 中持有當前線程的引用。同時還有一個成員變量 waitStatus，表示線程的等待狀態，初始值爲0。咱們還須要關注兩個值：

CANCELLED，值爲1，表示取消狀態，就是說我不要這個鎖了，請你把我移出去。
SINGAL，值爲-1，表示下一個節點正在掛起等待，注意是下一個節點，不是當前節點。

同時，加到鏈表末尾的操做使用了 CAS+死循環的模式，頗有表明性，拿出來看一看：

Node node = new Node(mode);
for (;;) {
    Node oldTail = tail;
    if (oldTail != null) {
        U.putObject(node, Node.PREV, oldTail);
        if (compareAndSetTail(oldTail, node)) {
            oldTail.next = node;
            return node;
        }
    } else {
        initializeSyncQueue();
    }
}
複製代碼

能夠看到，在死循環裏調用了 CAS 的方法。若是多個線程同時調用該方法，那麼每次循環都只有一個線程執行成功，其餘線程進入下一次循環，從新調用。N個線程就會循環N次。這樣就在無鎖的模式下實現了併發模型。

掛起等待

若是此節點的上一個節點是頭部節點，則再次嘗試獲取鎖，獲取到了就移除並返回。獲取不到就進入下一步；
判斷前一個節點的 waitStatus，若是是 SINGAL，則返回 true，並調用 LockSupport.park() 將線程掛起；
若是是 CANCELLED，則將前一個節點移除；
若是是其餘值，則將前一個節點的 waitStatus 標記爲 SINGAL，進入下一次循環。

能夠看到，一個線程最多有兩次機會，還競爭不到就去掛起等待。

final boolean acquireQueued(final Node node, int arg) {
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    }
}
複製代碼

釋放鎖

調用 tryRelease，此方法由子類實現。實現很是簡單，若是當前線程是持有鎖的線程，就將 state 減1。減完後若是 state 大於0，表示當前線程仍然持有鎖，返回 false。若是等於0，表示已經沒有線程持有鎖，返回 true，進入下一步；
若是頭部節點的 waitStatus 不等於0，則調用LockSupport.unpark()喚醒其下一個節點。頭部節點的下一個節點就是等待隊列中的第一個線程，這反映了 AQS 先進先出的特色。另外，即便是非公平鎖，進入隊列以後，仍是得按順序來。

public final boolean release(int arg) {
    if (tryRelease(arg)) { //將 state 減1
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        node.compareAndSetWaitStatus(ws, 0);
        
    Node s = node.next;
    if (s == null || s.waitStatus > 0) { 
        s = null;
        for (Node p = tail; p != node && p != null; p = p.prev)
            if (p.waitStatus <= 0)
                s = p;
    }
    if (s != null) //喚醒第一個等待的線程
        LockSupport.unpark(s.thread);
}
複製代碼

公平鎖如何實現

上面分析的是非公平鎖，那公平鎖呢？很簡單，在競爭鎖以前判斷一下等待隊列中有沒有線程在等待就好了。

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;
        }
    }
    ......
    return false;
}
複製代碼