Java并发包——Atomic操作

Java并发包——Atomic操作

摘要：本文主要学习了Java并发包下的atomic包中有关原子操作的一些类。

部分内容来自以下博客：

https://blog.csdn.net/qq_30379689/article/details/80785650

https://blog.csdn.net/lmb55/article/details/79547685

并发编程的线程安全问题

在并发编程中很容易出现并发安全的问题，有一个很简单的例子就是多线程更新变量i=1，比如多个线程执行i++操作，就有可能获取不到正确的值，而这个问题，最常用的方法是通过加锁操作（比如使用synchronized）进行控制来达到线程安全的目的。但是由于加锁操作采用的是悲观锁策略，并不是特别高效的一种解决方案。

实际上，JUC下的atomic包提供了一系列的操作简单、性能高效、并能保证线程安全的类，来实现原子更新基本类型、原子更新数组、原子更新引用类型以及原子更新对象属性。atomic包下的这些类都是采用的是乐观锁策略去更新数据，在Java中则是使用CAS操作具体实现。

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

CAS操作

乐观锁和悲观锁

悲观锁假设每一次对临界区资源的访问都会发生冲突，当有一个线程访问资源，其他线程就必须等待，所以采用了加锁（比如使用synchronized关键字）的方式解决冲突。

而乐观锁假设线程对资源的访问是没有冲突的，同时所有的线程执行都不需要等待，可以持续执行。如果遇到冲突的话，就使用一种叫做CAS（Compare-and-Swap，比较并交换）的技术来鉴别线程冲突，如果检测到冲突发生，就重试当前操作到没有冲突为止。

什么是CAS

CAS（Compare and Swap，比较并交换）是乐观锁技术（又称为无锁技术），当多个线程尝试使用CAS同时更新同一个变量时，只有其中一个线程能更新变量的值，而其它线程都失败，失败的线程并不会被挂起，而是被告知这次竞争中失败，并可以再次尝试。

原理

CAS包含了三个参数：V，A，B。

其中，V表示要读写的内存位置，A表示旧的预期值，B表示新值。

CAS指令执行时，当且仅当V的值等于预期值A时，才会将V的值设为B，如果V和A不同，说明可能是其他线程做了更新，那么当前线程就什么都不做，最后，CAS返回的是V的真实值。

而在多线程的情况下，当多个线程同时使用CAS操作一个变量时，只有一个会成功并更新值，其余线程均会失败，但失败的线程不会被挂起，而是不断的再次循环重试，这个过程也称为自旋。正是基于这样的原理，CAS即时没有使用锁，也能发现其他线程对当前线程的干扰，从而进行及时的处理。

CAS和synchronized

元老级的Synchronized（未优化前）最主要的问题是在存在线程竞争的情况下会出现线程阻塞和唤醒锁带来的性能问题，因为这是一种互斥同步，也称为阻塞同步。而CAS并不是将线程挂起，当CAS操作失败后会进行一定的尝试，而非进行耗时的挂起唤醒的操作，因此也叫做非阻塞同步。这是两者主要的区别。

CAS和volatile

volatile原理：对于值的操作，会立即更新到主存中，当其他线程获取最新值时会从主存中获取。

atomic原理：对于值的操作，是基于底层硬件处理器提供的原子指令，保证并发时线程的安全。

volatile不是线程安全的，要使volatile变量提供理想的线程安全，必须同时满足下面两个条件：

1）对变量的写操作不依赖于当前值。

2）该变量没有包含在具有其他变量的不变式中。

缺点

CPU开销较大：在并发量比较高的情况下，如果许多线程反复尝试更新某一个变量，却又一直更新不成功，循环往复，会给CPU带来很大的压力。

不能保证代码块的原子性：CAS机制所保证的只是一个变量的原子性操作，而不能保证整个代码块的原子性。比如需要保证3个变量共同进行原子性的更新，就不得不使用synchronized了。

ABA问题：CAS判断变量操作成功的条件是V的值和A的值是一致的，这个逻辑有个小小的缺陷，就是如果V的值一开始为N，A的值读到的也是N，在准备修改为新值前的期间曾经被改成了M，后来又被改回为N，经过两次的线程修改对象的值还是旧值，那么CAS操作就会误任务该变量从来没被修改过。

ABA问题解决办法

Java并发包中提供了一个带有时间戳的对象引用AtomicStampedReference，其内部不仅维护了一个对象值，还维护了一个时间戳，当AtomicStampedReference对应的数值被修改时，除了更新数据本身，还需要更新时间戳，只有对象值和时间戳都满足期望值，才能修改成功。

atomic包下的类

java.util.concurrent.atomic中的类可以分成4组：

标量类：AtomicBoolean，AtomicInteger，AtomicLong，AtomicReference。

数组类：AtomicIntegerArray，AtomicLongArray，AtomicReferenceArray。

更新器类：AtomicLongFieldUpdater，AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater。

复合变量类：AtomicMarkableReference，AtomicStampedReference。

标量类：AtomicBoolean，AtomicInteger，AtomicLong，AtomicReference

