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