高并发之深度解析CAS [理论+案例+源码]

1、理论

2、问题

3、实操——Atomic打头的类

4、原理：基于类Striped64分散热点[空间换时间]

1、理论

CAS，即compare and swap，比较并交换，包含三个操作数，内存位置、预期原值及更新值。

3个操作数，位置内存值V，旧的预期值A，要修改的更新值B。

当且仅当旧的预期值A和内存值V相同时，将内存值V修改为B，否则什么都不做，或重新来过，当它重来重试的这种行为称为——自旋！

CAS是JDK提供的非阻塞原子性操作，它通过硬件保证了比较-更新的原子性。它是非阻塞的且自身具有原子性，也就是说这玩意儿效率更高且通过硬件保证，说明这玩意更可靠[不是synchronized，不牵扯用户态和内核态的切换，底层是CPU原语]。

CAS是一条CPU的原子指令（cmpxchg指令），不会造成所谓的数据不一致问题，Java的Unsafe类提供的CAS方法（如compareAndSwapXXX）底层实现即为CPU指令cmpxchg。

执行cmpxchg指令的时候，会判断当前系统是否为多核系统，如果是就给总线加锁，只有一个线程会对总线加锁成功，加锁成功之后会执行CAS操作，也就是说CAS的原子性实际上是CPU实现独占的，比起用synchronized重量级锁，这里的排他时间要短很多，所以在多线程情况下性能会比较好。

    public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5);public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);
/**
* var1：表示要操作的对象
* var2：表示要操作对象中属性地址的偏移量
* var4：表示需要修改数据的期望的值
* var5/var6：表示需要修改为的新值
*//*** 用一个死循环不断尝试给变量赋新值* @param var1 内存地址* @param var2 地址偏移量，var1和var2一起确定当前对象* @param var4 需更新的值* @return 更新后的值*/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;}

注：千万不要自己使用Unsafe类，可能造成内存混乱。

2、问题

1.如果CAS长时间一直不成功，可能会给CPU带来很大消耗【空转、干耗】；

2.ABA问题

说明：比如说一个线程1从内存位置V中取出A，这个时候另一个线程2也从内存中取出A，并且线程2进行了一些操作将值变成了B，然后线程3又将V位置的数据变成A，这个时候线程1进行CAS操作发现内存中任然是A，预期OK，然后线程1操作成功。

尽管线程1的CAS操作成功，但是不代表这个过程就是没有问题的；其实看你的业务走向，如果觉得”ABA“是有问题的，可以加一个版本号解决，其实我们以Atomic打头的这些类有封装了版本号的一个AtomicStampedReference类。

3、实操——Atomic打头的类

基本类型原子类：

AtomicInteger

AtomicBoolean

AtomicLong

数组类型原子类：

AtomicIntegerArray

AtomicLongArray

AtomicRefrenceArray

引用类型原子类：

AtomicRefrence<V>

AtomicStampedReference<V>：流水标记，版本号，解决多次的问题

AtomicMarkableReference<V>：标记为，true，false，解决一次性问题

对象的属性修改原子类：(粒度更细)

AtomicIntegerFieldUpdater       原子更新对象中int类型字段的值

AtomicLongFieldUpdater         原子更新对象中Long类型字段的值

AtomicReferenceFieldUpdater        原子更新引用类型字段的值

使用要求：

1、更新的对象属性必须使用public volatile 修饰符；

2、因为对象的属性修改类型原子类都是抽象类，所以每次使用都必须使用静态方法newUpdater()创建一个更新器，并且需要设置想要更新的类和属性。

案例如下：

