聊聊并发——CAS算法

Posted 2021-12-03 匠心

一、原子类

1、CAS算法

　　强烈建议读者看这篇之前，先看这篇 初识JUC 的前两节，对原子性，原子变量，内存可见性有一个初步认识。

　　CAS（Compare and Swap）是一种硬件对并发的支持，针对多处理器操作而设计的处理器中的一种特殊指令，用于管理对共享数据的并发访问，是硬件对于并发操作共享数据的支持。它是一个原子性的操作，对应到CPU指令为cmpxchg。它是一条CPU并发原语。
　　CAS包含了3个操作数：内存值V，比较值A，更新值B。当且仅当V == A时，V = B，否则不执行任何操作。
　　CAS算法：当多个线程并发的对主存中的数据进行修改的时候。有且只有一个线程会成功，其他的都会失败（同时操作，只是会失败而已，并不会被锁之类的）。
　　CAS是一种无锁的非阻塞算法，是乐观锁的一种实现。不存在上下文切换的问题。
　　CAS比普通同步锁效率高，原因：CAS算法当这一次不成功的时候，它下一次不会阻塞，也就是它不会放弃CPU的执行权，它可以立即再次尝试，再去更新。
　　通俗的说：我要将变量 i 由 2 修改为 3。当内存中 i == 2，且修改成功，才为成功。若内存中 i 由于其他线程的操作已经不是 2 了，那此次我的修改视为失败。

2、简单使用

　　JDK 1.5 以后java.util.concurrent.atomic包下提供了常用的原子变量。它支持单个变量上的无锁线程安全编程。这些原子变量具备以下特点：volatile的内存可见性；CAS算法保证数据的原子性。

　　atomic包描述：图片来源API文档

　　代码示例：原子变量使用

 public class Main {
     public static void main(String[] args) {
         AtomicInteger integer = new AtomicInteger(2);
 
         boolean b = integer.compareAndSet(3, 5);
         System.out.println(b);
         System.out.println(integer.get());
 
         b = integer.compareAndSet(2, 10);
         System.out.println(b);
         System.out.println(integer.get());
 
         // 等价于 i++
         integer.getAndIncrement();
 
         // 等价于 ++i
         integer.incrementAndGet();
     }
 }
 
 // 结果
 false
 2
 true
 10

　　分析：很简单，设置初始值为 2。
　　①由 3 修改成5，而设置初始值内存值为2，所以修改失败，返回false。
　　②由 2 修改成10，初始值内存值为2，所以修改成功，返回true。

3、源码分析

　　这些原子变量底层就是通过CAS算法来保证数据的原子性。
　　源码示例：AtomicInteger 类

 public class AtomicInteger extends Number implements java.io.Serializable {
     private static final long serialVersionUID = 6214790243416807050L;
 
     // setup to use Unsafe.compareAndSwapInt for updates
     private static final Unsafe unsafe = Unsafe.getUnsafe();
     private static final long valueOffset;
 
     // 获取value在内存的地址偏移量
     static {
         try {
             valueOffset = unsafe.objectFieldOffset
                 (AtomicInteger.class.getDeclaredField("value"));
         } catch (Exception ex) { throw new Error(ex); }
     }
 
     private volatile int value;
 
     public AtomicInteger(int initialValue) {
         value = initialValue;
     }
 
     public AtomicInteger() {
     }
 
     public final int get() {
         return value;
     }
 
     public final void set(int newValue) {
         value = newValue;
     }
 
     public final boolean compareAndSet(int expect, int update) {
         return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
     }
 
     public final int getAndIncrement() {
         return unsafe.getAndAddInt(this, valueOffset, 1);
     }
 
     public final int incrementAndGet() {
         return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
     }
 
 }

　　说明：public final boolean compareAndSet(int expect, int update)
　　变量valueOffset：通过静态代码块获取变量value在内存中的偏移地址。
　　变量value：用volatile修饰，这里体现了"多线程之间的内存可见性"。
　　this：即 AtomicInteger 对象本身。
　　很容易理解：就是将当前对象 this 的变量value，由期望值 expect 修改为 update。

　　源码示例：Unsafe 类

 public final class Unsafe {
 
     public native void throwException(Throwable var1);
 
     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);
 
     public native int getIntVolatile(Object var1, long var2);
 
 
     public final int getAndAddInt(Object var1, long var2, int var4) {
         int var5;
         do {
             // 获取对象var1的变量var2的内存值
             var5 = this.getIntVolatile(var1, var2);
         } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
 
         return var5;
     }
 
 }

　　Unsafe是CAS的核心类，其所有方法都是native修饰的。也就是说Unsafe类中的方法都直接调用操作系统底层资源执行相应任务，是由C/C++编写的本地方法。CAS算法的实现，也是由Unsafe类通过调用本地方法直接操作特定内存数据来实现的。
　　getAndIncrement()方法能够在多线程环境保证变量的原子性自增。但源码中，并没有加synchronized或者lock锁，那么，它是如何保证的呢？其实很简单：

　　先获取一次变量的内存值，然后通过CAS算法进行比较更新。失败了就一直不停的重试，是一个循环的过程，这个过程也称作自旋。
　　这就是为什么 AtomicInteger 的自增操作具备原子性。

 private AtomicInteger i = new AtomicInteger();
 public int getI() {
     return i.getAndIncrement();
 }

4、CAS的缺点

