Java併發基礎：瞭解無鎖CAS就從源碼分析

時間 2019-12-05

標籤 java 併發基礎瞭解 cas 源碼分析欄目 Java 简体版

原文原文鏈接

CAS的全稱爲Compare And Swap，直譯就是比較交換。是一條CPU的原子指令，其做用是讓CPU先進行比較兩個值是否相等，而後原子地更新某個位置的值，其實現方式是基於硬件平臺的彙編指令，在intel的CPU中，使用的是cmpxchg指令，就是說CAS是靠硬件實現的，從而在硬件層面提高效率。html

CSA 原理

利用CPU的CAS指令，同時藉助JNI來完成Java的非阻塞算法，其它原子操做都是利用相似的特性完成的。
在 java.util.concurrent 下面的源碼中，Atomic, ReentrantLock 都使用了Unsafe類中的方法來保證併發的安全性。java

CAS操做是原子性的，因此多線程併發使用CAS更新數據時，能夠不使用鎖，JDK中大量使用了CAS來更新數據而防止加鎖來保持原子更新。linux

CAS 操做包含三個操做數：內存偏移量位置（V）、預期原值（A）和新值(B)。若是內存位置的值與預期原值相匹配，那麼處理器會自動將該位置值更新爲新值。不然，處理器不作任何操做。算法

源碼分析

下面來看一下 java.util.concurrent.atomic.AtomicInteger.java，getAndIncrement()，getAndDecrement()是如何利用CAS實現原子性操做的。windows

AtomicInteger 源碼解析

// 使用 unsafe 類的原子操做方式
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); }
}

valueOffset 字段表示 "value" 內存位置，在compareAndSwap 方法 ,第二個參數會用到.數組

關於偏移量安全

Unsafe 調用C 語言能夠經過偏移量對變量進行操做多線程

//volatile變量value
private volatile int value;

 /**
 * 建立具備給定初始值的新 AtomicInteger
 *
 * @param initialValue 初始值
 */
public AtomicInteger(int initialValue) {
    value = initialValue;
}

//返回當前的值
public final int get() {
    return value;
}
//原子更新爲新值並返回舊值
public final int getAndSet(int newValue) {
    return unsafe.getAndSetInt(this, valueOffset, newValue);
}
//最終會設置成新值
public final void lazySet(int newValue) {
    unsafe.putOrderedInt(this, valueOffset, newValue);
}
//若是輸入的值等於預期值，則以原子方式更新爲新值
public final boolean compareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

//方法至關於原子性的 ++i
public final int getAndIncrement() {
    //三個參數，一、當前的實例 二、value實例變量的偏移量 三、遞增的值。
    return unsafe.getAndAddInt(this, valueOffset, 1);
}
//方法至關於原子性的 --i
public final int getAndDecrement() {
    //三個參數，一、當前的實例 二、value實例變量的偏移量 三、遞減的值。
    return unsafe.getAndAddInt(this, valueOffset, -1);
}

實現邏輯封裝在 Unsafe 中 getAndAddInt 方法，繼續往下看，Unsafe 源碼解析併發

Unsafe 源碼解析

在JDK8中追蹤可見sun.misc.Unsafe這個類是沒法看見源碼的，打開openjdk8源碼看app

文件：openjdk-8-src-b132-03_mar_2014.zip

目錄：openjdk\jdk\src\share\classes\sun\misc\Unsafe.java

一般咱們最好也不要使用Unsafe類，除非有明確的目的，而且也要對它有深刻的瞭解才行。要想使用Unsafe類須要用一些比較tricky的辦法。Unsafe類使用了單例模式，須要經過一個靜態方法getUnsafe()來獲取。但Unsafe類作了限制，若是是普通的調用的話，它會拋出一個SecurityException異常；只有由主類加載器加載的類才能調用這個方法。

下面是sun.misc.Unsafe.java類源碼

//獲取Unsafe實例靜態方法
@CallerSensitive
public static Unsafe getUnsafe() {
    Class<?> caller = Reflection.getCallerClass();
    if (!VM.isSystemDomainLoader(caller.getClassLoader()))
        throw new SecurityException("Unsafe");
    return theUnsafe;
}

