Java CAS 原理詳解

本文轉載自查看原文 2022-01-09 14:53 8044 java

1. 背景

在JDK 5之前Java語言是靠 synchronized 關鍵字保證同步的，這會導致有鎖。鎖機制存在以下問題：

在多線程競爭下，加鎖、釋放鎖會導致比較多的上下文切換和調度延時，引起性能問題。
一個線程持有鎖會導致其它所有需要此鎖的線程掛起。
如果一個優先級高的線程等待一個優先級低的線程釋放鎖會導致優先級倒置，引起性能風險。

Volatile關鍵字能夠在並發條件下，強制將修改后的值刷新到主內存中來保持內存的可見性。通過 CPU內存屏障禁止編譯器指令性重排來保證並發操作的有序性

如果多個線程同時操作 Volatile 修飾的變量，也會造成數據的不一致。

public class Test {
    public volatile int inc = 0;
     
    public void increase() {
        inc++;
    }     
    public static void main(String[] args) {
        final Test test = new Test();
        for(int i=0;i<10;i++){
            new Thread(){
                public void run() {
                    for(int j=0;j<1000;j++)
                        test.increase();
                };
            }.start();
        }    
        while(Thread.activeCount()>1)
            Thread.yield();
        System.out.println(test.inc);
    }
}

事實上運行它會發現每次運行結果都不一致，都是一個小於10000的數字。

假如某個時刻變量 inc 的值為10：

線程1對變量進行自增操作，線程1先讀取了變量inc的原始值，然后線程1被阻塞了；
然后線程2對變量進行自增操作，線程2也去讀取變量inc的原始值，由於線程1只是對變量inc進行讀取操作，而沒有對變量進行修改操作，所以不會導致線程2的工作內存中緩存變量inc的緩存行無效，所以線程2會直接去主存讀取inc的值，發現inc的值時10，然后進行加1操作，並把11寫入工作內存，最后寫入主存。
然后線程1接着進行加1操作，由於已經讀取了inc的值，注意此時在線程1的工作內存中inc的值仍然為10，所以線程1對inc進行加1操作后inc的值為11，然后將11寫入工作內存，最后寫入主存。
那么兩個線程分別進行了一次自增操作后，inc只增加了1。

之所以出現還是 volatile 只是保證讀寫具有原子性，但是對於 ++ 操作的復合操作是不存在原子操作的。只能在有限的一些情形下使用 volatile 變量替代鎖。要使 volatile 變量提供理想的線程安全，比如：對變量的寫操作不依賴於當前值。

volatile 是不錯的機制，但是 volatile 不能保證原子性。因此對於同步最終還是要回到鎖機制上來。

獨占鎖是一種悲觀鎖，synchronized 就是一種獨占鎖，會導致其它所有需要鎖的線程掛起，等待持有鎖的線程釋放鎖。而另一個更加有效的鎖就是樂觀鎖。所謂樂觀鎖就是，每次不加鎖而是假設沒有沖突而去完成某項操作，如果因為沖突失敗就重試，直到成功為止。樂觀鎖用到的機制就是 CAS，Compare and Swap。

2. CAS 原理

CAS 全稱是 compare and swap，是一種用於在多線程環境下實現同步功能的機制。CAS 操作包含三個操作數 -- 內存位置、預期數值和新值。CAS 的實現邏輯是將內存位置處的數值與預期數值想比較，若相等，則將內存位置處的值替換為新值。若不相等，則不做任何操作。

在 Java 中，Java 並沒有直接實現 CAS，CAS 相關的實現是通過 C++ 內聯匯編的形式實現的。Java 代碼需通過 JNI 才能調用。

CAS 是一條 CPU 的原子指令（cmpxchg指令），不會造成所謂的數據不一致問題，Unsafe 提供的 CAS 方法（如compareAndSwapXXX）底層實現即為 CPU 指令 cmpxchg

對 java.util.concurrent.atomic 包下的原子類 AtomicInteger 中的 compareAndSet 方法進行分析，相關分析如下：

