刨根問底 Java 併發之 CAS

前言

後端開發中你們確定遇到過實現一個線程安全的計數器這種需求，根據經驗你應該知道咱們要在多線程中實現 「共享變量」 的原子性和可見性問題，因而鎖成爲一個不可避免的話題，今天咱們討論的是與之對應的無鎖 CAS。本文會從怎麼來的、是什麼、怎麼用、原理分析、遇到的問題等不一樣的角度帶你真正搞懂 CAS。

爲何要無鎖

咱們一想到在多線程下保證安全的方式頭一個要拎出來的確定是鎖，無論從硬件、操做系統層面都或多或少在使用鎖。鎖有什麼缺點嗎？固然有了，否則 JDK 裏爲何出現那麼多各式各樣的鎖，就是由於每一種鎖都有其優劣勢。

使用鎖就須要得到鎖、釋放鎖，CPU 須要經過上下文切換和調度管理來進行這個操做，對於一個 「獨佔鎖」 而言一個線程在持有鎖後沒執行結束其餘的哥們就必須在外面等着，等到前面的哥們執行完畢 CPU 大哥就會把鎖拿出來其餘的線程來搶了（非公平）。鎖的這種概念基於一種悲觀機制，它老是認爲數據會被修改，因此你在操做一部分代碼塊以前先加一把鎖，操做完畢後再釋放，這樣就安全了。其實在 JDK1.5 使用 synchronized就能夠作到。

可是像上面的操做在多線程下會讓 CPU 不斷的切換，很是消耗資源，咱們知道可使用具體的某一類鎖來避免部分問題。那除了鎖的方式還有其餘的嗎？固然，有人就提出了無鎖算法，比較有名的就是咱們今天要說的 CAS（compare and swap），和鎖不一樣的是它是一種樂觀的機制，它認爲別人去拿數據的時候不會修改，可是在修改數據的時候去判斷一下數據此時的狀態，這樣的話 CPU 不會切換，在讀多的狀況下性能將獲得大幅提高。當前咱們使用的大部分 CPU 都有 CAS 指令了，從硬件層面支持無鎖，這樣開發的時候去調用就能夠了。

不管是鎖仍是無鎖都有其優劣勢，後面咱們也會經過例子說明 CAS 的問題。

什麼是CAS?

前面提了無鎖的 CAS，那到底 CAS 是個啥呢？我已經火燒眉毛了，咱們來看看維基百科的解釋

比較並交換(compare and swap, CAS)，是原子操做的一種，可用於在多線程編程中實現不被打斷的數據交換操做，從而避免多線程同時改寫某一數據時因爲執行順序不肯定性以及中斷的不可預知性產生的數據不一致問題。該操做經過將內存中的值與指定數據進行比較，當數值同樣時將內存中的數據替換爲新的值。

CAS 給咱們提供了一種思路，經過 「比較」 和 「替換」 來完成原子性，來看一段代碼：

`int cas(long *addr, long old, long new) {`
 `/* 原子執行 */`
 `if(*addr != old)`
 `return 0;`
 `*addr = new;`
 `return 1;`
`}`

這是一段 c 語言代碼，能夠看到有 3 個參數，分別是：

• *addr: 進行比較的值
• old: 內存當前值
• new: 準備修改的新值，寫入到內存

只要咱們當前傳入的進行比較的值和內存裏的值相等，就將新值修改爲功，不然返回 0 告訴比較失敗了。學過數據庫的同窗都知道悲觀鎖和樂觀鎖，樂觀鎖老是認爲數據不會被修改。基於這種假設 CAS 的操做也認爲內存裏的值和當前值是相等的，因此操做老是能成功，咱們能夠不須要加鎖就實現多線程下的原子性操做。

在多線程狀況下使用 CAS 同時更新同一個變量時，只有其中一個線程能更新變量的值，而其它線程都失敗，失敗的線程並不會被阻塞掛起，而是告訴它此次修改失敗了，你能夠從新嘗試，因而能夠寫這樣的代碼。

`while (!cas(&addr, old, newValue)) {`
`}`
`// success`
`printf("new value = %ld", addr);`

不過這樣的代碼相信你可能看出其中的蹊蹺了，這個咱們後面來分析，下面來看看 Java 裏是怎麼用 CAS 的。

Java 中的CAS

