CAS

CAS（Compare And Swap） 比较并交换 

 

 

前言

 

在JDK 5之前Java语言是靠synchronized关键字保证同步的，这会导致有锁。锁机制存在以下问题：

• 在多线程竞争下，加锁、释放锁会导致比较多的上下文切换和调度延时，引起性能问题。
  
• 一个线程持有锁会导致其它所有需要此锁的线程挂起。
  
• 如果一个优先级高的线程等待一个优先级低的线程释放锁会导致优先级倒置，引起性能风险。
  

volatile是不错的机制，但是volatile不能保证原子性。因此对于同步最终还是要回到锁机制上来。

独占锁是一种悲观锁，synchronized就是一种独占锁，会导致其它所有需要锁的线程挂起，等待持有锁的线程释放锁。而另一个更加有效的锁就是乐观锁。所谓乐观锁就是，每次不加锁而是假设没有冲突而去完成某项操作，如果因为冲突失败就重试，直到成功为止。乐观锁用到的机制就是CAS。

JDK 5增加了并发包java.util.concurrent.*,其下面的类使用CAS算法实现了区别于synchronized同步锁的一种乐观锁。JDK 5之前Java语言是靠synchronized关键字保证同步的，这是一种独占锁，也是是悲观锁。

在Java发展初期，java语言是不能够利用硬件提供的这些便利来提升系统的性能的。而随着java不断的发展,Java本地方法(JNI)的出现，使得java程序越过JVM直接调用本地方法提供了一种便捷的方式，因而java在并发的手段上也多了起来。而在Doug Lea提供的cucurenct包中，CAS理论是它实现整个java包的基石。

 

 

一、定义 

 

什么是CAS？

CAS从字面上理解就是比较并更新，简单来说：从某一内存上取值V，和预期值A进行比较，如果内存值V和预期值A的结果相等，那么我们就把新值B更新到内存，如果不相等，那么就重复上述操作直到成功为止。（值相同能改）

 

CAS它是一条CPU并发原语（原子语言，不可分割，不可中断）。它的功能是判断内存某个位置的值是否为预期值，如果是则修改为新的值，这个过程是原子的。

CAS并发原语体现在JAVA语言中就是sun.misc.Unsafe类中的各个方法。调用UnSafe类中的CAS方法，JVM会帮我们实现CAS汇编指令。这是一种完全依赖硬件的功能，通过它实现了原子操作。再次强调，由于CAS是一种系统原语，原语属于操作系统用语范畴，是由若干条指令组成的，用于完成某个功能的一个过程，并且原语执行必须是连续的，在执行过程中不允许被中断，也就是说CAS是一条CPU的原子指令，不会造成所谓的数据不一致问题。

 

示例

