Java - 原子变量和CAS

---

首先引入一个Counter类

/**
 1. Created by Joe on 2018/4/10.
 */
public class Counter {
    private int count;

    public synchronized void incr() {
        count++;
    }

    public synchronized int getCount() {
        return count;
    }
}

在这个Counter类中，我们使用了synchronized关键字来保证了自增操作的原子性。但是对于counter++来说，使用synchronized的成本比较高昂，比如： 锁的获取和释放，获取不到锁时需要等待，线程上下文的切换。

在这种情况下就可以考虑使用原子变量，在Java并发包中提供了以下几种基本原子变量。

1. AtomicBoolean： 常用于在程序中表示一个标志位。
2. AtomicInteger
3. AtomicLong：常用于在程序中生成唯一序列号。
4. AtomicReference：用来以原子方式更新复杂类型。

除了以上这四个基本类型以外，还有针对数组的AtomicLongArray，AtomicReferenceArray；以及用于原子方式更新对象字段的类，如AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater等。

同时Java 8也新增了几个类，用于在高并发统计汇总的场景之中，比如：LongAdder，LongAccumulator，DoubleAdder和DoubleAccumulator。

一、AtomicInteger

1.1 基本用法

AtomicInteger 有两个构造方法：

/**
 * Creates a new AtomicInteger with the given initial value.
 *
 * @param initialValue the initial value
 */
public AtomicInteger(int initialValue) {
    value = initialValue;
}

/**
 * Creates a new AtomicInteger with initial value {@code 0}.
 */
public AtomicInteger() {
}

第一个方法给定了一个初始值，第二个构造方法的默认值为0。

我们也可以直接获取或者设置AtomicInteger的值，方法如下：

/**
 * Gets the current value.
 *
 * @return the current value
 */
public final int get() {
    return value;
}

/**
 * Sets to the given value.
 *
 * @param newValue the new value
 */
public final void set(int newValue) {
    value = newValue;
}

之所以称之为原子变量，是因为它包含了一些以原子方式实现组合操作的方法，如下：

//以原子方式获取旧值并且设置新值
public final int getAndSet(int newValue)
//以原子方式获取旧值并且给当前值+1
public final int getAndIncrement()
//以原子方式获取旧值并且给当前值-1
public final int getAndDecrement()
//以原子方式获取旧值并且给当前值+delta
public final int getAndAdd(int delta)
//以原子方式给当前值+1并获取新值
public final int incrementAndGet()
//以原子方式给当前值-1并获取新值
public final int decrementAndGet()
//以原子方式当前值-delta并获取新值
public final int addAndGet(int delta)

以上方法都依赖于另外一个 public 方法：

public final boolean compareAndSet(int expect, int update)

compareAndSet是一个非常重要的方法，名为比较并设置，其简称为CAS。该方法有两个参数expect和update。 该方法以原子的方式实现了以下的功能： 如果当前值等于 expect, 则更新为 update，否则不更新。如果更新成功返回true，否则返回false。

AtomicInteger 可以在程序中用作一个计数器，用于多个线程并发更新，保证代码的正确性。例如以下代码所示：

public class AtomicIntegerDemo {
    private static AtomicInteger counter = new AtomicInteger(0);
    