網上也有一些辦法來用主類加載器加載用戶代碼，最簡單方法是利用Java反射，方法以下：

private static Unsafe unsafe;

static {
    try {
        //經過反射獲取rt.jar下的Unsafe類
        Field field = Unsafe.class.getDeclaredField("theUnsafe");
        field.setAccessible(true);
        unsafe = (Unsafe) field.get(null);
    } catch (Exception e) {
        System.out.println("Get Unsafe instance occur error" + e);
    }
}

獲取到Unsafe實例以後，咱們就能夠隨心所欲了。Unsafe類提供瞭如下這些功能：

https://www.cnblogs.com/pkufork/p/java_unsafe.html

//native硬件級別的原子操做
    //相似的有compareAndSwapInt,compareAndSwapLong,compareAndSwapBoolean,compareAndSwapChar等等。
    public final native boolean compareAndSwapInt(Object o, long offset,int expected,int x);

    //內部使用自旋的方式進行CAS更新（while循環進行CAS更新，若是更新失敗，則循環再次重試）
    public final int getAndAddInt(Object o, long offset, int delta) {
        int v;
        do {
            //獲取對象內存地址偏移量上的數值v
            v = getIntVolatile(o, offset);
            //若是如今仍是v,設置爲 v + delta,不然返回false,繼續循環再次重試.
        } while (!compareAndSwapInt(o, offset, v, v + delta));
        return v;
    }

利用 Unsafe 類的 JNI compareAndSwapInt 方法實現，使用CAS實現一個原子操做更新，

compareAndSwapInt 四個參數：

一、當前的實例
二、實例變量的內存地址偏移量
三、預期的舊值
四、要更新的值

unsafe.cpp 深層次解析

// unsafe.cpp
/*
 * 這個看起來好像不像一個函數，不過不用擔憂，不是重點。UNSAFE_ENTRY 和 UNSAFE_END 都是宏，
 * 在預編譯期間會被替換成真正的代碼。下面的 jboolean、jlong 和 jint 等是一些類型定義（typedef）：
 *
 * 省略部份內容
 */
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。

經過上述的分析，能夠發現AtomicInteger原子類的內部幾乎是基於前面分析過Unsafe類中的CAS相關操做的方法實現的，這也同時證實AtomicInteger getAndIncrement自增操做實現過程,是基於無鎖實現的。

CAS的ABA問題及其解決方案

假設這樣一種場景，當第一個線程執行CAS(V,E,U)操做。在獲取到當前變量V，準備修改成新值U前，另外兩個線程已連續修改了兩次變量V的值，使得該值又恢復爲舊值，這樣的話，咱們就沒法正確判斷這個變量是否已被修改過，以下圖：

這就是典型的CAS的ABA問題，通常狀況這種狀況發現的機率比較小，可能發生了也不會形成什麼問題，好比說咱們對某個作加減法，不關心數字的過程，那麼發生ABA問題也沒啥關係。可是在某些狀況下仍是須要防止的，那麼該如何解決呢？在Java中解決ABA問題，咱們可使用如下原子類

AtomicStampedReference類

AtomicStampedReference原子類是一個帶有時間戳的對象引用，在每次修改後，AtomicStampedReference不只會設置新值並且還會記錄更改的時間。當AtomicStampedReference設置對象值時，對象值以及時間戳都必須知足指望值才能寫入成功，這也就解決了反覆讀寫時，沒法預知值是否已被修改的窘境

底層實現爲：經過Pair私有內部類存儲數據和時間戳, 並構造volatile修飾的私有實例

接着看 java.util.concurrent.atomic.AtomicStampedReference類的compareAndSet（）方法的實現：

private static class Pair<T> {
    final T reference;
    final int stamp;
  
    //最好不要重複的一個數據,決定數據是否能設置成功,時間戳會重複
    private Pair(T reference, int stamp) {
        this.reference = reference;
        this.stamp = stamp;
    }
    static <T> Pair<T> of(T reference, int stamp) {
        return new Pair<T>(reference, stamp);
    }
}

同時對當前數據和當前時間進行比較，只有二者都相等是纔會執行casPair()方法，

單從該方法的名稱就可知是一個CAS方法，最終調用的仍是Unsafe類中的compareAndSwapObject方法

