我所知道的atomic包

以你之姓@ 2022-12-11 04:25 149阅读 0赞

## 一 概述 ##

对于原子操作而言，它是一个不可中断的操作，即使是在多线程环境下也能够保证操作的不可中断性。

它类似与锁，且比锁更加有优势，因为粒度更加细，原子变量可以把竞争范围缩小到变量级别，这是我们可以获得的最小粒度的情况了，通常锁的粒度都比原子变量大。同时，一般情况下使用原子类的效率比使用锁的效率更高，除了某些高度竞争场景。

## 二 原子类的种类 ##

<table> 
 <tbody> 
 <tr> 
 <td colspan="1">Atomic*基本类型原子类</td> 
 <td>AtomicInteger</td> 
 </tr> 
 <tr> 
 <td>AtomicLong</td> 
 </tr> 
 <tr> 
 <td>AtomicBoolean</td> 
 </tr> 
 <tr> 
 <td colspan="1">Atomic*Array数组类型原子类</td> 
 <td>AtomicIntegerArray</td> 
 </tr> 
 <tr> 
 <td>AtomicLongArray</td> 
 </tr> 
 <tr> 
 <td>AtomicReferenceArray</td> 
 </tr> 
 <tr> 
 <td colspan="1">Atomic*Reference引用类型原子类</td> 
 <td>AtomicReference</td> 
 </tr> 
 <tr> 
 <td>AtomicStampedReference</td> 
 </tr> 
 <tr> 
 <td>AtomicMarkableReference</td> 
 </tr> 
 <tr> 
 <td colspan="1">Atomic*FieldUpdater升级类型原子类</td> 
 <td>AtomicIntegerfieldUpdater</td> 
 </tr> 
 <tr> 
 <td>AtomicLongFieldUpdater</td> 
 </tr> 
 <tr> 
 <td>AtomicReferenceFieldUpdater</td> 
 </tr> 
 <tr> 
 <td colspan="1">Adder累加器</td> 
 <td>LongAdder</td> 
 </tr> 
 <tr> 
 <td>DoubleAdder</td> 
 </tr> 
 <tr> 
 <td colspan="1">Accumulator累加器</td> 
 <td>LongAccumulator</td> 
 </tr> 
 <tr> 
 <td>DoubleAccumulator</td> 
 </tr> 
 </tbody> 
</table>

## 三 Atomic\*基本类型原子类 ##

基本数据类型原子类是对基本数据类型基于CAS原理的封装，为基本数据类型提供原子操作的功能，使得在高并发场景中不通过加锁也能够保证线程安全。

**以AtomicInteger为例，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 {
                //通过unsafe工具获取到value在内存中的地址
                valueOffset = unsafe.objectFieldOffset
                    (AtomicInteger.class.getDeclaredField("value"));
            } catch (Exception ex) { throw new Error(ex); }
        }
        
        //获取当前值
        public final int get() {
            return value;
        }
    
        //获取当前值，并设置新的值
        public final int getAndSet(int newValue) {
            return unsafe.getAndSetInt(this, valueOffset, newValue);
        }
        
        //获取当前值并自增，替代非原子操作a++
        public final int getAndIncrement() {
            return unsafe.getAndAddInt(this, valueOffset, 1);
        }
    
        //获取当前值并自减,替代非原子操作a--
        public final int getAndDecrement() {
            return unsafe.getAndAddInt(this, valueOffset, -1);
        }
        
        //获取当前值并加上delta，delta可为负数
        public final int getAndAdd(int delta) {
            return unsafe.getAndAddInt(this, valueOffset, delta);
        }
        
        //当输入的值等于预期值是，则以原子方式将该值更新为值update,CAS思想的示例
        public final boolean compareAndSet(int expect, int update) {
            return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
        }
    }

## 四 Atomic\*Array数组类型原子类 ##

通过基本数据类型数的组类型原子类是对包含有基本数据类型数据的数组基于CAS原理的封装，为基本数据类型数组中的每个基本数据类型提供原子操作的功能，使得在高并发场景中不通过加锁也能够保证线程安全。

**以AtomicIntegerArray为例，AtomicIntegerArray源码**

public class AtomicIntegerArray implements java.io.Serializable {
        private static final long serialVersionUID = 2862133569453604235L;
    
        private static final Unsafe unsafe = Unsafe.getUnsafe();
        private static final int base = unsafe.arrayBaseOffset(int[].class);
        private static final int shift;
        private final int[] array;
    
        static {
            int scale = unsafe.arrayIndexScale(int[].class);
            if ((scale & (scale - 1)) != 0)
                throw new Error("data type scale not a power of two");
            shift = 31 - Integer.numberOfLeadingZeros(scale);
        }
    
    }

## 五 AtomicReference引用类型原子类 ##

引用类型原子类AtomicReference的作用和AtomicInteger并没有本质的区别，AtomicInteger可以让一个正数保证原子性，而AtomicReference可以让一个对象保证原子性，所以用法和AtomicInteger类似，当然，AtomicReference的功能明显要比AtomicInteger强，因为一个对象中可以包含很多属性。