仍是前面的問題，若是讓你用 Java 的 API 來實現你可能會想到兩種方式，一種是加鎖（多是 synchronized 或者其餘種類的鎖），另外一種是使用 atomic 類，如 AtomicInteger，這一系列類是在 JDK1.5 的時候出現的，在咱們經常使用的 java.util.concurrent.atomic 包下，咱們來看個例子：

`ExecutorService executorService = Executors.newCachedThreadPool();`
`AtomicInteger   atomicInteger   = new AtomicInteger(0);`
`for (int i = 0; i < 5000; i++) {`
 `executorService.execute(atomicInteger::incrementAndGet);`
`}`
`System.out.println(atomicInteger.get());`
`executorService.shutdown();`

這個例子開啓了 5000 個線程去進行累加操做，無論你執行多少次答案都是 5000。這麼神奇的操做是如何實現的呢？就是依靠 CAS 這種技術來完成的，咱們揭開 AtomicInteger 的老底看看它的代碼：

`public class AtomicInteger extends Number implements java.io.Serializable {`
 `private static final long serialVersionUID = 6214790243416807050L;`
 `// setup to use Unsafe.compareAndSwapInt for updates`
 `private static final Unsafe unsafe = Unsafe.getUnsafe();`
 `private static final long valueOffset;`
 `static {`
 `try {`
 `valueOffset = unsafe.objectFieldOffset`
 `(AtomicInteger.class.getDeclaredField("value"));`
 `} catch (Exception ex) { throw new Error(ex); }`
 `}`
 `private volatile int value;`
 `/**`
 `* Creates a new AtomicInteger with the given initial value.` 
 `* @param initialValue the initial value`
 `*/`
 `public AtomicInteger(int initialValue) {`
 `value = initialValue;`
 `}`
 `/**`
 `* Gets the current value.` 
 `* @return the current value`
 `*/`
 `public final int get() {`
 `return value;`
 `}`
 `/**`
 `* Atomically increments by one the current value.` 
 `* @return the updated value`
 `*/`
 `public final int incrementAndGet() {`
 `return unsafe.getAndAddInt(this, valueOffset, 1) + 1;`
 `}`
`}`

這裏我只帖出了咱們前面例子相關的代碼，其餘都是相似的，能夠看到 incrementAndGet 調用了 unsafe.getAndAddInt 方法。Unsafe 這個類是 JDK 提供的一個比較底層的類，它不讓咱們程序員直接使用，主要是怕操做不當把機器玩壞了。。。（其實能夠經過反射的方式獲取到這個類的實例）你會在 JDK 源碼的不少地方看到這傢伙，咱們先說說它有什麼能力：

• 內存管理：包括分配內存、釋放內存
• 操做類、對象、變量：經過獲取對象和變量偏移量直接修改數據
• 掛起與恢復：將線程阻塞或者恢復阻塞狀態
• CAS：調用 CPU 的 CAS 指令進行比較和交換
• 內存屏障：定義內存屏障，避免指令重排序

這裏只是大體提一下經常使用的操做，具體細節能夠在文末的參考連接中查看。下面咱們繼續看 unsafe 的 getAndAddInt 在作什麼。

`public final int getAndAddInt(Object var1, long var2, int var4) {`
 `int var5;`
 `do {`
 `var5 = this.getIntVolatile(var1, var2);`
 `} while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));`
 `return var5;`
`}`
`public native int getIntVolatile(Object var1, long var2);`
`public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);`

其實很簡單，先經過 getIntVolatile 獲取到內存的當前值，而後進行比較，展開 compareAndSwapInt 方法的幾個參數：

• var1: 當前要操做的對象（其實就是 AtomicInteger 實例）
• var2: 當前要操做的變量偏移量（能夠理解爲 CAS 中的內存當前值）
• var4: 指望內存中的值
• var5: 要修改的新值

因此 this.compareAndSwapInt(var1, var2, var5, var5 + var4) 的意思就是，比較一下 var2 和內存當前值 var5 是否相等，若是相等那我就將內存值 var5 修改成 var5 + var4（var4 就是 1，也能夠是其餘數）。

這裏咱們還須要解釋一下 「偏移量」 是個啥？你在前面的代碼中可能看到這麼一段：

`// setup to use Unsafe.compareAndSwapInt for updates`
`private static final Unsafe unsafe = Unsafe.getUnsafe();`
`private static final long valueOffset;`
`static {`
 `try {`
 `valueOffset = unsafe.objectFieldOffset`
 `(AtomicInteger.class.getDeclaredField("value"));`
 `} catch (Exception ex) { throw new Error(ex); }`
