并发编程CAS机制原理分析及ABA问题解决

一、为什么需要CAS机制&＃xff1f;

为什么需要CAS机制呢&＃xff1f;我们先从一个错误现象谈起。我们经常使用volatile关键字修饰某一个变量&＃xff0c;表明这个变量是全局共享的一个变量&＃xff0c;同时具有了可见性和有序性。但是却没有原子性。比如说一个常见的操作a&＃43;&＃43;。这个操作其实可以细分成三个步骤&＃xff1a;

&＃xff08;1&＃xff09;从内存中读取a

&＃xff08;2&＃xff09;对a进行加1操作

&＃xff08;3&＃xff09;将a的值重新写入内存中

在单线程状态下这个操作没有一点问题&＃xff0c;但是在多线程中就会出现各种各样的问题了。因为可能一个线程对a进行了加1操作&＃xff0c;还没来得及写入内存&＃xff0c;其他的线程就读取了旧值。造成了线程的不安全现象。如何去解决这个问题呢&＃xff1f;最常见的方式就是使用AtomicInteger来修饰a。我们可以看一下代码&＃xff1a;

public class Test3 {//使用AtomicInteger定义astatic AtomicInteger a &＃61; new AtomicInteger();public static void main(String[] args) {Test3 test &＃61; new Test3();Thread[] threads &＃61; new Thread[5];for (int i &＃61; 0; i <5; i&＃43;&＃43;) {threads[i] &＃61; new Thread(() -> {try {for (int j &＃61; 0; j <10; j&＃43;&＃43;) {//使用getAndIncrement函数进行自增操作System.out.println(a.incrementAndGet()); Thread.sleep(500);}} catch (Exception e) {e.printStackTrace();}});threads[i].start();}}
}

现在我们使用AtomicInteger类并且调用了incrementAndGet方法来对a进行自增操作。这个incrementAndGet是如何实现的呢&＃xff1f;我们可以看一下AtomicInteger的源码。
 

/*** Atomically increments by one the current value.* &＃64;return the updated value*/public final int incrementAndGet() {return unsafe.getAndAddInt(this, valueOffset, 1) &＃43; 1;}

我们到这一步可以看到其实就是usafe调用了getAndAddInt的方法实现的&＃xff0c;但是现在我们还看不出什么&＃xff0c;我们再深入到源码中看看getAndAddInt方法又是如何实现的&＃xff0c;

public final int getAndAddInt(Object var1, long var2, int var4) { int var5; do { var5 &＃61; this.getIntVolatile(var1, var2); } while(!this.compareAndSwapInt(var1, var2, var5, var5 &＃43; var4)); return var5;
}

到了这一步就稍微有点眉目了&＃xff0c;原来底层调用的是compareAndSwapInt方法&＃xff0c;这个compareAndSwapInt方法其实就是CAS机制。因此如果我们想搞清楚AtomicInteger的原子操作是如何实现的&＃xff0c;我们就必须要把CAS机制搞清楚&＃xff0c;这也是为什么我们需要掌握CAS机制的原因。

二、分析CAS

1、基本含义

CAS全拼又叫做compareAndSwap&＃xff0c;从名字上的意思就知道是比较交换的意思。比较交换什么呢&＃xff1f;

过程是这样&＃xff1a;它包含 3 个参数 CAS&＃xff08;V&＃xff0c;E&＃xff0c;N&＃xff09;&＃xff0c;V表示要更新变量的值&＃xff0c;E表示预期值&＃xff0c;N表示新值。仅当 V值等于E值时&＃xff0c;才会将V的值设为N&＃xff0c;如果V值和E值不同&＃xff0c;则说明已经有其他线程做两个更新&＃xff0c;则当前线程则什么都不做。最后&＃xff0c;CAS 返回当前V的真实值。

我们举一个我之前举过的例子来说明这个过程&＃xff1a;

比如说给你儿子订婚。你儿子就是内存位置&＃xff0c;你原本以为你儿子是和杨贵妃在一起了&＃xff0c;结果在订婚的时候发现儿子身边是西施。这时候该怎么办呢&＃xff1f;你一气之下不做任何操作。如果儿子身边是你预想的杨贵妃&＃xff0c;你一看很开心就给他们订婚了&＃xff0c;也叫作执行操作。现在你应该明白了吧。

