CAS的全稱爲Compare And Swap,直譯就是比較交換。是一條CPU的原子指令,其做用是讓CPU先進行比較兩個值是否相等,而後原子地更新某個位置的值,其實現方式是基於硬件平臺的彙編指令,在intel的CPU中,使用的是cmpxchg
指令,就是說CAS是靠硬件實現的,從而在硬件層面提高效率。html
利用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
// 使用 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 語言能夠經過偏移量對變量進行操做bash
//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
源碼解析多線程
在JDK8中追蹤可見sun.misc.Unsafe
這個類是沒法看見源碼的,打開openjdk8
源碼看併發
文件: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類提供瞭如下這些功能:
//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_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(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 cmp(A) == 當前內存存偏移量位置 Pair(V),就更新值 Pair 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實例進行更新。