AtomicInteger ai = new AtomicInteger(5);
System.out.println(ai.compareAndSet(5, 2019) + " 当前值：" + ai.get());	// true 当前值：2019
System.out.println(ai.compareAndSet(5, 2020) + " 当前值：" + ai.get());	// false 当前值：2019
System.out.println(ai.compareAndSet(2019, 2020) + " 当前值：" + ai.get());	// true 当前值：2020
System.out.println(ai.getAndIncrement());		// 2020
System.out.println(ai.getAndIncrement());		// 2021

　　

 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;

    static {
        try {
            valueOffset = unsafe.objectFieldOffset
                (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }
　　 // 保证可见性
    private volatile int value;
    
	// ...省略很多代码
    
    public final boolean compareAndSet(int expect, int update) {
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    }
    // 每调用一次自增1
    public final int getAndIncrement() {
        // this为当前对象，valueOffset 为内存偏移量，即内存地址
        return unsafe.getAndAddInt(this, valueOffset, 1);
    }
    
}

　　

getAndIncrement 为什么不加synchronized也能保证原子性？

 （1）依靠底层的Unsafe类。Unsafe是CAS的核心类（Unsafe类在rt.jar包下，rt：runtime的缩写），由于Java方法无法直接访问底层操作系统，需要通过本地navtive方法来访问，Unsafe相当于一个后门，基于该类可以直接操作特定内存的数据。Unsafe类存在于sum.misc包中，其内部方法操作可以像C的指针一样直接操作内存，因为Java中的CAS操作的执行依赖于Unsafe类的方法。注意Unsafe类中的所有方法都是navite修饰的，也就是说Unsafe类中的方法都直接调用操作系统底层资源执行相应的任务。

（2）变量valueOffset，表示该变量值在内存中的偏移地址，因为Unsafe就是根据内存偏移地址获取数据的。

（3）变量value用volatile修饰，保证了多线程之间的内存可见性。value是一个volatile变量，即当进行+1操作时，其它线程均可

 

原理

（1）通过this.getIntVolatile 获取当前内存地址的值。即先从物理内存中猎取真实的值，拷贝到自已的工作内存中；

（2）通过this.compareAndSwapInt ：如果当前对象内存中的地址和var5一样，则进行var5+var4（var5+1）操作。否则一直自旋继续执行do内的代码，即如果有其它线程修改过此值，则继续猎取内存地址的值（循环通过的方式来等待其它的线程完成任务）。

可以理解为：JMM内存模型，拿工作内存中的预期值与主内存中的值去比较，如果一致，可以修改。如果不一致，则再来一遍。

 

执行过程：

1. 假设线程A和线程B两个线程同时执行getAndAddInt操作（分别跑在不同的CPU上（多核CPU））：
2. AtomicInteger里面的value原始值为3，即内存中的AtomicInteger的value为3，根据JMM模型，线程A和线程B各自持有一份值为3的value的副本分别在各自的工作内存；
3. 线程A通过getIntVolatile(var1, var2)拿到value值3，这时线程A被挂起；
4. 线程B也通过getIntVolatile(var1, var2) 方法获取到value值3，此时刚好线程B没有被挂起，并执行compareAndSwapInt方法，比较内存值也为3，成功修改内存值4，线程B打完收工，一切正常；
5. 这里线程A恢复，执行compareAndSwapInt方法比较，发现自已手里的值数字和主内存的值数字4不一致，说明该值已被其它线程抢先一步修改过了，那A线程本次修改失败，只能重新读取再 来一遍了；
6. 线程A重新获取value值，因为变量value被volatile修饰，所以其它线程对它的修改，线程A总是能够看到，线程A继续执行compareAndSwapInt进行比较替换，直到成功。 

个人理解：Unsafe通过getAndAddInt 将（当前对象 + 地址 + 值1）做出精确操作。

 

 

Unsafe类中的compareAndSwapInt，是一个本地方法，该方法的实现位于openjdkhotspotsrcsharevmprimsunsafe.cpp中

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);
  return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
UNSAFE_END

 

cas能做什么？

上面我们了解了cas是什么了，那么它能解决什么问题呢？它可以解决多线程并发安全的问题，以前我们对一些多线程操作的代码都是使用synchronize关键字，来保证线程安全的问题；现在我们将cas放入到多线程环境里我们看一下它是怎么解决的，我们假设有A、B两个线程同时执行一个int值value自增的代码，并且同时获取了当前的value，我们还要假设线程B比A快了那么0.00000001s，所以B先执行，线程B执行了cas操作之后，发现当前值和预期值相符，就执行了自增操作，此时这个value = value + 1;然后A开始执行，A也执行了cas操作，但是此时value的值和它当时取到的值已经不一样了，所以此次操作失败，重新取值然后比较成功，然后将value值更新，这样两个线程进入，value值自增了两次，符合我们的预期。

 

 

二、CAS存在的问题

 

1、经典的ABA问题

何为ABA呢？

我们还是以两个线程T1、T2进行自增操作为例，线程T1、T2同时获取当前的值A，只不过此时线程T2比较快，它在T1操作之前，进行了两次操作，第一次将值从A 改为了B，之后又将B改为了A，那么在线程T1操作的时候发现当前的值还是A，符合预期，那么它也会更新成功，从操作上看并没有什么不对，更新成功也是对的，但是这样是有隐患的；

