【JUC】原子操作类及LongAddr源码分析

1. 十八罗汉

底层使用Unsafe类的CAS方法，而无需使用synchronized等重量锁使操作变得线程安全

1. AtomicBoolean
2. AtomicInteger
3. AtomicIntegerArray
4. AtomicIntegerFieldUpdater
5. AtomicLong
6. AtomicLongArray
7. AtomicLongFieldUpdater
8. AtomicMarkableReference
9. AtomicReference
10. AtomicReferenceArray
11. AtomicReferenceFieldUpdater
12. AtomicStampedReference
13. DoubleAccumulator
14. DoubleAdder
15. LongAccumulator
16. LongAdder
17. Striped64，LongAdder是Striped64的子类
18. Number，Striped64，AtomicInteger等是Number的子类

2. 原子类再分类

2.1 基本类型原子类

1. AtomicBoolean
2. AtomicInteger
3. AtomicLong

基本使用

public class Temp {static int num = 0;static AtomicInteger atomicNum = new AtomicInteger();public static void main(String[] args) throws InterruptedException {int size = 50;CountDownLatch count = new CountDownLatch(size);for (int i = 0; i < size; i++) {new Thread(() -> {try {for (int j = 0; j < 1000; j++) {num++;atomicNum.getAndIncrement();}} finally {count.countDown();}}).start();}count.await();System.out.println(num);System.out.println(atomicNum.get());}
}

输出

39224
50000

2.2 数组类型原子类

如何理解？类比数组即可

1. AtomicIntegerArray
2. AtomicLongArray
3. AtomicReferenceArray

2.3 引用类型原子类

1. AtomicReference
2. AtomicStampedReference
   • 用版本号解决CAS的ABA问题，可以统计出修改过几次
3. AtomicMarkableReference
   • 用状态戳解决CAS的ABA问题，可以标记出是否修改过

2.4 对象的属性修改原子类

1. AtomicIntegerFieldUpdater：原子更新对象中int类型字段的值
2. AtomicLongFieldUpdater：原子更新对象中Long类型字段的值
3. AtomicReferenceFieldUpdater：原子更新对象中引用类型字段的值

目的：以一种线程安全的方式操作非线程安全对象内的某些字段

要求：

• 更新的对象属性必须使用public volatile修饰
• 使用静态放法newUpdater()创建一个更新器，并且需要设置想要更新的类和属性

/*** 需求：多线程环境下初始化资源类，要求只能初始化一次*/
public class Temp {public volatile Boolean isInit = Boolean.FALSE;private static final AtomicReferenceFieldUpdater<Temp, Boolean> ATOMIC_REFERENCE_FIELD_UPDATER =AtomicReferenceFieldUpdater.newUpdater(Temp.class, Boolean.class, "isInit");public void init(Temp temp) {if (ATOMIC_REFERENCE_FIELD_UPDATER.compareAndSet(temp, Boolean.FALSE, Boolean.TRUE)) {System.out.println(Thread.currentThread().getName()+":init start");try { TimeUnit.SECONDS.sleep(1); } catch (Exception e) {}System.out.println(Thread.currentThread().getName()+":init finished");} else {System.out.println(Thread.currentThread().getName()+":已有线程正在初始化");}}public static void main(String[] args) throws InterruptedException {int size = 5;CountDownLatch count = new CountDownLatch(size);Temp temp = new Temp();for (int i = 0; i < size; i++) {new Thread(() -> {try {temp.init(temp);} finally {count.countDown();}}).start();}count.await();}
}

2.5 原子操作增强类

1. DoubleAccumulator：一个或多个变量共同维护使用的函数更新的运行double值
2. DoubleAdder：一个或多个变量共同维持最初的零和double总和
3. LongAccumulator：一个或多个变量共同维护使用的函数更新的运行long值
4. LongAdder：一个或多个变量共同维持最初的零和long总和