到這咱們就很清晰AtomicStampedReference的內部實現思想了，

經過一個鍵值對Pair存儲數據和時間戳，在更新時對數據和時間戳進行比較，

只有二者都符合預期纔會調用Unsafe的compareAndSwapObject方法執行數值和時間戳替換，也就避免了ABA的問題。

/**
 * 原子更新帶有版本號的引用類型。
 * 該類將整數值與引用關聯起來，可用於原子的更數據和數據的版本號。
 * 能夠解決使用CAS進行原子更新時，可能出現的ABA問題。
 */
public class AtomicStampedReference<V> {
    //靜態內部類Pair將對應的引用類型和版本號stamp做爲它的成員
    private static class Pair<T> {
      
        //最好不要重複的一個數據,決定數據是否能設置成功,建議時間戳
        final T reference;
        final int stamp;
        private Pair(T reference, int stamp) {
            this.reference = reference;
            this.stamp = stamp;
        }
      
        //根據reference和stamp來生成一個Pair的實例
        static <T> Pair<T> of(T reference, int stamp) {
            return new Pair<T>(reference, stamp);
        }
    }
  
    //做爲一個總體的pair變量被volatile修飾
    private volatile Pair<V> pair;
 
    //構造方法，參數分別是初始引用變量的值和初始版本號
    public AtomicStampedReference(V initialRef, int initialStamp) {
        pair = Pair.of(initialRef, initialStamp);
    }
  
    ....
  
    private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();
  
    private static final long pairOffset = objectFieldOffset(UNSAFE, "pair", AtomicStampedReference.class);
 
    //獲取pair成員的偏移地址
    static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
                                  String field, Class<?> klazz) {
        try {
            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
        } catch (NoSuchFieldException e) {
            NoSuchFieldError error = new NoSuchFieldError(field);
            error.initCause(e);
            throw error;
        }
    }
}

/**
 * @param 指望(老的)引用
 * @param       (新的)引用數據
 * @param 指望(老的)標誌stamp(時間戳)值
 * @param       (新的)標誌stamp(時間戳)值
 * @return 是否成功
 */
public boolean compareAndSet(V expectedReference,V   newReference,int expectedStamp,int newStamp) {
       
    Pair<V> current = pair;
    return
        // 指望(老的)引用 == 當前引用
        expectedReference == current.reference &&
        // 指望(老的)標誌stamp(時間戳)值 == 當前標誌stamp(時間戳)值
        expectedStamp == current.stamp &&
      
        // (新的)引用數據 == 當前引用數據 而且 (新的)標誌stamp(時間戳)值 ==當前標誌stamp(時間戳)值
        ((newReference == current.reference && newStamp == current.stamp) ||
          #原子更新值
         casPair(current, Pair.of(newReference, newStamp)));
       
}
 
 //當引用類型的值與指望的一致的時候，原子的更改版本號爲新的值。該方法只修改版本號，不修改引用變量的值，成功返回true
public boolean attemptStamp(V expectedReference, int newStamp) {
    Pair<V> current = pair;
    return
        expectedReference == current.reference &&
        (newStamp == current.stamp ||
         casPair(current, Pair.of(expectedReference, newStamp)));
}

/**
 * CAS真正實現方法
 */
private boolean casPair(Pair<V> cmp, Pair<V> val) {
        return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
}

指望 Pair<V> cmp（A） == 當前內存存偏移量位置 Pair（V），就更新值 Pair<V> val（B）成功返回true 不然 false

public static void main(String[] args) {
    AtomicStampedReference<Integer> num = new AtomicStampedReference<Integer>(1, 0);

    Integer i = num.getReference();
    int stamped = num.getStamp();

    if (num.compareAndSet(i, i + 1, stamped, stamped + 1)) {
        System.out.println("測試成功");
    }
}

經過以上原子更新方法，可見 AtomicStampedReference就是利用了Unsafe的CAS方法+Volatile關鍵字對存儲實際的引用變量和int的版本號的Pair實例進行更新。

參考：
https://www.cnblogs.com/nullllun/p/9039049.html
https://blog.csdn.net/a67474506/article/details/48310515