public class AtomicInteger extends Number implements java.io.Serializable {    
    // setup to use Unsafe.compareAndSwapInt for updates
    private static final Unsafe unsafe = Unsafe.getUnsafe();    
    private static final long valueOffset;   
    static {       
        try {            
            // 計算變量 value 在類對象中的偏移
            valueOffset = unsafe.objectFieldOffset
                (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }    

    private volatile int value;    
    public final boolean compareAndSet(int expect, int update) {        

        /**
         * compareAndSet 實際上只是一個殼子，主要的邏輯封裝在 Unsafe 的 
         * compareAndSwapInt 方法中
         */
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    }    
    // ......
}
　　public final class Unsafe {    
      // compareAndSwapInt 是 native 類型的方法，繼續往下看
      public final native boolean compareAndSwapInt(Object o, long offset,  int expected,  int x);    
      // ......
  }

// unsafe.cpp
/*
 * 這個看起來好像不像一個函數，不過不用擔心，不是重點。UNSAFE_ENTRY 和 UNSAFE_END 都是宏，
 * 在預編譯期間會被替換成真正的代碼。下面的 jboolean、jlong 和 jint 等是一些類型定義（typedef）：
 * 
 * jni.h
 *     typedef unsigned char   jboolean;
 *     typedef unsigned short  jchar;
 *     typedef short           jshort;
 *     typedef float           jfloat;
 *     typedef double          jdouble;
 * 
 * jni_md.h
 *     typedef int jint;
 *     #ifdef _LP64 // 64-bit
 *     typedef long jlong;
 *     #else
 *     typedef long long jlong;
 *     #endif
 *     typedef signed char jbyte;
 */
UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x))
  UnsafeWrapper("Unsafe_CompareAndSwapInt");
  oop p = JNIHandles::resolve(obj);  

  // 根據偏移量，計算 value 的地址。這里的 offset 就是 AtomaicInteger 中的 valueOffset
  jint* addr = (jint *) index_oop_from_field_offset_long(p, offset);  

  // 調用 Atomic 中的函數 cmpxchg，該函數聲明於 Atomic.hpp 中
  return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
UNSAFE_END

 atomic.cppunsigned Atomic::cmpxchg(unsigned int exchange_value,  volatile unsigned int* dest, unsigned int compare_value) {
  assert(sizeof(unsigned int) == sizeof(jint), "more work to do");  

  /*
   * 根據操作系統類型調用不同平台下的重載函數，這個在預編譯期間編譯器會決定調用哪個平台下的重載
   * 函數。相關的預編譯邏輯如下：
   * 
   * atomic.inline.hpp：
   *    #include "runtime/atomic.hpp"
   *    
   *    // Linux
   *    #ifdef TARGET_OS_ARCH_linux_x86
   *    # include "atomic_linux_x86.inline.hpp"
   *    #endif
   *   
   *    // 省略部分代碼
   *    
   *    // Windows
   *    #ifdef TARGET_OS_ARCH_windows_x86
   *    # include "atomic_windows_x86.inline.hpp"
   *    #endif
   *    
   *    // BSD
   *    #ifdef TARGET_OS_ARCH_bsd_x86
   *    # include "atomic_bsd_x86.inline.hpp"
   *    #endif
   * 
   * 接下來分析 atomic_windows_x86.inline.hpp 中的 cmpxchg 函數實現
   */
  return (unsigned int)Atomic::cmpxchg((jint)exchange_value, (volatile jint*)dest,  (jint)compare_value);
}

上面的分析看起來比較多，不過主流程並不復雜。如果不糾結於代碼細節，還是比較容易看懂的。接下來，我會分析 Windows 平台下的 Atomic::cmpxchg 函數。繼續往下看吧。

// atomic_windows_x86.inline.hpp
#define LOCK_IF_MP(mp) __asm cmp mp, 0  \
                       __asm je L0      \
                       __asm _emit 0xF0 \
                       __asm L0:
              inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {  // alternative for InterlockedCompareExchange
  int mp = os::is_MP();
  __asm {
    mov edx, dest
    mov ecx, exchange_value
    mov eax, compare_value    LOCK_IF_MP(mp)
    cmpxchg dword ptr [edx], ecx
  }
}

上面的代碼由 LOCK_IF_MP 預編譯標識符和 cmpxchg 函數組成。為了看到更清楚一些，我們將 cmpxchg 函數中的 LOCK_IF_MP 替換為實際內容。如下：

inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {  
  // 判斷是否是多核 CPU
  int mp = os::is_MP();
  __asm {    // 將參數值放入寄存器中
    mov edx, dest    // 注意: dest 是指針類型，這里是把內存地址存入 edx 寄存器中
    mov ecx, exchange_value
    mov eax, compare_value    
    // LOCK_IF_MP
    cmp mp, 0
    /*
     * 如果 mp = 0，表明是線程運行在單核 CPU 環境下。此時 je 會跳轉到 L0 標記處，
     * 也就是越過 _emit 0xF0 指令，直接執行 cmpxchg 指令。也就是不在下面的 cmpxchg 指令
     * 前加 lock 前綴。
     */
    je L0    /*
     * 0xF0 是 lock 前綴的機器碼，這里沒有使用 lock，而是直接使用了機器碼的形式。至於這樣做的
     * 原因可以參考知乎的一個回答：
     *     https://www.zhihu.com/question/50878124/answer/123099923
     */ 
    _emit 0xF0L0:    /*
     * 比較並交換。簡單解釋一下下面這條指令，熟悉匯編的朋友可以略過下面的解釋:
     *   cmpxchg: 即“比較並交換”指令
     *   dword: 全稱是 double word，在 x86/x64 體系中，一個 
     *          word = 2 byte，dword = 4 byte = 32 bit
     *   ptr: 全稱是 pointer，與前面的 dword 連起來使用，表明訪問的內存單元是一個雙字單元
     *   [edx]: [...] 表示一個內存單元，edx 是寄存器，dest 指針值存放在 edx 中。
     *          那么 [edx] 表示內存地址為 dest 的內存單元
     *          
     * 這一條指令的意思就是，將 eax 寄存器中的值（compare_value）與 [edx] 雙字內存單元中的值
     * 進行對比，如果相同，則將 ecx 寄存器中的值（exchange_value）存入 [edx] 內存單元中。
     */
    cmpxchg dword ptr [edx], ecx
  }
}

到這里 CAS 的實現過程就講完了，CAS 的實現離不開處理器的支持。如上面源代碼所示，程序會根據當前處理器的類型來決定是否為 cmpxchg 指令添加 lock 前綴。如果程序是在多處理器上運行，就為 cmpxchg 指令加上 lock 前綴（lock cmpxchg）。反之，如果程序是在單處理器上運行，就省略 lock 前綴（單處理器自身會維護單處理器內的順序一致性，不需要 lock 前綴提供的內存屏障效果）。

intel 的手冊對 lock 前綴的說明如下：

確保對內存的讀 - 改 - 寫操作原子執行。在 Pentium 及 Pentium 之前的處理器中，帶有 lock 前綴的指令在執行期間會鎖住總線，使得其他處理器暫時無法通過總線訪問內存。很顯然，這會帶來昂貴的開銷。從 Pentium 4，Intel Xeon 及 P6 處理器開始，intel 在原有總線鎖的基礎上做了一個很有意義的優化：如果要訪問的內存區域（area of memory）在 lock 前綴指令執行期間已經在處理器內部的緩存中被鎖定（即包含該內存區域的緩存行當前處於獨占或以修改狀態），並且該內存區域被完全包含在單個緩存行（cache line）中，那么處理器將直接執行該指令。由於在指令執行期間該緩存行會一直被鎖定，其它處理器無法讀 / 寫該指令要訪問的內存區域，因此能保證指令執行的原子性。這個操作過程叫做緩存鎖定（cache locking），緩存鎖定將大大降低 lock 前綴指令的執行開銷，但是當多處理器之間的競爭程度很高或者指令訪問的內存地址未對齊時，仍然會鎖住總線。
禁止該指令與之前和之后的讀和寫指令重排序。
把寫緩沖區中的所有數據刷新到內存中。

上面的第 2 點和第 3 點所具有的內存屏障效果，足以同時實現 volatile 讀和 volatile 寫的內存語義。

經過上面的這些分析，現在我們終於能明白為什么 JDK 文檔說 CAS 同時具有 volatile 讀和 volatile 寫的內存語義了。

