JUC Atomic原子类分析

朱雀 2022-05-23 13:26 272阅读 0赞

## 基本介绍 ##

JUC中atomic子包里面提供了很多Atomic类，用于多线程并发下原子更新（CAS）不同类型对象，主要包括以下几大类:

### 原子更新基本类型类 ###

*  AtomicInteger
 *  AtomicLong
 *  AtomicBoolean

### 原子更新数组类型 ###

*  AtomicIntegerArray
 *  AtomicLongArray
 *  AtomicReferenceArray

### 原子更新引用类型类 ###

*  AtomicReference
 *  AtomicStampedReference
 *  AtomicMarkableReference

### 原子更新字段类 ###

*  AtomicIntegerFieldUpdater
 *  AtomicLongFieldUpdater
 *  AtomicReferenceFieldUpdater

> 其中原子更新数组类型貌似不太常用

所有原子更新类的实现依赖Unsafe的以下方法:

*  compareAndSwapInt(Object o, long offset, int expected, int x)
 *  compareAndSwapLong(Object o, long offset, long expected, long x)
 *  compareAndSwapObject(Object o, long offset, Object expected, Object x)

实现的通用方法:

*  lazySet
 *  compareAndSet
 *  weakCompareAndSet
 *  set
 *  get

对于AtomicInteger和AtomicLong，还提供了原子自增、自减等方法:

*  getAndIncrement
 *  incrementAndGet
 *  getAndAdd
 *  addAndGet
 *  getAndDecrement
 *  decrementAndGet

## Examples ##

下面提供一两个atomic类的使用示例

### AtomicInteger ###

public class AtomicIntegerTest { 
    
        public static void main(String[] args) throws InterruptedException {
    
            final AtomicInteger value = new AtomicInteger(0);
            Runnable r = new Runnable() {
    
                @Override
                public void run() {
                    for (int i = 0; i < 100; i++) {
                        value.incrementAndGet();
                    }
                }
            };
            Thread thd1 = new Thread(r, "thd1");
            Thread thd2 = new Thread(r, "thd2");
            thd1.start();
            thd2.start();
    
            Thread.sleep(50);
            thd1.join();
            thd2.join();
    
            System.out.println("value : " + value.get());
        }
    
    }

最后输出:

value : 200

### AtomicIntegerFieldUpdater ###

public class AtomicIntegerFieldUpdaterTest { 
    
        private volatile int intVolatileVal;
        // 定义在同一个类里面，调用者即目标类，可访问到private类型的intVolatileVal
        private static final AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> updater1 = AtomicIntegerFieldUpdater.newUpdater(AtomicIntegerFieldUpdaterTest.class, "intVolatileVal");
    
        public int update(int val) {
            return updater1.getAndSet(this, val);
        }
    
        public static void main(String[] args) {
    
            AtomicIntegerFieldUpdaterTest t = new AtomicIntegerFieldUpdaterTest();
            System.out.println(t.update(3));
            System.out.println(t.update(4));
            System.out.println(t.update(5));
        }
    
    }

输出:

0
    3
    4

## Atomic类的实现分析 ##

### 基本类型类 + AtomicReference ###

这部分类的实现有一定通用的套路，我们拿AtomicInteger来分析（下面源码摘自AtomicInteger）:

