詳解并發編程中的 CAS 原子類
作者：碼農SharkChili
CAS全稱Compare-And-Swap，是一種無鎖編程算法，即比對讀取值與內存中的值的差異決定是否進行修改操作的一種(樂觀鎖機制)，該工具類常用于多線程共享變量的修改操作。
CAS全稱Compare-And-Swap，是一種無鎖編程算法，即比對讀取值與內存中的值的差異決定是否進行修改操作的一種(樂觀鎖機制)，該工具類常用于多線程共享變量的修改操作。而其底層實現也是基于硬件平臺的匯編指令，JVM只是封裝其調用僅此而已。而本文會基于以下大綱展開對CAS的探討。
一、詳解CAS原子類
1. 原子類基礎使用示例
使用封裝CAS操作的AtomicInteger操作多線程共享變量無需我們手動加鎖，避免了人為上鎖粒度把控不足所導致的安全性問題：
AtomicInteger count = new AtomicInteger();
//并行流并發操作原子類
IntStream.rangeClosed(1, 100_0000).parallel()
.forEach(i -> count.incrementAndGet());
Console.log("輸出結果:{}", count);//10000002. 詳解原子類的底層實現
原子類的并發累加本質上采用了一種樂觀鎖機制，其底層對于利用一個volatile修飾的變量value存儲用戶當前讀取到的變量值，在進行并發操作的自增時，會執行如下步驟：
- 原子類就會利用unsafe類到內存中讀取該變量的最新修改結果并存儲到臨時變量v中
- 再次讀取變量的值并進行比對，兩者一致，則基于 v進行累加更新，并寫到value中
- 兩者不一致，說明volatile讀到的值是過期的，重新從步驟1開始執行，直到讀取到最新的
對應的我們也給出這段實現的源碼，可以看到AtomicInteger就是利用unsafe的getAndAddInt完成并發自增的：
//直接基于內存地址讀取變量的工具類 unsafe
private static final jdk.internal.misc.Unsafe U = jdk.internal.misc.Unsafe.getUnsafe();
//并發操作變量原子類對應的value地址偏移量
private static final long VALUE = U.objectFieldOffset(AtomicInteger.class, "value");
//調用unsafe完成變量value的自增更新
public final int incrementAndGet() {
return U.getAndAddInt(this, VALUE, 1) + 1;
}步入unsafe的邏輯實現，可以看到其底層實現就如上文所說：
- 利用volatile讀獲取并發變量最新值到v中
- 傳入原子類對象和偏移量地址再次讀取變量的最新值，兩者一致直接自增并寫入到內存中
- 若不一致，回到步驟1循環執行，直到成功
@HotSpotIntrinsicCandidate
public final int getAndAddInt(Object o, long offset, int delta) {
int v;
do {
//利用volatile讀獲取最新值
v = getIntVolatile(o, offset);
} while (!weakCompareAndSetInt(o, offset, v, v + delta));//基于偏移量再次讀取value的最新值，如果一致則更新
return v;
}3. 手寫一個原子類
基于上述的源碼分析，不難看出原子類本質上就是通過比對本次操作變量的值和最新的值是否一致以判斷是否出現并發修改的情況，所以按照這個理念，我們也可以通過unsafe手寫一個原子類,大體步驟為：
- 定義一個原子類
- 初始化unsafe
- 編寫一個變量count記錄原子更新的值
- 編寫一個并發自增的方法通過拉取當前讀取的count值和最新count進行比對，一致再執行更新的操作
對應的我們也給出本次手寫的初始化unsafe和定位字段偏移量的代碼段,可以看到筆者通過反射的方式完成unsafe工具類的獲取和初始化，完成該操作后直接基于unsafe類定位到count變量的偏移地址，方便后續快速定位和比對變量值：
// 獲取Unsafe對象
private static Unsafe unsafe;
// 自增的count的值，volatile保證可見性
private volatile int count = 0;
// count字段的偏移量
private static long countOffSet;
static {
try {
//初始化unsafe
try {
Field field = Unsafe.class.getDeclaredField("theUnsafe");
field.setAccessible(true);
unsafe = (Unsafe) field.get(null);
} catch (Exception e) {
Console.log("獲取unsafe失敗，失敗原因:[{}]", e.getMessage(), e);
}
//初始化定位count變量的偏移量地址
countOffSet = unsafe.objectFieldOffset(CasCountInc.class.getDeclaredField("count"));
} catch (NoSuchFieldException e) {
Console.log("獲取count的偏移量報錯，錯誤原因:[{}]", e.getMessage(), e);
}對應的自增代碼如下,可以看到該邏輯即直接通過volatile讀獲取當前count的值，然后再進行cas操作時傳入count偏移量地址讓unsafe的compareAndSwapInt比對最新的count值和我們讀取的oldCount是否一致以決定是否更新：
public void inc() {
int oldCount = 0;
//基于cas完成自增
do {
//拉取本次的值
oldCount = count;
//通過樂觀鎖的方式比對舊有的值和偏移量中的新值是否一致，將值進行更新并設置到count中
} while (!unsafe.compareAndSwapInt(this, countOffSet, oldCount, oldCount + 1));
}完整的代碼如下所示：
public class CasCountInc {
// 獲取Unsafe對象
private static Unsafe unsafe;
// 自增的count的值，volatile保證可見性
private volatile int count = 0;
// count字段的偏移量
private static long countOffSet;
static {
try {
//初始化unsafe
try {
Field field = Unsafe.class.getDeclaredField("theUnsafe");
field.setAccessible(true);
unsafe = (Unsafe) field.get(null);
} catch (Exception e) {
Console.log("獲取unsafe失敗，失敗原因:[{}]", e.getMessage(), e);
}
//初始化定位count變量的偏移量地址
countOffSet = unsafe.objectFieldOffset(CasCountInc.class.getDeclaredField("count"));
} catch (NoSuchFieldException e) {
Console.log("獲取count的偏移量報錯，錯誤原因:[{}]", e.getMessage(), e);
}
}
public void inc() {
int oldCount = 0;
//基于cas完成自增
do {
//拉取本次的值
oldCount = count;
//通過樂觀鎖的方式比對舊有的值和偏移量中的新值是否一致，將值進行更新并設置到count中
} while (!unsafe.compareAndSwapInt(this, countOffSet, oldCount, oldCount + 1));
}
}對應的我們也給出測試單元的代碼，感興趣的讀者可以自行運行一下，發現結果確實是100w：
CasCountInc casCountInc = new CasCountInc();
IntStream.range(0, 100_0000).parallel()
.forEach(i -> casCountInc.inc());
Assert.equals(casCountInc.count, 100_0000);二、詳解更多原子類
1. 原子類更新基本類型
原子類基本類型的格式為Atomic+包裝類名，這里筆者列舉幾個比較常用的：
- AtomicBoolean: 原子更新布爾類型。
- AtomicInteger: 原子更新整型。
- AtomicLong: 原子更新長整型。
2. 原子類更新數組類型
- AtomicIntegerArray: 原子更新整型數組里的元素。
- AtomicLongArray: 原子更新長整型數組里的元素。
- AtomicReferenceArray: 原子更新引用類型數組里的元素。
對應我們給出AtomicIntegerArray原子操作數組的示例：
public class AtomicIntegerArrayDemo {
public static void main(String[] args) throws InterruptedException {
AtomicIntegerArray array = new AtomicIntegerArray(new int[] { 0, 0 });
System.out.println(array);
// 索引1位置+2
System.out.println(array.getAndAdd(1, 2));
System.out.println(array);
}
}3. 原子類更新引用類型
- AtomicReference: 原子更新引用類型。
- AtomicStampedReference: 原子更新引用類型, 內部使用Pair來存儲元素值及其版本號。
- AtomicMarkableReferce: 原子更新帶有標記位的引用類型。
對應的我們給出原子操作引用類型的代碼示例：
import java.util.concurrent.atomic.AtomicReference;
public class AtomicReferenceTest {
public static void main(String[] args){
// 創建兩個Person對象，它們的id分別是101和102。
Person p1 = new Person(101);
Person p2 = new Person(102);
// 新建AtomicReference對象，初始化它的值為p1對象
AtomicReference ar = new AtomicReference(p1);
// 通過CAS設置ar。如果ar的值為p1的話，則將其設置為p2。
ar.compareAndSet(p1, p2);
Person p3 = (Person)ar.get();
System.out.println("p3 is "+p3);
System.out.println("p3.equals(p1)="+p3.equals(p1));
System.out.println("p3.equals(p2)="+p3.equals(p2));
}
}
class Person {
volatile long id;
public Person(long id) {
this.id = id;
}
public String toString() {
return "id:"+id;
}
}4. 原子類更新成員變量
通過原子類型操作成員變量大體有以下幾個更新器：
- AtomicIntegerFieldUpdater: 原子更新整型的字段的更新器。
- AtomicLongFieldUpdater: 原子更新長整型字段的更新器。
- AtomicStampedFieldUpdater: 原子更新帶有版本號的引用類型
- AtomicReferenceFieldUpdater: 上面已經說過此處不在贅述。
如下所示，我們創建一個基礎類DataDemo，通過原子類CAS操作字段值進行自增操作。
public class TestAtomicIntegerFieldUpdater {
private static Logger logger = LoggerFactory.getLogger(TestAtomicIntegerFieldUpdater.class);
public static void main(String[] args) {
TestAtomicIntegerFieldUpdater tIA = new TestAtomicIntegerFieldUpdater();
tIA.doIt();
}
/**
* 返回需要更新的整型字段更新器
*
* @param fieldName
* @return
*/
public AtomicIntegerFieldUpdater<DataDemo> updater(String fieldName) {
return AtomicIntegerFieldUpdater.newUpdater(DataDemo.class, fieldName);
}
public void doIt() {
DataDemo data = new DataDemo();
// 修改公共變量，返回更新前的舊值 0
AtomicIntegerFieldUpdater<DataDemo> updater = updater("publicVar");
int oldVal = updater.getAndIncrement(data);
logger.info("publicVar 更新前的值[{}] 更新后的值 [{}]", oldVal, data.publicVar);
// 更新保護級別的變量
AtomicIntegerFieldUpdater<DataDemo> protectedVarUpdater = updater("protectedVar");
int oldProtectedVar = protectedVarUpdater.getAndAdd(data, 2);
logger.info("protectedVar 更新前的值[{}] 更新后的值 [{}]", oldProtectedVar, data.protectedVar);
// logger.info("privateVar = "+updater("privateVar").getAndAdd(data,2)); 私有變量會報錯
/*
* 下面報異常：must be integer
* */
// logger.info("integerVar = "+updater("integerVar").getAndIncrement(data));
//logger.info("longVar = "+updater("longVar").getAndIncrement(data));
}
class DataDemo {
// 公共且可見的publicVar
public volatile int publicVar = 0;
// 保護級別的protectedVar
protected volatile int protectedVar = 4;
// 私有變量
private volatile int privateVar = 5;
// final 不可變量
public final int finalVar = 11;
public volatile Integer integerVar = 19;
public volatile Long longVar = 18L;
}
}通過上述代碼我們可以總結出CAS字段必須符合以下要求：
- 變量必須使用volatile保證可見性
- 必須是當前對象可以訪問到的類型才可進行操作‘
- 只能是實例變量而不是類變量，即不可以有static修飾符
- 包裝類也不行
三、詳解CAS的ABA問題
1. 什么是ABA問題
CAS更新是一種樂觀鎖機制，所以在更新前會檢查值有沒有變化，如果沒有變化則認為沒人修改過，進而執行更新操作。在這種情況下我們試想這樣一個場景，我們現在希望完成并發情況的數字操作：
- 線程0將數值由0改為1，再由1改為0，按照正常的邏輯理解，本次數值發生變化了2次
- 線程1開始執行，發現數值是0，認為沒有發生變化，CAS成功，數值直接變化3:
對應的我們給出入下代碼示例：
AtomicReference<Integer> atomicReference = new AtomicReference<>(0);
new Thread(() -> {
//1.線程0將0改為1，再還原回0
atomicReference.compareAndSet(0, 1);
atomicReference.compareAndSet(1, 0);
}).start();
new Thread(() -> {
ThreadUtil.sleep(1000);
//2. 線程1嘗試將0改為3，發現是0直接修改
atomicReference.compareAndSet(0, 3);
}).start();
Console.log("value:{}", atomicReference.get());2. AtomicStampedReference如何解決ABA問題
源碼如下所示，可以看到AtomicStampedReference解決ABA問題的方式是基于當前修改操作的時間戳和元引用值是否一致，若一直則進行數據更新
public class AtomicStampedReference<V> {
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);
}
}
private volatile Pair<V> pair;
....
/**
* expectedReference ：更新之前的原始引用值
* newReference : 新值
* expectedStamp : 預期時間戳
* newStamp : 更新后的時間戳
*/
public boolean compareAndSet(V expectedReference,
V newReference,
int expectedStamp,
int newStamp) {
// 獲取當前的(元素值，版本號)對
Pair<V> current = pair;
return
// 引用沒變
expectedReference == current.reference &&
// 版本號沒變
expectedStamp == current.stamp &&
//可以看到這個括號里面用了一個短路運算如果當前版本與新值一樣就說更新過，就不往下走CAS代碼了
((newReference == current.reference &&
newStamp == current.stamp) ||
// 構造新的Pair對象并CAS更新
casPair(current, Pair.of(newReference, newStamp)));
}
private boolean casPair(Pair<V> cmp, Pair<V> val) {
// 調用Unsafe的compareAndSwapObject()方法CAS更新pair的引用為新引用
return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
}3. AtomicStampedReference解決ABA問題示例
代碼示例，我們下面就用other代碼模擬干擾現場，如果other現場先進行CAS更新再還原操作，那么main線程的版本號就會過時，CAS就會操作失敗
/**
* ABA問題代碼示例
*/
public class AtomicStampedReferenceTest {
private static AtomicStampedReference<Integer> atomicStampedRef =
new AtomicStampedReference<>(1, 0);
public static void main(String[] args) {
Thread main = new Thread(() -> {
System.out.println("操作線程" + Thread.currentThread() + ",初始值 a = " + atomicStampedRef.getReference());
int stamp = atomicStampedRef.getStamp(); //獲取當前標識別
try {
Thread.sleep(1000); //等待1秒 ，以便讓干擾線程執行
} catch (InterruptedException e) {
e.printStackTrace();
}
boolean isCASSuccess = atomicStampedRef.compareAndSet(1, 2, stamp, stamp + 1); //此時expectedReference未發生改變，但是stamp已經被修改了,所以CAS失敗
System.out.println("操作線程" + Thread.currentThread() + ",CAS操作結果: " + isCASSuccess);
}, "主操作線程");
Thread other = new Thread(() -> {
Thread.yield(); // 確保thread-main 優先執行
atomicStampedRef.compareAndSet(1, 2, atomicStampedRef.getStamp(), atomicStampedRef.getStamp() + 1);
System.out.println("操作線程" + Thread.currentThread() + ",【increment】 ,值 = " + atomicStampedRef.getReference());
atomicStampedRef.compareAndSet(2, 1, atomicStampedRef.getStamp(), atomicStampedRef.getStamp() + 1);
System.out.println("操作線程" + Thread.currentThread() + ",【decrement】 ,值 = " + atomicStampedRef.getReference());
}, "干擾線程");
main.start();
other.start();
}
}4. AtomicMarkableReference解決對象ABA問題
AtomicMarkableReference，它不是維護一個版本號，而是維護一個boolean類型的標記，標記對象是否有修改，從而解決ABA問題。
public boolean weakCompareAndSet(V expectedReference,
V newReference,
boolean expectedMark,
boolean newMark) {
return compareAndSet(expectedReference, newReference,
expectedMark, newMark);
}四、常見面試題
1. CAS為什么比synchronized快(重點)
CAS工作原理是基于樂觀鎖且操作是原子性的，與synchronized的悲觀鎖(底層需要調用操作系統的mutex鎖)相比，效率也會相對高一些。
2. CAS是不是操作系統執行的？(重點)
不是，CAS是主要是通過處理器的指令來保證原子性的，在上面的講解中我們都知道CAS操作底層都是調用Unsafe的native修飾的方法，以AtomicInteger為例對應的底層的實現是Unsafe的compareAndSwapInt：
public final native boolean compareAndSwapInt(Object o, long offset,
int expected,
int x);對應的我們給出這段代碼的c語言實現，即位于:https://github.com/openjdk/jdk/blob/jdk8-b01/hotspot/src/share/vm/prims/unsafe.cpp的unsafe.cpp:
可以看到出去前兩個形參后續的參數與compareAndSwapInt列表一一對應，這段代碼執行CAS操作時，本質上就是調用cmpxchg指令(Compare and Exchange)，cmpxchg指令會判斷當前服務器是否是多核，如果是則在指令前添加LOCK前綴保證cmpxchg操作的原子性，反之就不加Lock前綴直接執行比對后修改變量值這種樂觀鎖操作。
對應源碼如下，它首先獲取字段的偏移地址，然后傳入預期值e與原值比較，如果一致，則將新結果x寫入原子操作變量內存中：
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);
//比較如果期望值e和當前字段存儲的值一樣，則講值更新為x
return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
UNSAFE_END3. CAS存在那些問題?
但即便如此CAS仍然存在兩個問題：
- 可能存在長時間CAS：如下代碼所示，這就是AtomicInteger底層的UNSAFE類如何進行CAS的具體代碼 ，可以看出這個CAS操作需要拿到volatile變量后在進行循環CAS才有可能成功這就很可能存在自旋循環，從而給CPU帶來很大的執行開銷。
public final int getAndAddInt(Object paramObject, long paramLong, int paramInt)
{
int i;
do
//獲取最新結果
i = getIntVolatile(paramObject, paramLong);
//通過cas自旋操作完成自增
while (!compareAndSwapInt(paramObject, paramLong, i, i + paramInt));
return i;
}- CAS只能對一個變量進行原子操作：為了解決這個問題，JDK 1.5之后通過AtomicReference使得變量可以封裝成一個對象進行操作
ABA問題：總所周知CAS就是比對當前值與舊值是否相等，在進行修改操作，假設我現在有一個變量值為A，我改為B，再還原為A，這樣操作變量值是沒變的？那么CAS也會成功不就不合理嗎？這就好比一個銀行儲戶想查詢概念轉賬記錄，如果轉賬一次記為1，如果按照ABA問題的邏輯，那么這個銀行賬戶轉賬記錄次數有可能會缺少。為了解決這個問題JDK 1.5提供了AtomicStampedReference，通過比對版本號在進行CAS操作，那么上述操作就會變為1A->2B->3A,由于版本追加，那么我們就能捕捉到當前變量的變化了。
4. AtomicInteger自增到10000后如何歸零
AtomicInteger atomicInteger=new AtomicInteger(10000);
atomicInteger.compareAndSet(10000, 0);CAS 平時怎么用的，會有什么問題，為什么快，如果我用 for 循環代替 CAS 執行效率是一樣的嗎？(重點)
問題1： 一些需要并發計數并實時監控的場景可以用到。 問題2： CAS存在問題：CAS是基于樂觀鎖機制，所以數據同步失敗就會原地自旋，在高并發場景下開銷很大，所以線程數很大的情況下不建議使用原子類。 問題3：用 for 循環代替 CAS 存在問題: 如果并發量大的話，自旋的線程多了就會導致性能瓶頸。 for 循環代替 CAS執行效率是否一樣:大概率是CAS快，原因如下:
5. CAS是native方法更接近底層
for循環為了保證線程安全可能會用到sync鎖或者Lock無論那種都需要上鎖和釋放的邏輯，相比CAS樂觀鎖來說開銷很大。
責任編輯：趙寧寧
來源：
寫代碼的SharkChili