`}`
`private volatile int value;`

能夠看出在靜態代碼塊執行的時候將 AtomicInteger 類的 value 這個字段的偏移量獲取出來，拿這個 long 數據幹嗎呢？在 Unsafe 類裏不少地方都須要傳入 obj 和偏移量，結合咱們說 Unsafe 的諸多能力，其實就是直接經過更底層的方式將對象字段在內存的數據修改掉。

使用上面的方式就能夠很好的解決多線程下的原子性和可見性問題。因爲代碼裏使用了 do while 這種循環結構，因此 CPU 不會被掛起，比較失敗後重試，就不存在上下文切換了，實現了無鎖併發編程

CAS存在的問題

自旋的劣勢

你留意上面的代碼會發現一個問題，while 循環若是在最壞狀況下老是失敗怎麼辦？會致使 CPU 在不斷處理。像這種 while(!compareAndSwapInt) 的操做咱們稱之爲自旋，CAS 是樂觀的，認爲你們來並不都是修改數據的，現實可能出現很是多的線程過來都要修改這個數據，此時隨着併發量的增長會致使 CAS 操做長時間不成功，CPU 也會有很大的開銷。因此咱們要清楚，若是是讀多寫少的狀況也就知足樂觀，性能是很是好的。

ABA 問題

提到 CAS 不得不說 ABA 問題，它是說假如內存的值原來是 A，被一個線程修改成了 B，此時又有一個線程把它修改成了 A，那麼 CAS 確定是操做成功的。真的這樣作的話代碼可能就有 bug 了，對於修改數據爲 B 的那個線程它應該讀取到 B 而不是 A，若是你作過數據庫相關的樂觀鎖機制可能會想到咱們在比較的時候使用一個版本號 version 來進行判斷就能夠搞定。在 JDK 裏提供了一個 AtomicStampedReference 類來解決這個問題，來看一個例子：

`int stamp = 10001;`
`AtomicStampedReference<Integer> stampedReference = new AtomicStampedReference<>(0, stamp);`
`stampedReference.compareAndSet(0, 10, stamp, stamp + 1);`
`System.out.println("value: " + stampedReference.getReference());`
`System.out.println("stamp: " + stampedReference.getStamp());`

它的構造函數是 2 個參數，多傳入了一個初始 時間戳，用這個戳來給數據加了一個版本，這樣的話多個線程來修改若是提供的戳不一樣。在修改數據的時候除了提供一個新的值以外還要提供一個新的戳，這樣在多線程狀況下只要數據被修改了那麼戳必定會發生改變，另外一個線程拿到的是舊的戳因此會修改失敗。

嘗試應用

既然 CAS 提供了這麼好的 API，咱們不妨用它來實現一個簡易版的獨佔鎖。思路是當某個線程進入 lock 方法就比較鎖對象的內存值是不是 false，若是是則表明這把鎖它能夠獲取，獲取後將內存之修改成 true，獲取不到就自旋。在 unlock 的時候將內存值再修改成 false 便可，代碼以下：

`public class SpinLock {`
 `private AtomicBoolean mutex = new AtomicBoolean(false);`
 `public void lock() {`
 `while (!mutex.compareAndSet(false, true)) {`
 `// System.out.println(Thread.currentThread().getName()+ " wait lock release");`
 `}`
 `}`
 `public void unlock() {`
 `while (!mutex.compareAndSet(true, false)) {`
 `// System.out.println(Thread.currentThread().getName()+ " wait lock release");`
 `}`
 `}`
`}`

這裏使用了 AtomicBoolean 這個類，固然用 AtomicInteger 也是能夠的，由於咱們只保存一個狀態 boolean 佔用比較小就用它了。這個鎖的實現比較簡單，缺點很是明顯，因爲 while 循環致使的自旋會讓其餘線程都在佔用 CPU，可是也可使用，關於鎖的優化版本實現我會在後續的文章中進行改進和說明，正由於這些問題咱們也會在後續研究 AQS 這把利器的優勢。

CAS 源碼

看了上面的這些代碼和解釋相信你對 CAS 已經理解了，下面咱們要說的原理是前面的 native 方法中的 C++ 代碼寫了什麼，在 openjdk 的 /hotspot/src/share/vm/prims 目錄中有一個 Unsafe.cpp 文件中有這樣一段代碼：

注意：這裏以 hotspot 實現爲例

`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_ENDC`

好了，關於CAS今天就分享到這裏了，但願對你有所幫助。