介紹Java CAS 原理分析

java基礎教學專欄介紹分析Java CAS

推薦（免費）：java基礎教學

1、簡介

CAS 全名為compare and swap，一種用於在多執行緒環境下實現同步功能的機制。 CAS 運算包含三個運算元 -- 記憶體位置、預期數值和新值。 CAS 的實作邏輯是將記憶體位置的數值​​與預期數值想比較，若相等，則將記憶體位置的值替換為新值。若不相等，則不做任何操作。

在 Java 中，Java 並沒有直接實作 CAS，CAS 相關的實作是透過 C 內聯彙編的形式實現的。 Java 程式碼需透過 JNI 才能呼叫。關於實現上的細節，我將會在第3章進行分析。

前面說了 CAS 操作的流程，並不是很難。但僅有上面的說明還不夠，接下來我將會再介紹一點其他的背景知識。有這些背景知識，才能更好的理解後續的內容。

2.背景介紹

我們都知道，CPU 是透過匯流排和記憶體進行資料傳輸的。在多核心時代下，多個核心透過同一條匯流排和記憶體以及其他硬體進行通訊。如下圖：

圖片來源：《深入理解電腦系統》

上圖是較簡單的電腦結構圖，雖然簡單，但足以說明問題。在上圖中，CPU 透過兩個藍色箭頭標註的匯流排與記憶體通訊。大家考慮一個問題，CPU 的多個核心同時對同一片記憶體進行操作，若不加以控制，會導致什麼樣的錯誤？這裡簡單說明一下，假設核心1經32位元頻寬的匯流排向記憶體寫入64位元的數據，核心1要進行兩次寫入才能完成整個操作。若在核心1第一次寫入32位元的資料後，核心2從核心1寫入的記憶體位置讀取了64位元資料。由於核心1還未完全將64位的數據全部寫入內存中，核心2就開始從該內存位置讀取數據，那麼讀取出來的數據必定是混亂的。

不過對於這個問題，其實不用擔心。透過 Intel 開發人員手冊，我們可以了解到自奔騰處理器開始，Intel 處理器會保證以原子的方式讀寫按64位元邊界對齊的四字（quadword）。

根據上面的說明，我們可總結出，Intel 處理器可以保證單次存取記憶體對齊的指令以原子的方式執行。但如果是兩次訪存的指示呢？答案是無法保證。例如遞增指令inc dword ptr [...]，等價於DEST = DEST 1。此指令包含三個操作讀取->改->寫，涉及兩次訪問。考慮這樣一種情況，在記憶體指定位置處，存放了一個為1的數值。現在 CPU 兩個核心同時執行該條指令。兩個核心交替執行的流程如下：

1. 核心1 從記憶體指定位置出讀取數值1，並載入到暫存器中
2. 核心2 從記憶體指定位置出讀取數值1，並載入到暫存器中
3. 核心1 將暫存器中值遞減1
4. #核心2 將暫存器中值遞減1
5. 核心1 將修改後的值寫回記憶體
6. 核心2 將修改後的值寫回記憶體

經過執行上述流程，記憶體中的最終值時2，而我們期待的是3，這就出問題了。要處理這個問題，就要避免兩個或多個核心同時操作同一片記憶體區域。那麼要怎樣避免呢？這就要引入本文的主角 - lock 前綴。關於該指令的詳細描述，可以參考 Intel 開發人員手冊 Volume 2 Instruction Set Reference，Chapter 3 Instruction Set Reference A-L。我在這裡引用其中的一段，如下：

LOCK—Assert LOCK# Signal Prefix
Causes the processor's LOCK# signal to be asserted during execution of the accompanying instruction (turns the instruction into an micion into an mic instruction). In a multiprocessor environment, the LOCK# signal ensures that the processor has exclusive use of any shared memory while the signal is asserted.

##上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述的重點已經用上面描述。黑體標示了，在多處理器環境下，LOCK# 訊號可以確保處理器獨佔使用某些共享記憶體。 lock 可以加入下面的指令：