volatile 解决多线程内存不可见问题。对于一写多读，是可以解决变量同步问题，但是如果多写，同样无法解决线程安全问题。

说明: 如果是 count++操作，使用如下类实现: AtomicInteger count = new AtomicInteger0;count.addAndGet(1);如果是JDK8，推荐使用 LongAdder 对象，比 AtomicLong 性能更好( 减少乐观锁的重试次数 )；同时，使用LongAdder的空间代价更大，故越是高并发越推荐使用LongAdder 。

LongAdder只能用来计算加法，且从零开始计算；

LongAccumulator提供了自定义函数的操作且可以传入初始值。

3. 代码演示及性能比较：

/*** 需求：点赞数统计，50个线程，每个点赞1千万次*/
public class Temp {long i = 0;public synchronized void synchronizedIncrement() {i++;}AtomicLong atomicLong = new AtomicLong();public void atomicLongIncrement() {atomicLong.getAndIncrement();}LongAdder longAdder = new LongAdder();public void longAdderIncrement() {longAdder.increment();}LongAccumulator longAccumulator = new LongAccumulator(Long::sum, 0);public void longAccumulatorIncrement() {longAdder.increment();}public static void main(String[] args) throws InterruptedException {int threadNum = 50;int times = 10000000;process(threadNum, times, 1);process(threadNum, times, 2);process(threadNum, times, 3);process(threadNum, times, 4);}public static void process(int threadNum, int times, int type) throws InterruptedException {long startTime = System.currentTimeMillis();CountDownLatch countDownLatch = new CountDownLatch(threadNum);Temp temp = new Temp();for (int i = 0; i < threadNum; i++) {new Thread(() -> {try {for (int j = 0; j < times; j++) {switch (type) {case 1: temp.synchronizedIncrement();break;case 2: temp.atomicLongIncrement();break;case 3: temp.longAdderIncrement();break;case 4: temp.longAccumulatorIncrement();break;default: throw new RuntimeException("不存在的类型");}}} finally {countDownLatch.countDown();}}).start();}countDownLatch.await();long endTime = System.currentTimeMillis();System.out.println("costTime:" + (endTime - startTime) + " 毫秒，" + typeStr(type));}public static String typeStr(int type) {switch (type) {case 1: return "synchronizedIncrement";case 2: return "atomicLongIncrement";case 3: return "longAdderIncrement";case 4: return "longAccumulatorIncrement";default: throw new RuntimeException("不存在的类型");}}
}

执行结果：

costTime:22854 毫秒，synchronizedIncrement
costTime:4888 毫秒，atomicLongIncrement
costTime:215 毫秒，longAdderIncrement
costTime:214 毫秒，longAccumulatorIncrement

4. LongAddr原理

LongAddr继承了Striped64，LongAdder的基本思路就是分散热点，将value值分散到一个Cell数组中，不同线程会命中到数组的不同槽中，各个线程只对自己槽中的那个值进行CAS操作，这样热点就被分散了，冲突的概率就小很多。如果要获取真正的long值，只要将各个槽中的变量值累加返回

sum()会将所有Cell数组中的value和base累加作为返回值，核心的思想就是将之前AtomicLong一个value的更新压力分散到多个value中去，
从而降级更新热点。

Striped64比较重要的成员变量

    /** Number of CPUS, to place bound on table size *  CPU数量，即cells数组的最大长度*/static final int NCPU = Runtime.getRuntime().availableProcessors();/*** Table of cells. When non-null, size is a power of 2.* cells数组，长度为2的幂，2，4，8，16...，方便以后位运算*/transient volatile Cell[] cells;/*** Base value, used mainly when there is no contention, but also as* a fallback during table initialization races. Updated via CAS.* 基础value值，当并发较低时，只累加该值，主要用于没有竞争的情况，通过CAS更新*/transient volatile long base;/*** Spinlock (locked via CAS) used when resizing and/or creating Cells.* 创建或扩容Cells数组时使用的自旋锁变量调整单元格大小(扩容)，创建单元格时使用的锁*/transient volatile int cellsBusy;