    static class Visitor extends Thread {
        @Override
        public void run() {
            for(int i = 0; i < 1000; i++) {
                counter.incrementAndGet();
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        int num = 1000;
        Thread[] threads = new Thread[num];
        for(int i = 0; i < num; i++) {
            threads[i] = new Visitor();
            threads[i].start();
        }
        for(int i = 0; i < num; i++) {
            threads[i].join();
        }
        System.out.println(counter.get());
    }
}

该代码的程序输出只有一个值:1000000

1.2 基本原理

AtomicInteger类主要内部成员为：

private volatile int value;

该成员被声明为volatile，这为了保证内存可见性。

其大部分方法的更新方式实现都类似，接下来看incrementAndGet方法的实现：

public final int incrementAndGet() {
    for(;;) {
        int current = get();
        int next = current + 1;
        if(compareAndSet(current, next))
            return next;
    }
}

如果观看的是jdk 8的源码的话，你会发现源码已经有所不同，是如下的方式：

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

现在依旧看上面的那个代码。代码主体是一个死循环，先获取当前值current，并计算期望值next，然后调用CAS方法进行更新，如果更新没有成功，说明value被别的线程已经更改了。然后再去获取最新值并尝试更新直到成功为止。

与synchronized锁相比，这种原子更新方式代表一种不同的思维方式。

1. 其中synchronized是悲观的，它假定更新可能冲突，所以先获取锁，得到锁后才更新
2. 原子变量的更新逻辑是乐观，它假定冲突较少，并使用CAS更新，也就是进行冲突检测。如果确实冲突了，则继续尝试即可。

synchronized代表一种阻塞算法，当得不到锁的时候，进入锁等待队列，等待其他线程将其唤醒，因此产生了上下文切换的开销；原子变量的更新是非阻塞的，当更新冲突时则会进行重试，不会阻塞，也不会因此产生上下文切换的开销。

对于大部分比较简单的操作，无论是在低并发还是高并发的情况下，使用乐观非阻塞的方式都远高于悲观阻塞的方式。

原子变量相对比较简单，但对于较为复杂的数据结构和算法，非阻塞的方式往往难以实现和理解，幸运的是，Java并发包中已经提供了一些非阻塞容器，比如：

1. ConcurrentLinkedQueue和ConcurrentLinkedDeque：非阻塞并发队列
2. ConcurrentSkipListMap和ConcurrentSkipListSet：非阻塞并发Map和Set

记下来继续看一下compareAndSet方法的实现方式，代码如下：

public final boolean compareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

它调用了unsafe的compareAndSwapInt方法。 unsafe的类型为 sun.misc.Unsafe ，其定义为：

private static final Unsafe unsafe = Unsafe.getUnsafe();

它是Sun的私有实现，从名字上面来看其表示的意思是“不安全”。而在一般的应用程序不会直接使用。原理上，一般的计算机系统都会在硬件层面上直接支持CAS操作，而Java的实现都会利用这些特殊指令。从程序的角度看，可以将compareAndSet视为计算机的基本操作。

1.3 实现锁

基于CAS，除了可以实现乐观非阻塞之外，还可以实现悲观阻塞式算法，比如锁。接下来使用AtomicInteger实现一个简单的锁MyLock。

public class MyLock {
    private AtomicInteger status = new AtomicInteger(0);
        public void lock() {
            while(!status.compareAndSet(0, 1)) {
                Thread.yield();
        }
    }
    public void unlock() {
        status.compareAndSet(1, 0);
    }
}

在MyLock中，使用status表示锁的状态，0表示未锁定，1表示锁定。lock()、unlock()使用CAS方法更新变量。lock()方法只有在更新成功之后才会退出，从而实现阻塞。不过这种方法过于消耗CPU，实际开发中应该使用Java并发包中的类，例如ReentrantLock。

二、ABA问题

在使用CAS方式更新会有一个ABA问题。该问题是指，假设当前值为A，如果另一个线程先将A修改成B，再修改成A，而当前线程的CAS操作无法分辨当前值发生过变化。

而ABA会不会造成一个问题，与程序的逻辑有关。在大部分情况下并不会造成问题。如果发现发生了问题，解决方法是使用AtomicStampedReference，在修改值的同时增加一个时间戳，只有当值和时间戳都相同的时候才进行修改，其CAS的方法声明为：

/**
 * Atomically sets the value of both the reference and stamp
 * to the given update values if the
 * current reference is {@code ==} to the expected reference
 * and the current stamp is equal to the expected stamp.
 *
 * @param expectedReference the expected value of the reference
 * @param newReference the new value for the reference
 * @param expectedStamp the expected value of the stamp
 * @param newStamp the new value for the stamp
 * @return {@code true} if successful
 */
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)));
}

比如：

Pair pair = new Pair(100, 200);
int stamp = 1;
AtomicStampedReference<Pair> pairRef = new AtomicStampedReference<Pair>(pair, stamp);
int newStamp = 2;
pairRef.compareAndSet(pair, new Pair(200, 200), stamp, newStamp);

AtomicStampedReference在compareAndSet方法中需要修改两个值，其中一个是引用， 另外一个就是时间戳
。而在内部实现中，AtomicStampedReference会讲两个值组合在一起，修改的是一个值。

Pair<V> current = pair;

AtomicStampedReference将对引用值和时间戳的组合进行比较和修改，转换成为对Pair单个值的比较和修改。

三、总结

CAS是并发包的基础，基于它可以实现高效的、乐观、非阻塞数据结构和算法，也是并发包中锁、同步工具和各种容器的基础。