AtomicBoolean、AtomicInteger、AtomicLong、AtomicReference这四种基本类型用来处理布尔，整数，长整数，对象四种数据，其内部实现不是简单的使用synchronized，而是一个更为高效的方式CAS（compare and swap）+ volatile和native方法，从而避免了synchronized的高开销，执行效率大为提升。其实例各自提供对相应类型单个变量的访问和更新。每个类也为该类型提供适当的实用工具方法。

数组类：AtomicIntegerArray，AtomicLongArray，AtomicReferenceArray

AtomicIntegerArray、AtomicLongArray、AtomicReferenceArray类进一步扩展了原子操作，对这些类型的数组提供了支持。这些类在为其数组元素提供volatile访问语义方面也引人注目，这对于普通数组来说是不受支持的。其内部并不是像标量类一样维持一个volatile变量，而是全部由native方法实现。数组变量进行volatile没有意义，因此set/get就需要unsafe来做了，但是多了一个index来指定操作数组中的哪一个元素。

更新器类：AtomicLongFieldUpdater，AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater

AtomicReferenceFieldUpdater、AtomicIntegerFieldUpdater、AtomicLongFieldUpdater这三个类是基于反射的实用工具，可以提供对关联字段类型的访问，可用于获取任意选定volatile字段上的compareAndSet操作。它们主要用于原子数据结构中，该结构中同一节点的几个volatile字段都独立受原子更新控制。这些类在如何以及何时使用原子更新方面具有更大的灵活性，但相应的弊端是基于映射的设置较为拙笨、使用不太方便，而且在保证方面也较差。

复合变量类：AtomicMarkableReference，AtomicStampedReference

AtomicMarkableReference类将单个布尔值与引用关联起来。维护带有标记位的对象引用，可以原子方式更新带有标记位的引用类型。

AtomicStampedReference类将整数值与引用关联起来。维护带有整数“标志”的对象引用，可以原子更新带有版本号的引用类型。该类将整数值与引用关联起来，可用于原子的更新数据和版本号，可以解决使用CAS进行原子更新时，可能出现的ABA问题。

AtomicBoolean类源码分析

AtomicBoolean类提供了标量类、数组类、更新器类共同使用的一些方法，所以对这个类进行源码分析。

public class AtomicBoolean implements java.io.Serializable {
    // 设置序列化ID。
    private static final long serialVersionUID = 4654671469794556979L;
    // 使用unsafe进行更新。
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    // 记录旧的预期值，使用volatile关键字修饰来保证可见性并禁止重排序。
    private volatile int value;
    // 记录当前值。
    private static final long valueOffset;
    // 使用unsafe获取当前值。
    static {
        try {
            valueOffset = unsafe.objectFieldOffset(AtomicBoolean.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }

    // 空参的构造方法。
    public AtomicBoolean() {
    }

    // 传入指定值的构造方法。
    public AtomicBoolean(boolean initialValue) {
        value = initialValue ? 1 : 0;
    }

    // 获取预期值。
    public final boolean get() {
        return value != 0;
    }

    // 设置预期值。
    public final void set(boolean newValue) {
        value = newValue ? 1 : 0;
    }

    // 同set()方法类似，但不能保证设置后的值的可见性，也就是不保证能立即被其他线程读取到新设置的值。
    // 此方法通过unsafe的putOrderedInt函数实现，是一个native方法。set()方法使用了volatile关键字修饰的value来保证可见性。
    // 此方法和set()方法的在实现上的区别在于插入的内存屏障不同，set后面会插入store-load屏障，确保能被其他线上立即读取到，而lazySet则只插入store-store屏障，是一种底层的优化手段。
    // 如果是在使用了锁的代码环境下，由锁来保证共享变量的可见性，那么如果用这个方法来修改共享便利，减少不必要的内存屏障，可以提高程序执行的效率。
    public final void lazySet(boolean newValue) {
        int v = newValue ? 1 : 0;
        unsafe.putOrderedInt(this, valueOffset, v);
    }

    // 以原子方式设置当前值，并返回旧的预期值。
    public final boolean getAndSet(boolean newValue) {
        boolean prev;
        do {
            prev = get();
        } while (!compareAndSet(prev, newValue));
        return prev;
    }

    // 如果当前值（valueOffset）等于预期值（expect），则以原子方式将该值设置为更新值（update）。成功返回true，否则返回false，并且不修改原值。
    public final boolean compareAndSet(boolean expect, boolean update) {
        int e = expect ? 1 : 0;
        int u = update ? 1 : 0;
        return unsafe.compareAndSwapInt(this, valueOffset, e, u);
    }

    // 同compareAndSet()方法类似，但不能保证不存在happen-before的发生（也就是可能存在指令重排序导致此操作失败）。
    // 但是从Java源码来看，其实此方法并没有实现JSR规范的要求，最后效果和compareAndSet是等效的，都调用了unsafe.compareAndSwapInt()方法完成操作。
    public boolean weakCompareAndSet(boolean expect, boolean update) {
        int e = expect ? 1 : 0;
        int u = update ? 1 : 0;
        return unsafe.compareAndSwapInt(this, valueOffset, e, u);
    }

    // 返回预期值的字符串形式。
    public String toString() {
        return Boolean.toString(get());
    }
}

AtomicStampedReference类源码分析

AtomicStampedReference类使用整形记录版本，可以判断版本是否被人改过，可以解决CAS操作出现的ABA问题。而AtomicMarkableReference类使用的是布尔类型，不能解决问题，只能降低发生的几率。

public class AtomicStampedReference<V> {
    // 内部Pair类封装了内存值和标记。
    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);
        }
    }

    // 设置pair为volatile类型。
    private volatile Pair<V> pair;

    // 构造方法传入当前值和当前标记。
    public AtomicStampedReference(V initialRef, int initialStamp) {
        pair = Pair.of(initialRef, initialStamp);
    }

    // 获取当前值。
    public V getReference() {
        return pair.reference;
    }

    // 获取当前标记。
    public int getStamp() {
        return pair.stamp;
    }

    // 返回当前值，并且将标记保存在数组首位。
    public V get(int[] stampHolder) {
        Pair<V> pair = this.pair;
        stampHolder[0] = pair.stamp;
        return pair.reference;
    }

    // 设置pair对象。
    public void set(V newReference, int newStamp) {
        Pair<V> current = pair;
        if (newReference != current.reference || newStamp != current.stamp)
            this.pair = Pair.of(newReference, newStamp);
    }

    // 判断当前值和当前标记，则以原子方式更新pair对象。成功返回true，否则返回false，并且不修改原值。
    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)));
    }

    // 同compareAndSet()方法类似。
    public boolean weakCompareAndSet(V   expectedReference, V   newReference, int expectedStamp, int newStamp) {
        return compareAndSet(expectedReference, newReference, expectedStamp, newStamp);
    }

    // 同compareAndSet()方法类似，不过只判断当前值是否和预期值相同，失败的可能性很大。
    public boolean attemptStamp(V expectedReference, int newStamp) {
        Pair<V> current = pair;
        return
            expectedReference == current.reference &&
            (newStamp == current.stamp || casPair(current, Pair.of(expectedReference, newStamp)));
    }

    // 获取Unsafe对象。
    private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();
    // 内存值。
    private static final long pairOffset = objectFieldOffset(UNSAFE, "pair", AtomicStampedReference.class);

    // 使用原子方式更新pair对象。
    private boolean casPair(Pair<V> cmp, Pair<V> val) {
        return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
    }

    // 获取内存值。
    static long objectFieldOffset(sun.misc.Unsafe UNSAFE, String field, Class<?> klazz) {
        try {
            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
        } catch (NoSuchFieldException e) {
            NoSuchFieldError error = new NoSuchFieldError(field);
            error.initCause(e);
            throw error;
        }
    }
}