Java 的 CAS 會使用現代處理器上提供的高效機器級別原子指令，這些原子指令以原子方式對內存執行讀 - 改 - 寫操作，這是在多處理器中實現同步的關鍵（從本質上來說，能夠支持原子性讀 - 改 - 寫指令的計算機器，是順序計算圖靈機的異步等價機器，因此任何現代的多處理器都會去支持某種能對內存執行原子性讀 - 改 - 寫操作的原子指令）。同時，volatile 變量的讀 / 寫和 CAS 可以實現線程之間的通信。把這些特性整合在一起，就形成了整個 concurrent 包得以實現的基石。如果我們仔細分析 concurrent 包的源代碼實現，會發現一個通用化的實現模式：

首先，聲明共享變量為 volatile；
然后，使用 CAS 的原子條件更新來實現線程之間的同步；
同時，配合以 volatile 的讀 / 寫和 CAS 所具有的 volatile 讀和寫的內存語義來實現線程之間的通信。

AQS，非阻塞數據結構和原子變量類（java.util.concurrent.atomic 包中的類），這些 concurrent 包中的基礎類都是使用這種模式來實現的，而 concurrent 包中的高層類又是依賴於這些基礎類來實現的。從整體來看，concurrent 包的實現示意圖如下：

JVM中的CAS（堆中對象的分配）：　

Java 調用 new object() 會創建一個對象，這個對象會被分配到 JVM 的堆中。那么這個對象到底是怎么在堆中保存的呢？

首先，new object() 執行的時候，這個對象需要多大的空間，其實是已經確定的，因為 java 中的各種數據類型，占用多大的空間都是固定的（對其原理不清楚的請自行Google）。那么接下來的工作就是在堆中找出那么一塊空間用於存放這個對象。

在單線程的情況下，一般有兩種分配策略：

指針碰撞：這種一般適用於內存是絕對規整的（內存是否規整取決於內存回收策略），分配空間的工作只是將指針像空閑內存一側移動對象大小的距離即可。
空閑列表：這種適用於內存非規整的情況，這種情況下JVM會維護一個內存列表，記錄哪些內存區域是空閑的，大小是多少。給對象分配空間的時候去空閑列表里查詢到合適的區域然后進行分配即可。

但是JVM不可能一直在單線程狀態下運行，那樣效率太差了。由於再給一個對象分配內存的時候不是原子性的操作，至少需要以下幾步：查找空閑列表、分配內存、修改空閑列表等等，這是不安全的。解決並發時的安全問題也有兩種策略：

CAS：實際上虛擬機采用CAS配合上失敗重試的方式保證更新操作的原子性，原理和上面講的一樣。
TLAB：如果使用CAS其實對性能還是會有影響的，所以 JVM 又提出了一種更高級的優化策略：每個線程在 Java 堆中預先分配一小塊內存，稱為本地線程分配緩沖區（TLAB），線程內部需要分配內存時直接在 TLAB 上分配就行，避免了線程沖突。只有當緩沖區的內存用光需要重新分配內存的時候才會進行CAS操作分配更大的內存空間。

虛擬機是否使用TLAB，可以通過-XX:+/-UseTLAB參數來進行配置（jdk5及以后的版本默認是啟用TLAB的）。

3. CAS存在的問題

3.1 ABA 問題

談到 CAS，基本上都要談一下 CAS 的 ABA 問題。CAS 由三個步驟組成，分別是“讀取-比較-寫回”。考慮這樣一種情況，線程1和線程2同時執行 CAS 邏輯，兩個線程的執行順序如下：

時刻1：線程1執行讀取操作，獲取原值 A，然后線程被切換走
時刻2：線程2執行完成 CAS 操作將原值由 A 修改為 B
時刻3：線程2再次執行 CAS 操作，並將原值由 B 修改為 A
時刻4：線程1恢復運行，將比較值(compareValue)與原值(oldValue)進行比較，發現兩個值相等。

然后用新值(newValue)寫入內存中，完成 CAS 操作

如上流程，線程1並不知道原值已經被修改過了，在它看來並沒什么變化，所以它會繼續往下執行流程。對於 ABA 問題，通常的處理措施是對每一次 CAS 操作設置版本號。

ABA問題的解決思路其實也很簡單，就是使用版本號。在變量前面追加上版本號，每次變量更新的時候把版本號加1，那么A→B→A就會變成1A→2B→3A了。

java.util.concurrent.atomic 包下提供了一個可處理 ABA 問題的原子類 AtomicStampedReference，