ADD, ADC, AND, BTC, BTR, BTS, CMPXCHG, CMPXCH8B, CMPXCHG16B, DEC, INC, NEG, NOT, OR, SBB, SUB, XOR, XADD, and XCHG.

在 inc 指令前面加上 lock 前綴，即可讓指令具備原子性。多個核心同時執行同一條 inc 指令時，會以串列的方式進行，也就避免了上面所說的那種情況。那麼這裡還有一個問題，lock 前綴是怎麼保證核心獨佔某片記憶體區域的呢？答案如下：

在 Intel 處理器中，有兩種​​方式確保處理器的某個核心獨佔某片記憶體區域。第一種方式是透過鎖定匯流排，讓某個核心獨佔使用匯流排，但這樣代價太大。總線被鎖定後，其他核心就無法存取記憶體了，可能會導致其他核心短暫內停止工作。第二種方式是鎖定緩存，若某處記憶體資料緩存在處理器快取中。處理器發出的 LOCK# 訊號不會鎖定匯流排，而是鎖定快取行對應的記憶體區域。其他處理器在這片記憶體區域鎖定期間，無法對這片記憶體區域進行相關操作。相對於鎖定總線，鎖定快取的代價明顯比較小。關於匯流排鎖和快取鎖，更詳細的描述請參考 Intel 開發人員手冊 Volume 3 Software Developer’s Manual，Chapter 8 Multiple-Processor Management。

3.原始碼分析

有了上面的背景知識，現在我們就可以從容不迫的閱讀 CAS 的源碼了。本章的內容將對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.cpp
unsigned 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 0xF0
L0:
    /*
     * 比较并交换。简单解释一下下面这条指令，熟悉汇编的朋友可以略过下面的解释:
     *   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 的實作離不開處理器的支援。以上這麼多程式碼，其實核心程式碼就是一條有lock 前綴的 cmpxchg 指令，也就是lock cmpxchg dword ptr [edx], ecx。

4.ABA 問題

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

1. #時刻1：執行緒1執行讀取操作，取得原值A，然後執行緒被切換走
2. 時刻2：執行緒2執行完成CAS 操作將原值由A 修改為B
3. 時刻3：執行緒2再次執行CAS 操作，並將原值由B 修改為A
4. 時刻4：執行緒1恢復運行，將比較值（compareValue）與原值（oldValue）進行比較，發現兩個值相等。然後用新值（newValue）寫入記憶體中，完成CAS 操作

如上流程，線程1並不知道原值已經被修改過了，在它看來並沒什麼變化，所以它會繼續往下執行流程。對於 ABA 問題，通常的處理措施是對每個 CAS 操作設定版本號。 java.util.concurrent.atomic 包下提供了一個可處理 ABA 問題的原子類 AtomicStampedReference，具體的實現這裡就不分析了，有興趣的朋友可以自己去看看。

5.總結

寫到這裡，這篇文章總算接近尾聲了。雖然 CAS 本身的原理，包括實作都不是很難，但寫起來真的不太好寫。這裡面牽涉到了一些底層的知識，雖然能看懂，但想說明白，還是有點難度的。由於我底層的知識比較欠缺，上面的一些分析難免會出錯。所以如有錯誤，請輕噴，當然最好能說明怎麼錯的，感謝。

好了，這篇文章就到這裡。感謝閱讀，再見。

附錄

在前面原始碼分析一節中用到的幾個文件，這裡把路徑貼出來。有助於大家進行索引，如下：

openjdk/jdk/src/share/classes/sun/misc/Unsafe.javaopenjdk/ hotspot/src/share/vm/prims/unsafe.cppopenjdk/hotspot/src/share/vm/runtime/atomic.cpp

檔案名稱	#路徑
	#Unsafe.java
	unsafe.cpp
	atomic.cpp

######atomic_windows_x86.inline.hpp######openjdk/hotspot/src/os_cpu/windows_x86/vm/atomic_windows_x86.inline.hpp######## #

以上是介紹Java CAS 原理分析的詳細內容。更多資訊請關注PHP中文網其他相關文章！