併發 - CAS 的操做、實現、原理及優化

簡介

在 Java 中不少工具類都在使用 CAS（Compare And Set）用以提高併發的效率以及數據的準確性質。

• concurrent 和 concurrent.atomic 下面的不少 AtomicInteger 等類
• concurrent.locks 包下面的 ReentrantLock 、WriteLock 等
• 其它

對於大部分人來講，最多見的應該就是使用 AtomicXXX、以及在使用 Lock 相關的子類 的時候咱們知道他們的底層運用了 CAS，也知道 CAS 就是傳入一個更新前得期待值（expect）和一個須要更新的值（update），若是知足要求那麼執行更新，不然的話就算執行失敗，來達到數據的原子性。

咱們知道 CAS 確定用某一種方式在底層保證了數據的原子性，它的好處是

• 沒必要作同步阻塞的掛起以及喚醒線程這樣大量的開銷
• 將保證數據原子性的這個操做交給了底層硬件性能遠遠高於作同步阻塞掛起、喚醒等操做，因此它的併發性更好
• 能夠根據 CAS 返回的狀態決定後續操做來達到數據的一致性，好比 increment 失敗那就一值循環直到成功爲止（下文會講）等等

首先來看一個錯誤的 increment()

private int value = 0;

    public static void main(String[] args) {
        Test test = new Test();
        test.increment();
        System.out.println("期待值：" + 100 * 100 + "，最終結果值：" + test.value);
    }

    private void increment() {
        for (int i = 0; i < 100; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    value++;
                }
            }).start();
        }
    }
複製代碼

輸出：期待值：10000，最終結果值：9900

能夠發現輸出的結果值錯誤，這是由於 value++ 不是一個原子操做，它將 value++ 拆分紅了 3 個步驟 load、add、store，多線程併發有可能上一個線程 add 事後尚未 store 下一個線程又執行了 load 了這種重複形成獲得的結果可能比最終值要小。

固然在這裏加 volatile int value 也是沒有用的由於 32 位的 int 操做自己就是原子的，並且 volatile 也沒有辦法讓這 3 個操做原子性執行，它只能禁止某個指令重排序來保證其對應的內存可見，若是是 long 等 64 位操做類型的能夠加上 volatile，由於在 32 位的機器上寫操做可能會被分配到不一樣的總線事務上去操做（能夠想象成分紅了 2 步操做，第一步操做前 32 位後一步操做後 32 位），而總線事務的執行是由總線仲裁決定的不能保證它的執行順序（至關於前者加了 32 位可能就切換到其它的地方執行了，好比直接就讀取了，那麼數據的讀取就只讀取到了寫入一半的值）

使用 CAS 來保證 increment() 正確

咱們知道關於 CAS 的操做基本上都封裝在 Unsafe 這個包裏面，可是因爲 Unsafe 不容許咱們外部使用，它認爲這是一個不安全的操做，好比若是直接使用 Unsafe unsafe = Unsafe.getUnsafe(); 就會拋出 Exception in thread "main" java.lang.SecurityException: Unsafe。

咱們查看下源代碼，原來是由於它作了校驗

public static Unsafe getUnsafe() {
        Class var0 = Reflection.getCallerClass();
        if (!VM.isSystemDomainLoader(var0.getClassLoader())) {
            throw new SecurityException("Unsafe");
        } else {
            return theUnsafe;
        }
    }
複製代碼

因此咱們能夠經過反射來調用它（固然實際操做中不建議這麼使用，此處爲了演示方便）

public class Test {

    // value 的內存地址，便於直接找到 value
    private static long valueOffset = 0;

    {
        try {
            // 這個內存地址是和 value 這個成員變量的值綁定在一塊兒的
            valueOffset = getUnsafe().objectFieldOffset
                (Test.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }
    
    private int value;

    public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException {
        Test test = new Test();
        test.increment();
    }

    private void increment() throws NoSuchFieldException, IllegalAccessException {
        Unsafe unsafe = getUnsafe();
        for (int i = 0; i < 100; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    unsafe.getAndAddInt(this, valueOffset, 1);
                }
            }).start();
        }
        System.out.println("須要獲得的結果爲: " + 100 * 1000);
        System.out.println("實際獲得的結果爲: " + value);
    }

    // 反射獲取 Unsafe
    private Unsafe getUnsafe() throws NoSuchFieldException, IllegalAccessException {
        Field field = Unsafe.class.getDeclaredField("theUnsafe");
        field.setAccessible(true);
        return (Unsafe) field.get(null);
    }
}
複製代碼

這下咱們就能從輸出中看到結果是正確的了

CAS 底層的實現原理