5. LongAddr源码分析

小总结：LongAdder在无竞争的情况，跟AtomicLong一样，对同一个base进行操作，当出现竞争关系时则是采用化整为零分散热点的做法，用空间换时间用一个数组cells，将一个value拆分进这个数组cells。多个线程需要同时对value进行换作时候，可以对线程id进行hash得到hash值，再根据hash值映射到这个数组cells的某个下标，再对该下标所对应的值进行自增操作。当所有线程换作完毕，将数组cells的所有值和base都加起来作为最终结果

5.1 add()

public void add(long x) {// as是Striped64中的cells数组属性// b是Striped64中的base属性// v是当前线程hash到Cell中存储的值// m是cells的长度-1，hash时作为掩码使用// a是当前线程hash到的CellCell[] as; long b, v; int m; Cell a;// 首次首线程(as = cells) != null一定是false，此时走casBase方法，以CAS的方式更新base值，且只有当cas失败时，才会走到if中// 		条件1： cells不为空// 		条件2： cas操作base失败，说明其他线程先一步修改了base，出现竞争if ((as = cells) != null || !casBase(b = base, b + x)) {// true无竞争，false表示竞争激烈，多个线程hash到同一个cell，可能要扩容boolean uncontended = true;// 条件1：cells为空// 条件2：应该不会出现// 条件3：当前线程所在cell为空，说明当前线程还没有更新过cell，应初始化一个cell// 条件4：更新当前线程所在cell失败，说明竞争很激烈，多个线程hash到了同一个cell，应扩容if (as == null || (m = as.length - 1) < 0 ||// getProbe()方法返回的是线程中的threadLocalRandomProbe字段// 它是通过随机数生成的一个值，对于一个确定的线程这个值是固定的(除非刻意修改它)(a = as[getProbe() & m]) == null ||!(uncontended = a.cas(v = a.value, v + x)))longAccumulate(x, null, uncontended);}
}