模拟CAS原子操作的ABA问题

使用三个线程，对AtomicInteger类进行更新，代码如下：

public static void main(String[] args) throws Exception {
    AtomicInteger ab = new AtomicInteger(0);

    Thread t1 = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("t1 ... " + ab + " >>> 1 : " + ab.compareAndSet(0, 1));
        }
    });

    Thread t2 = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("t2 ... " + ab + " >>> 0 : " + ab.compareAndSet(1, 0));
        }
    });

    Thread t3 = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("t3 ... " + ab + " >>> 1 : " + ab.compareAndSet(0, 1));
        }
    });

    t1.start();
    t1.join();
    t2.start();
    t2.join();
    t3.start();
}

运行结果如下：

t1 ... 0 >>> 1 : true
t2 ... 1 >>> 0 : true
t3 ... 0 >>> 1 : true

结果说明：

假设有三个线程，t1将0更新成为1，t2又将1更新成为0，此时t3以为t1和t2没有执行，所以将0更新成为1，结果成功了，但是t3在更新的时候，t1和t2已经更新过了，所以此时应该是更新失败的。

同样是三个线程，使用AtomicStampedReference类进行更新，代码如下：

public static void main(String[] args) throws Exception {
    AtomicStampedReference<Integer> asr = new AtomicStampedReference<Integer>(0, 0);

    Thread t1 = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("t1 ... " + asr.getReference() + " >>> 1 : " + asr.compareAndSet(0, 1, 0, 1));
        }
    });

    Thread t2 = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("t2 ... " + asr.getReference() + " >>> 0 : " + asr.compareAndSet(1, 0, 1, 2));
        }
    });

    Thread t3 = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("t3 ... " + asr.getReference() + " >>> 1 : " + asr.compareAndSet(0, 1, 0, 1));
        }
    });

    t1.start();
    t1.join();
    t2.start();
    t2.join();
    t3.start();
}

运行结果如下：

t1 ... 0 >>> 1 : true
t2 ... 1 >>> 0 : true
t3 ... 0 >>> 1 : false

结果说明：

可以看到，在t1和t2执行完之后，版本号被改成了2，如果t3在更新的时候以为t1和t2没有更新，同时将版本号改为1，结果是失败的，从而禁止了ABA更新的发生。