CAS 操作时抱着乐观的态度进行的&＃xff0c;它总是认为自己可以成功完成操作。所以CAS也叫作乐观锁&＃xff0c;那什么是悲观锁呢&＃xff1f;悲观锁就是我们之前赫赫有名的synchronized。悲观锁的思想你可以这样理解&＃xff0c;一个线程想要去获得这个锁但是却获取不到&＃xff0c;必须要别人释放了才可以。

2、底层原理

想要弄清楚其底层原理&＃xff0c;深入到源码是最好的方式&＃xff0c;在上面我们已经通过源码看到了其实就是Usafe的方法来完成的&＃xff0c;在这个方法中使用了compareAndSwapInt这个CAS机制。因此&＃xff0c;现在我们有必要进一步深入进去看看&＃xff1a;

public final class Unsafe {// compareAndSwapInt 是 native 类型的方法public final native boolean compareAndSwapInt(Object o, long offset,int expected,int x);//剩余还有很多方法
}

我们可以看到这里面主要有四个参数&＃xff0c;第一个参数就是我们操作的对象a&＃xff0c;第二个参数是对象a的地址偏移量&＃xff0c;第三个参数表示我们期待这个a是什么值&＃xff0c;第四个参数表示的是a的实际值。

不过这里我们会发现这个compareAndSwapInt是一个native方法&＃xff0c;也就是说再往下走就是C语言代码&＃xff0c;如果我们保持好奇心&＃xff0c;可以继续深入进去看看。

UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x))UnsafeWrapper("Unsafe_CompareAndSwapInt");oop p &＃61; JNIHandles::resolve(obj);// 根据偏移量valueOffset&＃xff0c;计算 value 的地址jint* addr &＃61; (jint *) index_oop_from_field_offset_long(p, offset);// 调用 Atomic 中的函数 cmpxchg来进行比较交换return (jint)(Atomic::cmpxchg(x, addr, e)) &＃61;&＃61; e;
UNSAFE_END

上面的代码我们解读一下&＃xff1a;首先使用jint计算了value的地址&＃xff0c;然后根据这个地址&＃xff0c;使用了Atomic的cmpxchg方法进行比较交换。现在问题又抛给了这个cmpxchg&＃xff0c;真实实现的是这个函数。我们再进一步深入看看&＃xff0c;真相已经离我们不远了。

unsigned Atomic::cmpxchg(unsigned int exchange_value,volatile unsigned int* dest, unsigned int compare_value) {assert(sizeof(unsigned int) &＃61;&＃61; sizeof(jint), "more work to do");/** 根据操作系统类型调用不同平台下的重载函数&＃xff0c;这个在预编译期间编译器会决定调用哪个平台下的重载函数*/return (unsigned int)Atomic::cmpxchg((jint)exchange_value, (volatile jint*)dest, (jint)compare_value);
}

皮球又一次被完美的踢走了&＃xff0c;现在在不同的操作系统下会调用不同的cmpxchg重载函数&＃xff0c;我现在用的是win10系统&＃xff0c;所以我们看看这个平台下的实现&＃xff0c;别着急再往下走走&＃xff1a;

inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {int mp &＃61; os::is_MP();__asm {mov edx, destmov ecx, exchange_valuemov eax, compare_valueLOCK_IF_MP(mp)cmpxchg dword ptr [edx], ecx}
}

这块的代码就有点涉及到汇编指令相关的代码了&＃xff0c;到这一步就彻底接近真相了&＃xff0c;首先三个move指令表示的是将后面的值移动到前面的寄存器上。然后调用了LOCK_IF_MP和下面cmpxchg汇编指令进行了比较交换。现在我们不知道这个LOCK_IF_MP和cmpxchg是如何交换的&＃xff0c;没关系我们最后再深入一下。

真相来了&＃xff0c;他来了&＃xff0c;他真的来了。

inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {//1、 判断是否是多核 CPUint mp &＃61; os::is_MP();__asm {//2、 将参数值放入寄存器中mov edx, dest mov ecx, exchange_valuemov eax, compare_value //3、LOCK_IF_MP指令cmp mp, 0//4、 如果 mp &＃61; 0&＃xff0c;表明线程运行在单核CPU环境下。此时 je 会跳转到 L0 标记处&＃xff0c;直接执行 cmpxchg 指令je L0_emit 0xF0
//5、这里真正实现了比较交换
L0:/** 比较并交换。简单解释一下下面这条指令&＃xff0c;熟悉汇编的朋友可以略过下面的解释:* cmpxchg: 即“比较并交换”指令* dword: 全称是 double word 表示两个字&＃xff0c;一共四个字节* ptr: 全称是 pointer&＃xff0c;与前面的 dword 连起来使用&＃xff0c;表明访问的内存单元是一个双字单元 * 这一条指令的意思就是&＃xff1a;将 eax 寄存器中的值&＃xff08;compare_value&＃xff09;与 [edx] 双字内存单元中的值进行对比&＃xff0c;如果相同&＃xff0c;则将 ecx 寄存器中的值&＃xff08;exchange_value&＃xff09;存入 [edx] 内存单元中。*/cmpxchg dword ptr [edx], ecx}
}

到这一步了&＃xff0c;相信你应该理解了这个CAS真正实现的机制了吧&＃xff0c;最终是由操作系统的汇编指令完成的。

3、CAS机制的优缺点

&＃xff08;1&＃xff09;优点

一开始在文中我们曾经提到过&＃xff0c;cas是一种乐观锁&＃xff0c;而且是一种非阻塞的轻量级的乐观锁&＃xff0c;什么是非阻塞式的呢&＃xff1f;其实就是一个线程想要获得锁&＃xff0c;对方会给一个回应表示这个锁能不能获得。在资源竞争不激烈的情况下性能高&＃xff0c;相比synchronized重量锁&＃xff0c;synchronized会进行比较复杂的加锁&＃xff0c;解锁和唤醒操作。

&＃xff08;2&＃xff09;缺点

缺点也是一个非常重要的知识点&＃xff0c;因为涉及到了一个非常著名的问题&＃xff0c;叫做ABA问题。假设一个变量 A &＃xff0c;修改为 B之后又修改为 A&＃xff0c;CAS 的机制是无法察觉的&＃xff0c;但实际上已经被修改过了。这就是ABA问题&＃xff0c;

ABA问题会带来大量的问题&＃xff0c;比如说数据不一致的问题等等。我们可以举一个例子来解释说明。

你有一瓶水放在桌子上&＃xff0c;别人把这瓶水喝完了&＃xff0c;然后重新倒上去。你再去喝的时候发现水还是跟之前一样&＃xff0c;就误以为是刚刚那杯水。如果你知道了真相&＃xff0c;那是别人用过了你还会再用嘛&＃xff1f;举一个比较黄一点的例子&＃xff0c;一个女孩变了心之后又回来&＃xff0c;还是之前的那个女娃嘛&＃xff1f;其实本质已经变了&＃xff0c;但是自己从外观来看好像并没有变化。

1、基本的ABA问题

在CAS算法中&＃xff0c;需要取出内存中某时刻的数据(由用户完成)&＃xff0c;在下一时刻比较并交换(CPU保证原子操作)&＃xff0c;这个时间差会导致数据的变化。
假设有以下顺序事件&＃xff1a;

1、线程1从内存位置V中取出A
2、线程2从内存位置V中取出A
3、线程2进行了写操作&＃xff0c;将B写入内存位置V
4、线程2将A再次写入内存位置V
5、线程1进行CAS操作&＃xff0c;发现V中仍然是A&＃xff0c;交换成功

尽管线程1的CAS操作成功&＃xff0c;但线程1并不知道内存位置V的数据发生过改变

2、ABA问题示例

public class ABADemo {private static AtomicReference atomicReference &＃61; new AtomicReference(100);public static void main(String[] args) {new Thread(() -> {atomicReference.compareAndSet(100, 101);atomicReference.compareAndSet(101, 100);},"t1").start();new Thread(() -> {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(atomicReference.compareAndSet(100, 2019) &＃43; "\t修改后的值:" &＃43; atomicReference.get());},"t2").start();}
}

• 初始值为100&＃xff0c;线程t1将100改成101&＃xff0c;然后又将101改回100
• 线程t2先睡眠1秒&＃xff0c;等待t1操作完成&＃xff0c;然后t2线程将值改成2019
• 可以看到&＃xff0c;线程2修改成功

输出结果&＃xff1a;

true 修改后的值:2019

3、ABA问题解决

要解决ABA问题&＃xff0c;可以增加一个版本号&＃xff0c;当内存位置V的值每次被修改后&＃xff0c;版本号都加1

AtomicStampedReference