private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final long valueOffset;
    private volatile int value;
    
    static {
        try {
            valueOffset = unsafe.objectFieldOffset
                (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }

*  持有unsafe实例，负责CAS操作
 *  定义volatile类型的value存储值（对于AtomicInteger，value类型为int，对于AtomicLong，value类型为long）
 *  确定value在该类内存地址中的内存偏移地址valueOffset

compareAndSet使用unsafe提供的compareAndSwap(X)方法来实现，如AtomicInteger，调用的是:

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

由于value为volatile，提供多线程的可见性，因此get和set方法跟普通get和set方法一致:

public final int get() {
        return value;
    }
    
    public final void set(int newValue) {
        value = newValue;
    }

对于CAS自增和自减，使用循环CAS方式实现，如下为`AtomicInteger`的getAndIncrement方法实现:

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

> Tips: JDK 1.8 中 atomicinteger对于自增的实现，采用unsafe.getAndAddInt，取消了循环CAS带来的性能损耗?

### 引用字段更新类 ###

相比上面的实现复杂一点，这里拿AtomicIntegerFieldUpdater来分析，首先它是一个抽象类，构造方式为使用其提供的静态方法`newUpdater`:

public static <U> AtomicIntegerFieldUpdater<U> newUpdater(Class<U> tclass,
                                                              String fieldName) {
        return new AtomicIntegerFieldUpdaterImpl<U>
            (tclass, fieldName, Reflection.getCallerClass());
    }

*  tclass : 持有该原子更新字段的类对象
 *  fieldName : 需要实现原子更新的字段名

注意，对于需要实现原子更新的字段，必须定义为volatile，并且类型必须正确，如上分析的字段必须为Integer，在AtomicIntgerFieldUpdaterImpl实现类中将会做这部分检查:

AtomicIntegerFieldUpdaterImpl(final Class<T> tclass,
                                      final String fieldName,
                                      final Class<?> caller) {
            final Field field;
            final int modifiers;
            try {
                field = AccessController.doPrivileged(
                    new PrivilegedExceptionAction<Field>() {
                        public Field run() throws NoSuchFieldException {
                            return tclass.getDeclaredField(fieldName);
                        }
                    });
                modifiers = field.getModifiers();
                // 这里会检查调用者对目标字段是否有访问权限
                sun.reflect.misc.ReflectUtil.ensureMemberAccess(
                    caller, tclass, null, modifiers);
                ClassLoader cl = tclass.getClassLoader();
                ClassLoader ccl = caller.getClassLoader();
                if ((ccl != null) && (ccl != cl) &&
                    ((cl == null) || !isAncestor(cl, ccl))) {
                    sun.reflect.misc.ReflectUtil.checkPackageAccess(tclass);
                }
            } catch (PrivilegedActionException pae) {
                throw new RuntimeException(pae.getException());
            } catch (Exception ex) {
                throw new RuntimeException(ex);
            }
            // 检查字段类型为integer
            if (field.getType() != int.class)
                throw new IllegalArgumentException("Must be integer type");
            // 检查字段必须为volatile
            if (!Modifier.isVolatile(modifiers))
                throw new IllegalArgumentException("Must be volatile type");
    
            this.cclass = (Modifier.isProtected(modifiers)) ? caller : tclass;
            this.tclass = tclass;
            this.offset = U.objectFieldOffset(field);
        }

在`AtomicIntegerFieldUpdaterImpl`实现了compareAndSet等CAS更新方法，同时包括CAS自增、CAS自减等操作，其中注意对于get和set方法，需要调用Unsafe的getIntVolatile和setIntVolatile:

public final int get(T obj) {
        accessCheck(obj);
        return U.getIntVolatile(obj, offset);
    }
    
    public final void set(T obj, int newValue) {
        accessCheck(obj);
        U.putIntVolatile(obj, offset, newValue);
    }

同时注意lazySet的实现:

public final void lazySet(T obj, int newValue) {
        accessCheck(obj);
        U.putOrderedInt(obj, offset, newValue);
    }

## 一些疑惑点 ##

1） weakCompareAndSet 与 compareAndSet 区别？

代码实现上一样（摘自AtomicInteger）:

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

区别在于:

> weakCompareAndSet atomically reads and conditionally writes a variable but does not create any happens-before orderings, so provides no guarantees with respect to previous or subsequent reads and writes of any variables other than the target of the weakCompareAndSet.
> 
> compareAndSet and all other read-and-update operations such as getAndIncrement have the memory effects of both reading and writing volatile variables.

说的是`weakCompareAndSet`对变量的操作并没有严格按照`happens-before`规则来进行，那么可能会在对该变量执行操作前后出现重排序（可以理解为该变量在weakCAS下不具备volatile的语义？），而`compareAndSet`是严格按照`happens-before`规则来进行

2）lazySet 是什么鬼 ？

public final void lazySet(int newValue) {
        unsafe.putOrderedInt(this, valueOffset, newValue);
    }

> As probably the last little JSR166 follow-up for Mustang, we added a “lazySet” method to the Atomic classes (AtomicInteger, AtomicReference, etc). This is a niche method that is sometimes useful when fine-tuning code using non-blocking data structures. The semantics are that the write is guaranteed not to be re-ordered with any previous write, but may be reordered with subsequent operations (or equivalently, might not be visible to other threads) until some other volatile write or synchronizing action occurs).
> 
> The main use case is for nulling out fields of nodes in non-blocking data structures solely for the sake of avoiding long-term garbage retention; it applies when it is harmless if other threads see non-null values for a while, but you’d like to ensure that structures are eventually GCable. In such cases, you can get better performance by avoiding the costs of the null volatile-write. There are a few other use cases along these lines for non-reference-based atomics as well, so the method is supported across all of the AtomicX classes.
> 
> For people who like to think of these operations in terms of machine-level barriers on common multiprocessors, lazySet provides a preceeding store-store barrier (which is either a no-op or very cheap on current platforms), but no store-load barrier (which is usually the expensive part of a volatile-write).

取消store-load内存屏障，对于volatile 变量的写，不必立刻刷到主内存，因此可能导致其他线程在一定时间内会取到旧的value

## CAS中的ABA问题以及应对方案 ##

CAS中有一个比较经典的困境，会遇到ABA问题，这个问题如下描述：

一个变量a = A，一个线程thread1将对其进行CAS操作，期望将它的值从A变成C，不巧的是，另外一个修改线程thread2在thread1执行CAS的途中，将a的值由A变成B，然后再变回A，之后thread1的CAS看不到这个变化，因为最终a的值还是为A，是期望的值，所以thread1的CAS最终成功，但是中间过程的变化，被忽略了！

ABA问题在某些场景下会引起一些问题，为了解决ABA问题，引入版本号的概念，CAS的时候必须带上版本号进行判断，若不是预期的版本号，即使值一样，CAS更新也是失败的

JUC中提供的`AtomicStampedReference`就是为了解决ABA问题而存在的，该类中的CAS操作需要额外传递版本号，定义如下:

public boolean compareAndSet(V   expectedReference,
                                     V   newReference,
                                     int expectedStamp,
                                     int newStamp) {
        // ...
    
    }

## Atomic\*FieldUpdater的妙用 ##

当一个类中包含很多个成员变量需要做到原子更新，比如一个类聚合了很多个AtomicLong成员变量，用于实现一些状态的原子更新，当程序中这个类的创建达到一定的量级别，由于聚合了很多atomic类，那么内存会吃得比较猛。此时若使用`Atomic*FieldUpdater`，可在一定程度上降低内存开销！

> 将这些原子更新类变成volatile变量，然后定义静态(static)的Atomic\*FieldUpdater来实现这些volatile变量的原子更新

具体的分析可参考Netty核心开发者的一篇博客：[Lesser known concurrent classes - Atomic\*FieldUpdater][Lesser known concurrent classes - Atomic_FieldUpdater]；使用场景可以看下阿里开源连接池[druid][]里面的一个类[JdbcSqlStat][]（其他一些Stat类也用到），里面用了大量的AtomicIntegerFieldUpdater和AtomicLongFieldUpdater，用于原子更新定义的volatile int和long变量（都是一些采集状态字段）

## 参考 ##

*  [StackOverflow: how-can-weakcompareandset-fail-spuriously-if-it-is-implemented-exactly-like-comp][StackOverflow_ how-can-weakcompareandset-fail-spuriously-if-it-is-implemented-exactly-like-comp]
 *  [The JSR-133 Cookbook for Compiler Writers - Doug Lea][]
 *  [并发编程网: JUC中Atomic class之lazySet的一点疑惑][JUC_Atomic class_lazySet]
 *  [【不简单的Atomic工具】lazySet & set][Atomic_lazySet _ set]
 *  [J.U.C原子工具类AtomicXXX中，set和lazySet的区别][J.U.C_AtomicXXX_set_lazySet]

[Lesser known concurrent classes - Atomic_FieldUpdater]: http://normanmaurer.me/blog/2013/10/28/Lesser-known-concurrent-classes-Part-1/
[druid]: https://github.com/alibaba/druid
[JdbcSqlStat]: https://github.com/alibaba/druid/blob/master/src/main/java/com/alibaba/druid/stat/JdbcSqlStat.java
[StackOverflow_ how-can-weakcompareandset-fail-spuriously-if-it-is-implemented-exactly-like-comp]: https://stackoverflow.com/questions/2443239/how-can-weakcompareandset-fail-spuriously-if-it-is-implemented-exactly-like-comp
[The JSR-133 Cookbook for Compiler Writers - Doug Lea]: http://g.oswego.edu/dl/jmm/cookbook.html
[JUC_Atomic class_lazySet]: http://ifeve.com/juc-atomic-class-lazyset-que/
[Atomic_lazySet _ set]: http://zzzvvvxxxd.github.io/2016/03/24/atomic1/
[J.U.C_AtomicXXX_set_lazySet]: http://blog.csdn.net/aitangyong/article/details/41577503