Java - 原子变量和CAS

绝地灬酷狼 2022-05-27 10:07 282阅读 0赞

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