AtomicStampedReference内部维护了对象值和版本号&＃xff0c;在创建AtomicStampedReference对象时&＃xff0c;需要传入初始值和初始版本号&＃xff0c;
当AtomicStampedReference设置对象值时&＃xff0c;对象值以及状态戳都必须满足期望值&＃xff0c;写入才会成功

示例

public class ABADemo {private static AtomicStampedReference atomicStampedReference &＃61; new AtomicStampedReference(100,1);public static void main(String[] args) {new Thread(() -> {System.out.println("t1拿到的初始版本号:" &＃43; atomicStampedReference.getStamp());//睡眠1秒&＃xff0c;是为了让t2线程也拿到同样的初始版本号try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}atomicStampedReference.compareAndSet(100, 101,atomicStampedReference.getStamp(),atomicStampedReference.getStamp()&＃43;1);atomicStampedReference.compareAndSet(101, 100,atomicStampedReference.getStamp(),atomicStampedReference.getStamp()&＃43;1);},"t1").start();new Thread(() -> {int stamp &＃61; atomicStampedReference.getStamp();System.out.println("t2拿到的初始版本号:" &＃43; stamp);//睡眠3秒&＃xff0c;是为了让t1线程完成ABA操作try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("最新版本号:" &＃43; atomicStampedReference.getStamp());System.out.println(atomicStampedReference.compareAndSet(100, 2019,stamp,atomicStampedReference.getStamp() &＃43; 1) &＃43; "\t当前 值:" &＃43; atomicStampedReference.getReference());},"t2").start();}
}

1、初始值100&＃xff0c;初始版本号1
2、线程t1和t2拿到一样的初始版本号
3、线程t1完成ABA操作&＃xff0c;版本号递增到3
4、线程t2完成CAS操作&＃xff0c;最新版本号已经变成3&＃xff0c;跟线程t2之前拿到的版本号1不相等&＃xff0c;操作失败

输出结果&＃xff1a;

t1拿到的初始版本号:1
t2拿到的初始版本号:1
最新版本号:3
false 当前 值:100

AtomicMarkableReference

AtomicStampedReference可以给引用加上版本号&＃xff0c;追踪引用的整个变化过程&＃xff0c;如&＃xff1a;A -> B -> C -> D - > A&＃xff0c;通过AtomicStampedReference&＃xff0c;我们可以知道&＃xff0c;引用变量中途被更改了3次
但是&＃xff0c;有时候&＃xff0c;我们并不关心引用变量更改了几次&＃xff0c;只是单纯的关心是否更改过&＃xff0c;所以就有了AtomicMarkableReference
AtomicMarkableReference的唯一区别就是不再用int标识引用&＃xff0c;而是使用boolean变量——表示引用变量是否被更改过

示例

public class ABADemo {private static AtomicMarkableReference atomicMarkableReference &＃61; new AtomicMarkableReference(100,false);public static void main(String[] args) {new Thread(() -> {System.out.println("t1版本号是否被更改:" &＃43; atomicMarkableReference.isMarked());//睡眠1秒&＃xff0c;是为了让t2线程也拿到同样的初始版本号try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}atomicMarkableReference.compareAndSet(100, 101,atomicMarkableReference.isMarked(),true);atomicMarkableReference.compareAndSet(101, 100,atomicMarkableReference.isMarked(),true);},"t1").start();new Thread(() -> {boolean isMarked &＃61; atomicMarkableReference.isMarked();System.out.println("t2版本号是否被更改:" &＃43; isMarked);//睡眠3秒&＃xff0c;是为了让t1线程完成ABA操作try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("是否更改过:" &＃43; atomicMarkableReference.isMarked());System.out.println(atomicMarkableReference.compareAndSet(100, 2019,isMarked,true) &＃43; "\t当前 值:" &＃43; atomicMarkableReference.getReference());},"t2").start();}
}

1、初始值100&＃xff0c;初始版本号未被修改 false
2、线程t1和t2拿到一样的初始版本号都未被修改 false
3、线程t1完成ABA操作&＃xff0c;版本号被修改 true
4、线程t2完成CAS操作&＃xff0c;版本号已经变成true&＃xff0c;跟线程t2之前拿到的版本号false不相等&＃xff0c;操作失败

输出结果&＃xff1a;

t1版本号是否被更改:false
t2版本号是否被更改:false
是否更改过:true
false 当前 值:100