　　（1）ABA问题。
　　（2）循环时间变长：高并发情况下，使用CAS可能会存在一些线程一直循环修改不成功，导致循环时间变长，这会给CPU带来很大的执行开销。由于AtomicInteger中的变量是volatile的，为了保证内存可见性，需要保证缓存一致性，通过总线传输数据，当有大量的CAS循环时，会产生总线风暴。
　　（3）只能保证一个变量的原子操作：如果需要保证多个变量操作的原子性，是做不到的。对于这种情况只能使用synchronized或者juc包中的Lock工具。

二、ABA问题

1、介绍

　　代码示例：演示ABA问题

 // 原子引用类演示ABA问题
 public class ABATest {
     public static void main(String[] args) throws InterruptedException {
         AtomicReference<String> reference = new AtomicReference<>("A");
 
         // 线程 t1 由A修改B,又由B修改A
         new Thread(() -> {
             System.out.println(reference.compareAndSet("A", "B") + ". " + Thread.currentThread().getName() + " value is:" + reference.get());
             System.out.println(reference.compareAndSet("B", "A") + ". " + Thread.currentThread().getName() + " value is:" + reference.get());
         }, "t1").start();
 
 
         new Thread(() -> {
             // 让t1线程完成ABA操作
             try {
                 Thread.sleep(500);
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
             System.out.println(reference.compareAndSet("A", "C") + ". " + Thread.currentThread().getName() + " value is:" + reference.get());
 
         }, "t2").start();
 
         Thread.sleep(1000);
 
         System.out.println(reference.get());
     }
 }
 
 // 结果
 true. t1 value is:B
 true. t1 value is:A
 true. t2 value is:C
 C

　　如何理解ABA问题？
　　可能你会觉得，线程 t2 不就是要将"A"改为"C"嘛，虽然中间变化了，但对 t2 也没影响呀！
　　比如：你的银行卡里有10w，中间你领了工资1w，然后，又被扣除还了房贷1w，此时，你的银行卡里还是10w。虽然结果没变，但余额已经不是原来的余额了。而且，你一定在意中间你的钱去哪里了，所以是不一样的。
　　再比如：对于公司财务来说，可能某一时刻，账户是100w，你偷偷挪用了公款20w，后来又悄悄补上了。虽然结果没变，但中间的记账明细，其实我们是关心的，因为这个时候你已经犯法了。

2、解决

　　带时间戳的原子引用：Java提供了AtomicStampedReference来解决ABA问题。其实其实就是加了版本号，每一次的修改，版本号都 +1。比对的是 内存值 + 版本号 是否一致。
　　代码示例：解决ABA问题

 public class ABATest {
     public static void main(String[] args) throws InterruptedException {
 
         AtomicStampedReference<String> reference = new AtomicStampedReference<>("A", 1);
         final int stamp = reference.getStamp();
 
         // 线程 t1 由A修改B,又由B修改A
         new Thread(() -> {
             System.out.println(reference.compareAndSet("A", "B", stamp, stamp + 1) + ". " + Thread.currentThread().getName() + " value is:" + reference.getReference());
             System.out.println(reference.compareAndSet("B", "A", reference.getStamp(), reference.getStamp() + 1) + ". " + Thread.currentThread().getName() + " value is:" + reference.getReference());
         }, "t1").start();
         
 
         new Thread(() -> {
             // 让t1线程完成ABA操作
             try {
                 Thread.sleep(500);
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
             System.out.println(reference.compareAndSet("A", "C", stamp, stamp + 1) + ". " + Thread.currentThread().getName() + " value is:" + reference.getReference());
 
         }, "t2").start();
 
         Thread.sleep(1000);
 
         System.out.println(reference.getReference());
     }
 }
 
 // 结果
 true. t1 value is:B
 true. t1 value is:A
 false. t2 value is:A    // t2并没有修改成功
 A

　　compareAndSet()方法的 4 个参数：

　　expectedReference：表示期望的引用值
　　newReference：表示要修改后的新引用值
　　expectedStamp：表示期望的戳（版本号）
　　newStamp：表示修改后新的戳（版本号）

3、源码分析

 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);
         }
     }
     
     public boolean compareAndSet(V   expectedReference,
                                  V   newReference,
                                  int expectedStamp,
                                  int newStamp) {
         Pair<V> current = pair;
         return
             expectedReference == current.reference &&
             expectedStamp == current.stamp &&
             ((newReference == current.reference &&
               newStamp == current.stamp) ||
              casPair(current, Pair.of(newReference, newStamp)));
     }
 
     private boolean casPair(Pair<V> cmp, Pair<V> val) {
         return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
     }
 }

　　很简单，维护了一对Pair，里面除了引用reference，还有一个int类型的戳（版本号）。比较更新的时候，两个变量都要比较。

三、LongAdder

1、介绍

　　《阿里巴巴Java开发手册》推荐使用LongAdder。

　　AtomicLong，本质上是多个线程同时操作同一个目标资源，有且只有一个线程执行成功，其他线程都会失败，不断重试（自旋），自旋会成为瓶颈。
　　而LongAdder的思想就是把要操作的目标资源[分散]到数组Cell中，每个线程对自己的Cell变量的value进行原子操作，大大降低了失败的次数。
　　这就是为什么在高并发场景下，推荐使用LongAdder的原因。

　　参考文档：https://www.matools.com/api/java8
　　《阿里巴巴Java开发手册》百度网盘：https://pan.baidu.com/s/1aWT3v7Efq6wU3GgHOqm-CA 密码: uxm8

作者：Craftsman-L

出处：https://www.cnblogs.com/originator

本博客所有文章仅用于学习、研究和交流目的，版权归作者所有，欢迎非商业性质转载。

如果本篇博客给您带来帮助，请作者喝杯咖啡吧！点击下面打赏，您的支持是我最大的动力！