这个网上有好多关于ABA问题隐患的解读，我觉得有一个老哥使用链表的表述最为贴切，这个是我很久之前看的，我现在也找不到这个老哥关于这个问题解读的帖子了，大家自行搜索一下吧，为了解决这个问题，java引入了版本的概念，相当于上述操作变为了A1----B2----A3,这样就非常明确了，这个版本相信大家也猜到那就是valueOffset,所以在AtomicInteger中进行cas操作时除了this、expect、update之外还有一个valueOffset的参数进行版本的区分，就是为了解决ABA问题的。

 

2、 循环时间长开销大

自旋CAS如果长时间不成功，会给CPU带来非常大的执行开销。如果JVM能支持处理器提供的pause指令那么效率会有一定的提升，pause指令有两个作用，第一它可以延迟流水线执行指令（de-pipeline）,使CPU不会消耗过多的执行资源，延迟的时间取决于具体实现的版本，在一些处理器上延迟时间是零。第二它可以避免在退出循环的时候因内存顺序冲突（memory order violation）而引起CPU流水线被清空（CPU pipeline flush），从而提高CPU的执行效率。

如果一个线程特别倒霉，每次获取的值都被其他线程的修改了，那么它就会一直进行自旋比较，直到成功为止，在这个过程中cpu的开销十分的大，所以要尽量避免。

 

3. 只能保证一个共享变量的原子操作。

当对一个共享变量执行操作时，我们可以使用循环CAS的方式来保证原子操作，但是对多个共享变量操作时，循环CAS就无法保证操作的原子性，这个时候就可以用锁，或者有一个取巧的办法，就是把多个共享变量合并成一个共享变量来操作。比如有两个共享变量i＝2,j=a，合并一下ij=2a，然后用CAS来操作ij。从Java1.5开始JDK提供了AtomicReference类来保证引用对象之间的原子性，你可以把多个变量放在一个对象里来进行CAS操作。

 

 

三、concurrent包的实现

由于java的CAS同时具有 volatile 读和volatile写的内存语义，因此Java线程之间的通信现在有了下面四种方式：

• A线程写volatile变量，随后B线程读这个volatile变量。
• A线程写volatile变量，随后B线程用CAS更新这个volatile变量。
• A线程用CAS更新一个volatile变量，随后B线程用CAS更新这个volatile变量。
• A线程用CAS更新一个volatile变量，随后B线程读这个volatile变量。

 

Java的CAS会使用现代处理器上提供的高效机器级别原子指令，这些原子指令以原子方式对内存执行读-改-写操作，这是在多处理器中实现同步的关键（从本质上来说，能够支持原子性读-改-写指令的计算机器，是顺序计算图灵机的异步等价机器，因此任何现代的多处理器都会去支持某种能对内存执行原子性读-改-写操作的原子指令）。同时，volatile变量的读/写和CAS可以实现线程之间的通信。把这些特性整合在一起，就形成了整个concurrent包得以实现的基石。如果我们仔细分析concurrent包的源代码实现，会发现一个通用化的实现模式：

1. 首先，声明共享变量为volatile；
2. 然后，使用CAS的原子条件更新来实现线程之间的同步；
3. 同时，配合以volatile的读/写和CAS所具有的volatile读和写的内存语义来实现线程之间的通信。

 

 

四、JDK1.8 中的CAS

 

Unsafe类存在于sun.misc包中，不属于Java标准。其内部方法操作可以像C的指针一样直接操作内存，单从名称看来就可以知道该类是非安全的，毕竟Unsafe拥有着类似于C的指针操作，因此总是不应该首先使用Unsafe类，Java官方也不建议直接使用的Unsafe类，但我们还是很有必要了解该类，因为Java中CAS操作的执行依赖于Unsafe类的方法，注意Unsafe类中的所有方法都是native修饰的，也就是说Unsafe类中的方法都直接调用操作系统底层资源执行相应任务。

其中与CAS相关的方法有以下几个：

/**
 * var1为CAS操作的对象
 * offset为var1某个属性的地址偏移值
 * expected为期望值
 * var2为要设置的值，利用JNI来完成CPU指令的操作
 */
public final native boolean compareAndSwapObject(Object var1, long offset, Object expected, Object var2);
public final native boolean compareAndSwapInt(Object var1, long offset, int expected, int var2);
public final native boolean compareAndSwapLong(Object var1, long offset, long expected, long var2);

　　

 

 