**AtomicReference源码**

public class AtomicReference<V> implements java.io.Serializable {
 private static final long serialVersionUID = -1848883965231344442L;
 
 private static final Unsafe unsafe = Unsafe.getUnsafe();
 private static final long valueOffset;
 
 static {
 try {
 valueOffset = unsafe.objectFieldOffset
 (AtomicReference.class.getDeclaredField("value"));
 } catch (Exception ex) { throw new Error(ex); }
 }
 
 //如果目前的值与预期的相同，则将其原子的更新为update
 public final boolean compareAndSet(V expect, V update) {
 return unsafe.compareAndSwapObject(this, valueOffset, expect, update);
 }
 }

**AtomicReference实现自旋锁**

private AtomicReference<Thread> sign = new AtomicReference<>();
 
 public void lock() {
 Thread current = Thread.currentThread();
 /*当前线程获取锁时，先判断当前锁的线程是否null，如果为null则将锁的线程设置为当前线程
 此时如果存在其他线程获取该锁，由于它已经被当前线程所持有就会进行自旋操作直到该锁被释放
 ，它获得该锁*/
 while (!sign.compareAndSet(null, current)) {
 System.out.println("自旋获取失败，再次尝试");
 }
 }
 
 ///*当前线程释放锁时，先判断当前锁的线程是否被当前线程锁持有，如果是则将当前线程的锁释放，
 此时如果存在其他线程自旋等待该锁时，就会尝试获得该锁*/
 public void unlock() {
 Thread current = Thread.currentThread();
 sign.compareAndSet(current, null);
 }

## 六 AtomicIntegerFieldUpdater ##

AtomicIntegerFieldUpdater将普通变量升级为原子变量，由于原子类提供的原子操作是存在性能消耗的 ，所以当我们不需要全局使用原子操作时，我们可以利用AtomicIntegerFieldUpdater对某些需要保证原子操作的非原子操作升级。

**AtomicIntegerFieldUpdater源码**

public abstract class AtomicIntegerFieldUpdater<T> {
 
 @CallerSensitive
 public static AtomicIntegerFieldUpdater newUpdater(Class tclass, String fieldName) {
 return new AtomicIntegerFieldUpdaterImpl
 (tclass, fieldName, Reflection.getCallerClass());
 }
 
 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);
 }
 
 if (field.getType() != int.class)
 throw new IllegalArgumentException("Must be integer type");
 
 if (!Modifier.isVolatile(modifiers))
 throw new IllegalArgumentException("Must be volatile type");
 
 this.cclass = (Modifier.isProtected(modifiers) &&
 tclass.isAssignableFrom(caller) &&
 !isSamePackage(tclass, caller))
 ? caller : tclass;
 this.tclass = tclass;
 this.offset = U.objectFieldOffset(field);
 }
 }

根据AtomicIntegerFieldUpdater源码可知，通过AtomicIntegerFieldUpdater newUpdater(Class tclass, String fieldName)方法可以将tclass类（U）的某个字段由非原子操作升级为原子操作，底层时通过反射实现该功能。

值得注意的是，需要升级的变量不应该是私有变量和静态变量即不应该为被private和static关键字锁修饰的变量。

**private修饰的变量**

![20200928164422994.png][]

**static修饰的变量**

![20200928164644561.png][]

## 七 Adder累加器 ##

Adder累加器是从JDK1.8中引入，在高并发的情况下，LongAdder比AtomicLong的效率高。实质上是通过空间换时间的方法，当线程竞争激烈的时候，LongAdder把不同的线程对应到不同的Cell【内部结构】上进行修改，从而降低了不同线程之间发生冲突的概率，也是多段锁的理念【类似于JDK1.7中的ConcurrentHashMap中Segment使用多段锁原理】，提高了并发性能。

AtomicLong在多线程同时操作时，由于竞争激烈，每一次的原子操作都需要flush和refresh，导致很消耗资源。

AtomicLong操作如果在CPU中的某个核心中修改了数据后为了保证不同核心之间的数据一致性，会将数据修改后的内容flush到主内存（共享内存）中，再通过refresh将其共享到每个工作核心中。

每个线程都有自己的工作内存。AtomicLong在多线程的情况下会一致维护着不同线程于不同核心之间的的数据同步，所以非常消耗性能。

**AtomicLong的数据同步原理图**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NhbG1fZW5jb2Rl_size_16_color_FFFFFF_t_70][]

LongAdder不像AtomicLong一样，每次数据的更改都需要做同步，从而减少了高并发中线程之间的冲突，提高了工作效率。在LongAdder中每个线程都会拥有一个自己的计数器，仅用来在自己的线程内计数，使得不同线程之间不会相互干扰。

**AtomicAdder的原理图**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NhbG1fZW5jb2Rl_size_16_color_FFFFFF_t_70 1][]