package com.example.sycn;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;/*** @Author George* @Date 2022/11/24 19:30* @VDesc* 以一种线程安全的方式操作非线程安全对象的某些字段。* 需求：* 10个线程，每个线程转账1000，不使用synchronized，尝试使用AtomicIntegerFieldUpdater来实现*/
@Slf4j
public class AtomicIntegerFieldUpdaterDemo {public static void main(String[] args) throws InterruptedException {BankAccount bankAccount = new BankAccount();CountDownLatch countDownLatch = new CountDownLatch(10);for (int i = 1; i <= 10; i++) {new Thread(() -> {try {for (int j = 1; j <= 1000; j++) {//bankAccount.add();bankAccount.transMoney(bankAccount);}} finally {countDownLatch.countDown();}},String.valueOf(i)).start();}countDownLatch.await();log.debug("结果 {}",bankAccount.money);}
}class BankAccount{String bankName = "CCB";/*** 1、更新的对象属性必须使用public volatile 修饰符；*/public volatile int money = 0;public synchronized void add(){money++;}/*** 2、使用静态方法newUpdater()创建一个更新器，并且需要设置想要更新的类和属性。*/AtomicIntegerFieldUpdater<BankAccount> fieldUpdater = AtomicIntegerFieldUpdater.newUpdater(BankAccount.class,"money");public void transMoney(BankAccount bankAccount){fieldUpdater.getAndIncrement(bankAccount);}
}

原子操作增强类原理深度理解：下面的4个类java8才有

DoubleAccumulator

DoubleAdder

LongAccumulator 计算更复杂有加减乘除等等

LongAdder 计算简单方法

案例如下：

package com.example.sycn;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAccumulator;
import java.util.concurrent.atomic.LongAdder;/*** @Author George* @Date 2022/11/24 20:07* @VDesc* 设计一个累加器，每个累加器点赞5000万次，统计性能*/
@Slf4j
public class AccumulatorCompareDemo {public static final int _1W = 10000;public static final int threadNumber = 50;public static void main(String[] args) throws InterruptedException {ClickNumber clickNumber = new ClickNumber();long startTime;long endTime;CountDownLatch countDownLatch1 = new CountDownLatch(threadNumber);CountDownLatch countDownLatch2 = new CountDownLatch(threadNumber);CountDownLatch countDownLatch3 = new CountDownLatch(threadNumber);CountDownLatch countDownLatch4 = new CountDownLatch(threadNumber);startTime = System.currentTimeMillis();for (int i = 1; i <= threadNumber; i++) {new Thread(() -> {try {for (int j = 0; j < 100 * _1W; j++) {clickNumber.clickBySync();}} finally {countDownLatch1.countDown();}},String.valueOf(i)).start();}countDownLatch1.await();endTime = System.currentTimeMillis();log.debug("时间 {}毫秒，结果clickBySync：{}",(endTime-startTime),clickNumber.number);startTime = System.currentTimeMillis();for (int i = 1; i <= threadNumber; i++) {new Thread(() -> {try {for (int j = 0; j < 100 * _1W; j++) {clickNumber.clickByAtomicLong();}} finally {countDownLatch2.countDown();}},String.valueOf(i)).start();}countDownLatch2.await();endTime = System.currentTimeMillis();log.debug("时间 {}毫秒，clickByAtomicLong：{}",(endTime-startTime),clickNumber.atomicLong.get());startTime = System.currentTimeMillis();for (int i = 1; i <= threadNumber; i++) {new Thread(() -> {try {for (int j = 0; j < 100 * _1W; j++) {clickNumber.clickByLongAdder();}} finally {countDownLatch3.countDown();}},String.valueOf(i)).start();}countDownLatch3.await();endTime = System.currentTimeMillis();log.debug("时间 {}毫秒，clickByLongAdder：{}",(endTime-startTime),clickNumber.longAdder.sum());startTime = System.currentTimeMillis();for (int i = 1; i <= threadNumber; i++) {new Thread(() -> {try {for (int j = 0; j < 100 * _1W; j++) {clickNumber.clickLongAccumulator();}} finally {countDownLatch4.countDown();}},String.valueOf(i)).start();}countDownLatch4.await();endTime = System.currentTimeMillis();log.debug("时间 {}毫秒，clickLongAccumulator：{}",(endTime-startTime),clickNumber.longAccumulator.get());}}class ClickNumber{int number = 0;public synchronized void clickBySync(){number++;}AtomicLong atomicLong = new AtomicLong();public void clickByAtomicLong(){atomicLong.getAndIncrement();}LongAdder longAdder = new LongAdder();public void clickByLongAdder(){longAdder.increment();}LongAccumulator longAccumulator = new LongAccumulator((x,y) -> x + y,0);public void clickLongAccumulator(){longAccumulator.accumulate(1);}
}

4、原理：基于类Striped64分散热点[空间换时间]

下面是 Striped64 几个重要的成员变量和方法说明

base：类似于AtomicLong中全局的value值。在没有竞争情况下数据直接累加到base上，或者cells扩容时，也需要将数据写入到base上

collide：表示扩容意向，false一定不会扩容，true可能会扩容。

cellsBusy：初始化cells或者扩容cells需要获取锁，0：表示无锁状态 1：表示其它线程已经持有了锁

casCellsBusy()：通过CAS操作修改cellsBusy的值，CAS成功代表获取锁，返回true

NCPU：当前计算机CPU数量，Cell数组扩容时会使用到

getProbe()：获取当前线程的hash值

advanceProbe()：重置当前线程的hash值

Striped64 重要的四个成员变量

    /** Number of CPUS, to place bound on table size CPU数量，即cells数组的最大长度*/static final int NCPU = Runtime.getRuntime().availableProcessors();/*** Table of cells. When non-null, size is a power of 2.* cells数组，为2的幂，2，4，8，16.....，方便以后位运算*/transient volatile Cell[] cells;/*** Base value, used mainly when there is no contention, but also as* a fallback during table initialization races. Updated via CAS.* 基础value值，当并发较低时，只累加该值主要用于没有竞争的情况，通过CAS更新*/transient volatile long base;/*** Spinlock (locked via CAS) used when resizing and/or creating Cells.* 创建或者扩容Cells数组时使用的自旋锁变量调整单元格大小（扩容），创建单元格时使用的锁*/transient volatile int cellsBusy;

详解LongAdder为什么会比AtomicInteger快，核心就是LongAdder进行了热点分散[创建最大值为CPU核数的数组处理是所有的请求]，以下是LongAdder类add()方法详解

    public void add(long x) {/*** as是Striped64中的cells数组属性* b是Striped64中的base属性* v是当前线程hash到的Cell[槽位]中存储的值* m是cells的长度减1，hash是作为掩码使用* a是当前线程hash到的Cell*/Cell[] as; long b, v; int m; Cell a;/*** 首次首线程((as = cells) != null)一定是false，此时走casBase方法，以CAS的方式更新base值，且只有当cas失败时，才会走到if中* 条件1：cells不为空* 条件2：cas操作base失败，说明其它线程先一步修改了base正在出现竞争*/if ((as = cells) != null || !casBase(b = base, b + x)) {// true无竞争  false表示竞争激烈，多个线程hash到同一个Cell，可能要扩容boolean uncontended = true;/*** 条件1：cells为空，第一次来as也即是空，所以新建Cell[2]的数组[Striped64]* 条件2：应该不会出现* 条件3：当前线程所在的Cell为空，说明当前线程还没有更新过Cell，应该初始化一个Cell* 条件4：更新当前线程所在的Cell失败，说明现在竞争很激烈，多个线程hash到了同一个Cell，应扩容*/if (as == null || (m = as.length - 1) < 0 ||/*** getProbe()方法返回的是线程中的threadLocalRandomProbe字段* 它是通过随机数生成的一个值，对于一个确定的线程这个值是固定的(除非刻意修改它)*/(a = as[getProbe() & m]) == null || !(uncontended = a.cas(v = a.value, v + x))) {longAccumulate(x, null, uncontended);// 调用Striped64中的方法处理}}}

从上面的LongAdder类add()方法分析可以看出，最后又将重点落到了我们的Striped64类的longAccumulate()方法

图解说明 LongAdder类add()方法：

由上图可知，LongAdder在并发量大的时候除base值接收请求，处理请求外，在增加了Cell[]数组来接收处理请求，一下子吞吐量就上去了，下面是Striped64的longAccumulate源码解析，建议和上面的add()方法一起看，建议按CASE2、CASE3、CASE1步骤来看

/*** @param x                 需要增加的值* @param fn                默认传递的时null* @param wasUncontended    竞争标识，如果是false则代表有竞争。只有cells初始化之后，并且当前线程CAS竞争修改失败，才会是false*/final void longAccumulate(long x, LongBinaryOperator fn,boolean wasUncontended) {// 存储线程的probe值int h;// 如果getProbe()方法返回0，说明随机数未初始化if ((h = getProbe()) == 0) {// 使用ThreadLocalRandom为当前线程重新计算一个hash值，强制初始化ThreadLocalRandom.current(); // force initialization// 重新获取probe值，hash值被重置就好比一个全新的线程一样，所以设置了wasUncontended竞争状态为tureh = getProbe();// 重新计算了当前线程的hash后认为此次不算是一次竞争，都未初始化，肯定还不存在竞争激烈wasUncontended竞争状态为truewasUncontended = true;}boolean collide = false; // True if last slot nonempty[如果hash取模映射得到的Cell但愿不是null，则为true，此值也可以看作是扩容意向]for (;;) {Cell[] as; Cell a; int n; long v;// CASE1:cells已经被初始化了if ((as = cells) != null && (n = as.length) > 0) {if ((a = as[(n - 1) & h]) == null) { // 当前线程的hash值运算后映射到的Cell单元为null，说明该Cell没有被使用if (cellsBusy == 0) {       // Try to attach new Cell【尝试附加新单元格】；Cell[]数组没有正在扩容Cell r = new Cell(x);   // Optimistically create【乐观地创建】；创建一个Cell单元if (cellsBusy == 0 && casCellsBusy()) {boolean created = false;try {               // Recheck under lock【在锁定状态下重新检查】Cell[] rs; int m, j;// 在Cell单元附加到Cell[]数组上if ((rs = cells) != null &&(m = rs.length) > 0 &&rs[j = (m - 1) & h] == null) {rs[j] = r;created = true;}} finally {cellsBusy = 0;}if (created) {break;}continue;           // Slot is now non-empty}}collide = false;}else if (!wasUncontended) // CAS already known to fail；wasUnconteded表示前一次CAS更新Cell单元是否成功了{wasUncontended = true; // Continue after rehash；重新置为true，后面会重新计算线程的hash值} else if (a.cas(v = a.value, ((fn == null) ? v + x : fn.applyAsLong(v, x)))) { // 尝试CAS更新Cell单元值break;} else if (n >= NCPU || cells != as) {  // 当Cell数组的大小超过CPU核数后，永远不会再进行扩容collide = false;            // At max size or stale；扩容标识，置为false，表示不会再进行扩容} else if (!collide) {collide = true;} else if (cellsBusy == 0 && casCellsBusy()) {      // 尝试加锁进行扩容try {if (cells == as) {      // Expand table unless staleCell[] rs = new Cell[n << 1];// 按位左移1位来操作，扩容大小为之前容量的2倍for (int i = 0; i < n; ++i)  // 扩容后再将之前数组的元素拷贝到新数组中rs[i] = as[i];cells = rs;}} finally {cellsBusy = 0;      // 释放锁设置cellsBusy = 0，设置扩容状态，然后继续循环执行}collide = false;continue;               // Retry with expanded table}h = advanceProbe(h);        // 计算当前线程新的hash值}// CASE2:cells没有加锁且没有初始化，则尝试对它进行加锁，并初始化cells数组else if (cellsBusy == 0 && cells == as && casCellsBusy()) {boolean init = false;try {                           // Initialize table// 不double check，就会再次new一个cell数组，上一个线程对应数组中的值将会被篡改if (cells == as) {Cell[] rs = new Cell[2];// 新建一个大小为2的Cell数组rs[h & 1] = new Cell(x);// 找到当前线程hash到数组中的位置并创建其对应的Cellcells = rs;init = true;}} finally {cellsBusy = 0;}if (init) {break;}}// CASE3:cells正在进行初始化，则尝试直接在基数base上进行累加操作【兜底的】else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x)))) {break;                          // Fall back on using base}}}

图解 LongAdder的add()方法 + Striped64类的longAccumulate()方法

其实学到这里，有点感想，synchronized和CAS，其实sync的”锁“粒度要比CAS大，而且关系到用户态和内核态的切换；而CAS的底层是CPU原语，更深一点，为保证高并发下每个线程执行成功，C/C++底层会锁总线[这里的总线应该就是通信总线，也就是一个通信协议，属于硬件与硬件之间交互、交流的协议[或叫方式]]{这段中括号的话我只能跟着我学单片机时的感想而来，不能保证完全正确，但是个人会一直学习，假以时日定能完全吃透，功法随心}，而AtomicLong与LongAdder，LongAdder抗并发要高也是基于计算机的常识设计，一个扛不住，我就分散，多搞几个，寻求最优解。

帘外雨潺潺，春意阑珊。罗衾不耐五更寒。梦里不知身是客，一晌贪欢。独自莫凭栏，无限江山，别时容易见时难，落花流水春去也，天上人间。——李煜

日后继续更新完善此文~~