/** 如果CAS成功，return oldValue, oldValue =  oldValue + addValue
 *  如果CAS失败，自旋，一直运行，直到成功为止
 */
public final Xxx getAndAddXxx(Object var1, long offset, long addValue) {
    int oldValue;
    do {
        oldValue = this.getIntVolatile(var1, offset);
    } while(!this.compareAndSwapInt(var1, offset, oldValue, oldValue + addValue));
    return oldValue;
}

/** 如果CAS成功，return oldValue, oldValue =  newValue
 *  如果CAS失败，自旋，一直运行，直到成功为止
 */
public final Xxx getAndSetXxx(Object var1, long offset, Object newValue) {
    int oldValue;
    do {
        oldValue = this.getXxxVolatile(var1, offset);
    } while(!this.compareAndSwapXxx(var1, offset, oldValue, newValue));
    return oldValue;
}

　　

 

 

一般不建议使用Unsafe类，除非对它有很深入的了解。

java.util.concurrent包中大量使用了CAS原理，如AtomicInteger类，都是调用上面几个Unsafe方法保证多线程数据的正确性。以下是AtomicInteger的CAS操作相关源码

public class AtomicInteger extends Number implements java.io.Serializable {
    private static final long serialVersionUID = 6214790243416807050L;

    // setup to use Unsafe.compareAndSwapInt for updates
    // Unsafe类，提供一系列增强Java的功能，如内存管理、操作类/对象/变量、多线程同步等。不建议开发者调用
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    // 获取对象某个属性的地址偏移值
    private static final long valueOffset;

    static {
        try {
            // value相对“起始地址”的偏移量
            valueOffset = unsafe.objectFieldOffset
                    (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }

    // value值， volatile修饰，保证不同线程间的可见性
    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; }

    /**
     * Eventually sets to the given value.
     *
     * @param newValue the new value
     * @since 1.6
     */
    public final void lazySet(int newValue) {
        //有序或者有延迟的putIntVolatile方法
        unsafe.putOrderedInt(this, valueOffset, newValue);
    }

    /**
     * Atomically sets to the given value and returns the old value.
     * @param newValue the new value
     * @return the previous value
     */
    public final int getAndSet(int newValue) {
        return unsafe.getAndSetInt(this, valueOffset, newValue);
    }

    /**
     * Atomically sets the value to the given updated value
     * if the current value {@code ==} the expected value.
     * @param expect the expected value
     * @param update the new value
     * @return {@code true} if successful. False return indicates that
     * the actual value was not equal to the expected value.
     */
    public final boolean compareAndSet(int expect, int update) {
        // JNI调用，实现CAS
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    }

    /**
     * i++ 操作
     * Atomically increments by one the current value.
     * @return the previous value
     */
    public final int getAndIncrement() {
        return unsafe.getAndAddInt(this, valueOffset, 1);
    }

    /**
     * i-- 操作
     * Atomically decrements by one the current value.
     * @return the previous value
     */
    public final int getAndDecrement() {
        return unsafe.getAndAddInt(this, valueOffset, -1);
    }

    /**
     * return i, i = i + n 操作
     * Atomically adds the given value to the current value.
     * @param delta the value to add
     * @return the previous value
     */
    public final int getAndAdd(int delta) {
        return unsafe.getAndAddInt(this, valueOffset, delta);
    }

    /**
     * ++i 操作
     * Atomically increments by one the current value.
     * @return the updated value
     */
    public final int incrementAndGet() {
        return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
    }

    /**
     * --i 操作
     * Atomically decrements by one the current value.
     * @return the updated value
     */
    public final int decrementAndGet() {
        return unsafe.getAndAddInt(this, valueOffset, -1) - 1;
    }

    /**
     * i = i + n ,return i操作
     * Atomically adds the given value to the current value.
     * @param delta the value to add
     * @return the updated value
     */
    public final int addAndGet(int delta) {
        return unsafe.getAndAddInt(this, valueOffset, delta) + delta;
    }
    // 其余函数，略...