從Java1.5開始JDK的atomic包里提供了一個類AtomicStampedReference來解決ABA問題。這個類的compareAndSet方法作用是首先檢查當前引用是否等於預期引用，並且當前標志是否等於預期標志，如果全部相等，則以原子方式將該引用和該標志的值設置為給定的更新值。

3.2 循環時間長開銷大

自旋CAS（不成功，就一直循環執行，直到成功）如果長時間不成功，會給 CPU 帶來非常大的執行開銷。如果JVM能支持處理器提供的 pause 指令那么效率會有一定的提升，pause指令有兩個作用，第一它可以延遲流水線執行指令（de-pipeline），使 CPU 不會消耗過多的執行資源，延遲的時間取決於具體實現的版本，在一些處理器上延遲時間是零。第二它可以避免在退出循環的時候因內存順序沖突（memory order violation）而引起 CPU 流水線被清空（CPU pipeline flush），從而提高 CPU 的執行效率。

3.3 只能保證一個共享變量的原子操作

當對一個共享變量執行操作時，我們可以使用循環 CAS 的方式來保證原子操作，但是對多個共享變量操作時，循環 CAS 就無法保證操作的原子性，這個時候就可以用鎖，或者有一個取巧的辦法，就是把多個共享變量合並成一個共享變量來操作。比如有兩個共享變量 i＝2，j=a，合並一下 ij=2a，然后用CAS來操作ij。從Java1.5開始JDK提供了 AtomicReference 類來保證引用對象之間的原子性，你可以把多個變量放在一個對象里來進行 CAS 操作。

CAS 與 Synchronized 的使用情景：　　　

對於資源競爭較少（線程沖突較輕）的情況，使用synchronized同步鎖進行線程阻塞和喚醒切換以及用戶態內核態間的切換操作額外浪費消耗cpu資源；而CAS基於硬件實現，不需要進入內核，不需要切換線程，操作自旋幾率較少，因此可以獲得更高的性能。
對於資源競爭嚴重（線程沖突嚴重）的情況，CAS自旋的概率會比較大，從而浪費更多的CPU資源，效率低於synchronized。

補充： synchronized 在 jdk1.6 之后，已經改進優化。synchronized 的底層實現主要依靠 Lock-Free 的隊列，基本思路是自旋后阻塞，競爭切換后繼續競爭鎖，稍微犧牲了公平性，但獲得了高吞吐量。在線程沖突較少的情況下，可以獲得和 CAS 類似的性能；而線程沖突嚴重的情況下，性能遠高於 CAS。

其他

什么是happen-before

JMM 可以通過 happens-before 關系向程序員提供跨線程的內存可見性保證（如果 A 線程的寫操作 a 與 B 線程的讀操作 b 之間存在 happens-before 關系，盡管 a 操作和 b 操作在不同的線程中執行，但 JMM 向程序員保證 a 操作將對 b 操作可見）。

具體的定義為：

如果一個操作happens-before另一個操作，那么第一個操作的執行結果將對第二個操作可見，而且第一個操作的執行順序排在第二個操作之前。
兩個操作之間存在happens-before關系，並不意味着Java平台的具體實現必須要按照happens-before關系指定的順序來執行。如果重排序之后的執行結果，與按happens-before關系來執行的結果一致，那么這種重排序並不非法（也就是說，JMM允許這種重排序）。

具體的規則：

程序順序規則：一個線程中的每個操作，happens-before於該線程中的任意后續操作。
監視器鎖規則：對一個鎖的解鎖，happens-before於隨后對這個鎖的加鎖。
volatile變量規則：對一個volatile域的寫，happens-before於任意后續對這個volatile域的讀。
傳遞性：如果A happens-before B，且B happens-before C，那么A happens-before C。
start()規則：如果線程A執行操作ThreadB.start()（啟動線程B），那么A線程的ThreadB.start()操作happens-before於線程B中的任意操作。
Join()規則：如果線程A執行操作ThreadB.join()並成功返回，那么線程B中的任意操作happens-before於線程A從ThreadB.join()操作成功返回。
程序中斷規則：對線程interrupted()方法的調用先行於被中斷線程的代碼檢測到中斷時間的發生。
對象finalize規則：一個對象的初始化完成（構造函數執行結束）先行於發生它的finalize()方法的開始。