咱們繼續探討， getAndAddInt 調用了 unsafe.compareAndSwapInt(Object obj, long valueOffset, int expect, int update) 這個方法在 Hotspot 究竟是如何實現的，咱們發現調用的是 native 的 unsafe.compareAndSwapInt(Object obj, long valueOffset, int expect, int update)，咱們翻看 Hotspot 源碼發如今 unsafe.cpp 中定義了這樣一段代碼

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);
  jint* addr = (jint *) index_oop_from_field_offset_long(p, offset);
  return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
UNSAFE_END
複製代碼

從中咱們能夠看到它是使用了 Atomic::cmpxchg(x, addr, e) 這個操做來完成的，在不一樣的底層硬件會有不同的代碼 Hotspot 向上幫咱們屏蔽了細節。這個實現方法在 solaris，windows，linux_x86 等都有不同的實現方法，咱們用咱們最多見的服務器 linux_x86 來講，它的實現代碼以下

inline jint Atomic::cmpxchg (jint exchange_value, volatile jint*  dest, 
jint compare_value) {
  int mp = os::is_MP();
  __asm__ volatile (LOCK_IF_MP(%4) "cmpxchgl %1,(%3)"
                    : "=a" (exchange_value)
                    : "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
                    : "cc", "memory");
  return exchange_value;
}
複製代碼

從以上代碼能夠看出幾點

• Hotspot 直接調用底層彙編來實現對應的功能
• __asm__ 表示的是後續是一段彙編代碼
• volatile 此處的 volatile 和 Java 中的有些區別，這裏使用用以告訴編譯器再也不對這段代碼進行彙編優化
• LOCK_IF_MP 表示的是若是操做系統是多核的那麼就須要加鎖來保證其原子性
• cmpxchgl 就是彙編中的比較而且交換

從這裏就能看出來，CAS 底層也是在用鎖來保證其原子性的。在 Intel 早期的實現中是直接將總線鎖住，這樣致使其它沒有得到總線事務訪問權的處理器沒法執行後續的操做，性能會極大的下降。

後續 Intel 對其進行了優化升級，在 x86 處理器中能夠只須要鎖定 特定的內存地址，那麼其它處理器也就能夠繼續使用總線來訪問內存數據了，只不過是若是其它總線也要訪問被鎖住的內存地址數據時會阻塞而已，這樣來大幅度的提高了性能。

可是思考一下如下幾點問題的

1. 併發量很是高，可能致使都在不停的爭搶該值，可能致使不少線程一致處於循環狀態而沒法更新數據，從而致使 CPU 資源的消耗太高
2. ABA 問題，好比說上一個線程增長了某個值，又改變了某個值，而後後面的線程覺得數據沒有發生過變化，其實已經被改動了

JAVA8 對於 CAS 的優化

固然 ABA 的問題可使用增長版本號來控制，每次操做版本號 + 1，版本號變動了說明值就被改過一次了，在 Java 中 AtomicStampedReference 這個類提供了這種問題的解決方案。

而對於說第一個問題來講在 Java8 中也有了對應的優化，Java 8 中提供了一些新的工具類用以解決這種問題，以下

咱們挑一個來看，其它都是相似的

能夠看到他是能夠序列化的，而且必須是 Number 類型的，繼承 Striped64 可以支持動態的分段

它的原理主要採用CAS分段機制與自動分段遷移機制，最開始是在 base 上面進行 CAS 操做，後續併發線程過多，那麼就將這大量的線程分配到 cells 數組中去，每一個數組的線程單獨去執行累加操做，最終再合併結果

圖來自【基礎鞏固篇】Java 8中對CAS的優化

總結

能夠看到跟作直接作同步掛起或者喚醒線程相好比果可以合理的使用 CAS 進行操做的話或者是將其兩者合併使用，那麼在併發性能上可以提高一個量級

• 對於像 ReentrantLock 之類的都是使用的將同步阻塞 + CAS 這種方式來實現高性能的鎖，好比 ReentrantLock 中 tryAcuqire() 若是使用 CAS 未能獲取到對應的鎖，那麼就將其放入阻塞隊列，等待後續的喚醒
• 好比自旋鎖在指定的次數經過 CAS 都未能獲取到鎖的話就掛起進入阻塞隊列等待被喚醒
• 好比使用 AtomicInteger 進行自增的時候就會一值不停的輪詢判斷更新，直到操做成功爲止
• 使用輪詢 CAS 處理而不嵌入阻塞掛起和喚醒的話，它的優點就是在於可以快速響應用戶請求減小資源消耗，由於線程的掛起和喚醒涉及到用戶態內核態的調用又涉及到線程「快照」數據的相關保存，對於響應和資源消耗是又慢又高，不過咱們也須要考慮在 CPU 輪詢上的開銷，因此能夠將兩者必定程度上的融合在一塊兒使用。
• 因此理解 CAS 仍是很是重要的

參考： JAVA 中的 CAS