5.2 longAccumulate()

    /*** Handles cases of updates involving initialization, resizing,* creating new Cells, and/or contention. See above for* explanation. This method suffers the usual non-modularity* problems of optimistic retry code, relying on rechecked sets of* reads.** @param x the value * @param fn the update function, or null for add (this convention* avoids the need for an extra field or function in LongAdder). * @param wasUncontended false if CAS failed before call * x: 需要增加的值，一般默认都是1* fn: 默认传递的是null* wasUncontended: 竞争标识，如果是false标识有竞争。只有cells初始化之后，并且当前线程CAS修改失败才会是false*/final void longAccumulate(long x, LongBinaryOperator fn,boolean wasUncontended) {// h：存储线程的probe值int h;// 如果getProbe()方法返回0，说明随机数未初始化if ((h = getProbe()) == 0) { // 这个操作相当于给当前线程生成一个非0的hash值// 使用ThreadLocalRandom为当前线程重新计算一个hash值，强制初始化ThreadLocalRandom.current(); // force initialization// 重新获取probe值，hash值被重置就好比一个全新的线程一样，所以设置了wasUncontended竞争状态为trueh = getProbe();// 重新计算了当前线程的hash后认为此次不算是一次竞争，都未初始化，肯定还不存在竞争激烈，故设置wasUncontended竞争状态为truewasUncontended = true;}// 如果hash取模映射得到的cell单元不是null，则为true，此值也可以看做是扩容意向boolean collide = false;                // True if last slot nonemptyfor (;;) {// 自旋：按CASE 2,3,1看代码Cell[] as; Cell a; int n; long v;// CASE 1：cells已经被初始化了if ((as = cells) != null && (n = as.length) > 0) {// CASE 1.1 :当前线程hash到的cell还未初始化，则需要进行初始化处理，初始化的值为x即默认的1if ((a = as[(n - 1) & h]) == null) {if (cellsBusy == 0) {       // Try to attach new CellCell r = new Cell(x);   // Optimistically create// 双端检查并尝试抢锁if (cellsBusy == 0 && casCellsBusy()) {boolean created = false;try {               // Recheck under lockCell[] rs; int m, j;if ((rs = cells) != null &&(m = rs.length) > 0 &&rs[j = (m - 1) & h] == null) {rs[j] = r;created = true;}} finally {cellsBusy = 0;}if (created)break;continue;           // Slot is now non-empty}}collide = false;}// CASE 1.2：竞争激烈，允许重新计算hash值else if (!wasUncontended)       // CAS already known to failwasUncontended = true;      // Continue after rehash// CASE 1.3：当前线程hash到已初始化的槽位，使用cas进行更新，更新成功则跳出循环else if (a.cas(v = a.value, ((fn == null) ? v + x :fn.applyAsLong(v, x))))break;// CASE 1.4：如果cells的长度已经大于等于CPU核数，就不要再扩容了,继续hash;cells != as说明已有线程正在扩容else if (n >= NCPU || cells != as)collide = false;            // At max size or stale// CASE 1.5：将扩容意向置为true，如果再hash仍然不满足上诉条件便扩容else if (!collide)collide = true;// CASE 1.6：尝试扩容，将老数组复制到新数组上else if (cellsBusy == 0 && casCellsBusy()) {try {// 当前的cells数组和最先赋值的as是同一个，代表没有被其他线程扩容过if (cells == as) {      // Expand table unless staleCell[] rs = new Cell[n << 1];for (int i = 0; i < n; ++i)rs[i] = as[i];cells = rs;}} finally {cellsBusy = 0;}collide = false;continue;                   // Retry with expanded table}// 允许重置当前线程的hash值h = advanceProbe(h);}// CASE 2：cells没有加锁且没有初始化，则尝试对它进行加锁，并初始化cells数组;cellsBusy=0表示无锁状态，cellsBusy=1为持锁状态else if (cellsBusy == 0 && cells == as && casCellsBusy()) {// boolean init = false;try {                           // Initialize tableif (cells == as) {// 与条件中cells == as形成双重检查Cell[] rs = new Cell[2];rs[h & 1] = new Cell(x); // 按照当前线程hash到数组中的位置并创建其对应的Cellcells = rs;init = true;}} finally {cellsBusy = 0;}if (init)break;}// CASE 3：cells正在进行初始化，则尝试直接在基数base上进行累加操作；兜底操作，当新线程进来，而数组正在初始化时，则用base进行累加else if (casBase(v = base, ((fn == null) ? v + x :fn.applyAsLong(v, x))))break;                          // Fall back on using base}}

5.3 sum()

    /*** Returns the current sum.  The returned value is <em>NOT</em> an* atomic snapshot; invocation in the absence of concurrent* updates returns an accurate result, but concurrent updates that* occur while the sum is being calculated might not be* incorporated.** @return the sum*/public long sum() {Cell[] as = cells; Cell a;long sum = base;if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += a.value;}}return sum;}

sum执行时，并没有限制对base和cells的更新。所以LongAdder不是强一致性的，它是最终一致性的

6. 小总结

6.1 AtomicLong

原理：CAS+自旋

场景：低并发下的全局计算，AtomicLong能保证并发情况下计数的准确性，内部使用CAS来解决并发安全问题

缺陷：高并发后性能急剧下降，N个线程CAS操作修改线程的值，每次只有一个成功，其它N-1失败，失败的不停的自旋直到成功，这样大量失败自旋的情况，很浪费CPU资源

6.2 LongAdder

原理：CAS+Base+Cell数组分散，空间换时间并分散了热点数据

场景：高并发下的全局计算

缺陷：sum求和时还有计算线程修改结果的话，最后的结果不够准确