**分段过程：**在第一个线程的计数器数值ctr'为1时，可能其他线程的计数器ctr“已经为3了，它们之间并不存在竞争关系，所以在数据的操作的过程中，并不需要同步机制，所以也不需要flush和refresh操作来保证数据同步。每个线程的计数器是完全独立的，所以LongAdder是通过空间换时间。

**汇总过程：**LongAdder通过分段累加的方法，内部存在一个base变量和一个cell\[\]数组来共同参与计数。在竞争不激烈的时候，直接累加到base变量上；当竞争很激烈的时候，各个线程分散累加到自己的cell\[i\]中，每个cpu核心都是cell\[\]中的一个成员，而每个线程都会累加到自己对应的cell\[i\]中，对应的时候是通过Hash运算，根据运算得到的不同值将不同的线程分配到不同的cell\[i\]，每个cell\[i\]都是不同的计数器，从而避免了不同线程之间的干扰，不同的cell\[i\]之间也不存在不同竞争关系。

**LongAdder的sum方法源码解析**

//为加锁 
 public long sum() {
 //Cell数组是竞争激烈的时候，各个线程会分散到累加自己的Cell[i]中，Cell[]中的每个Cell[i]对应一个线程
 Cell[] as = cells; Cell a;
 //base变量是在线程竞争不激烈的情况下，累加操作直接作用在该变量上
 long sum = base;
 if (as != null) {
 for (int i = 0; i < as.length; ++i) {
 if ((a = as[i]) != null)
 sum += a.value; //sum增加每个cell中的值，最终为base和每个cell中的值累加和
 }
 }
 return sum;
 }

**AtomicLong与LongAdder的使用情况**

在低争用的时候，AtomicLong和LongAdder这两个类具有类似的特征，当时在竞争的激烈的时候，LongAdder的吞吐量要高的多，但是要消耗更多的空间。

LongAdder适合用在统计求和和计数的场景，而且LongAdder基本只提供了add方法，而AtomicLong具有cas方法。

## 八 Accumulator累加器 ##

Accumulator累加器和Adder均为JDK1.8中新增的原子类，Accumulator比Adder的适用范围更广，Adder适用于加操作而Accmulator还适合乘除等操作，它的实现类似于数学归纳法。

**以LongAccumulator为例，LongAccumulator源码**

public class LongAccumulator extends Striped64 implements Serializable {
        private static final long serialVersionUID = 7249069246863182397L;
    
        private final LongBinaryOperator function;
        private final long identity;
    
        public LongAccumulator(LongBinaryOperator accumulatorFunction,
                               long identity) {
            this.function = accumulatorFunction;
            base = this.identity = identity;
        }
        
        //面向函数编程
        @FunctionalInterface
        public interface LongBinaryOperator {
    
            long applyAsLong(long left, long right);
        }
    }

从源码可以看出，在JDK1.8中引入了功能性接口，以支持lambda表达式，操作实例代码。

public class AccumulatorTest {
        /*
        * (x,y)->x+y Lambda表达式
        * Lambda表达式的参数列表的数据类型可以省略不写，因为JVM编译器会通过上下文推断出来，数据类型，即“类型推断”
        * 将表达式中的x作为初始值，然后将y作为算数值
        * 第一步将identity=0赋值给x
        * 第二步将算式的结果赋值给y即y =（x + y）
        * 第三步将1赋值给x，然后执行y = 1 + 0 = 1
        * 数学归纳法
        * */
        public static void main(String[] args) {
            //初始化
            LongAccumulator accumulator = new LongAccumulator((x,y)->x+y,0);
            //为x重新赋值
            accumulator.accumulate(1);
            System.out.println("1+0的最终结果为："+accumulator.getThenReset());
            //利用线程池
            ExecutorService executorService = Executors.newFixedThreadPool(8);
            //range(1,10) 等价 [1,10) 等价 [1,9]
            IntStream.range(1,10).forEach(i->executorService.submit(()->accumulator.accumulate(i)));
            //等待线程池中所有的线程执行完
            executorService.shutdown();
            //判断线程池中所有的线程是否执行完
            while (!executorService.isTerminated()){
    
            }
            System.out.println("1~9累加的最终结果为："+accumulator.getThenReset());
        }
    }

**结果**

![20200928195328643.png][]

Accumulator适用于并行计算，利用计算机的多核运算，当然这种运算是没有顺序的。如果希望数据按顺序进行计算我们可以通过for循环实现顺序执行。

[20200928164422994.png]: /images/20221123/b8018f76dade4cf991e2bedc1e5d9cd6.png
[20200928164644561.png]: /images/20221123/b2e025caeae1491a96aef420a1116e6a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NhbG1fZW5jb2Rl_size_16_color_FFFFFF_t_70]: /images/20221123/920cc16f19834033981b9f04e1dc8662.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NhbG1fZW5jb2Rl_size_16_color_FFFFFF_t_70 1]: /images/20221123/8e18c2bdd235401194482d06e8b2c63c.png
[20200928195328643.png]: /images/20221123/76de6aa73b73499fba3f878ccfe97164.png