JDK 8 原子操作类 LongAdder--粉丝服务平台-粉丝头条-fensifuwu.com

JDK 8 原子操作类 LongAdder

科技 07-03 来源：儒雅程序员

AtomicLong 通过 CAS 提供了非阻塞的原子性操作，相比使用阻塞算法的同步器来说它的性能己经很好了，但是 JDK 开发组并不满足于此。使用 AtomicLong 时，在高并发下大量线程会同时去竞争更新同→个原子变量，但是由于同时只有一个线程的 CAS 操作会成功，这就造成了大量线程竞争失败后，会通过无限循环不断进行自旋尝试CAS 的操作，而这会白白浪费 CPU 资源。

因此 JDK 8 新增了一个原子性递增或者递减类 LongAdder 用来克服在高并发下使用 AtomicLong 的缺点。既然 AtomicLong 的性能瓶颈是由于过多线程同时去竞争一个变量的更新而产生的，那么如果把一个变量分解为多个变量，让同样多的线程去竞争多个资源

LongAdder 维护了一个延迟初始化的原子性更新数组(默认情况下 Cell 数组是 null〕和一个基值变量 base。由于 Cells 占用的内存是相对比较大的，所以一开始并不创建它，而是在需要时创建，也就是惰性加载。

当一开始判断 Cell 数组是 null 并且并发线程较少时，所有的累加操作都是对 base 变量进行的。保持Cel数组的大小为2的N次方，在初始化时Cell数组中的Cell元素个数为 2，数组里面的变量实体是 Cell 类型。 Cell 类型是 AtomicLong 的一个改进，用来减少缓存的争用，也就是解决伪共享问题。

对于大多数孤立的多个原子操作进行字节填充是浪费的，因为原子性操作都是无规律

地分散在内存中的 (也就是说多个原子性变量的内存地址是不连续的)，多个原子变量被放入同一个缓存行的可能性很小。但是原子性数组元素的内存地址是连续的，所以数组内的多个元素能经常共享缓存行，因此这里使用 @sun.misc.Contended注解对 Cell类进行字节填充，这防止了数组中多个元素共享一个缓存行，在性能上是一个提升。

LongAdder类继承自 Striped64类，在 Striped64 内部维护着三个变量。 LongAdder 的真实值其实是 base 的值与 Cell 数组里面所有 Cell元素中的 value值的累加， base是个基础值，默认为 0。 cellsBusy用来实现自旋锁，状态值只有 0和 1，当创建 Cell元素，扩容 Cell 数组或者初始化 Cel 数组时，使用 CAS 操作该变量来保证同时只有一个线程可以进行其中之一的操作。

java.util.concurrent.atomic#LongAdder:sum

返回当前的值，内部操作是累加所有 Cell 内部的 value值后再累加 base。例如下面的代码，由于计算总和时没有对 Cell 数组进行加锁，所以在累加过程中可能有其他线程对 Cell 中的值进行了修改，也有可能对数组进行了扩容，所以 sum 返回的值并不是非常精确的，其返回值并不是一个调用 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;
 }

java.util.concurrent.atomic#LongAdder:reset

base置为0，如果Cell数组有元素，则元素值被重置为 0。

public void reset() {
 Cell[] as = cells; Cell a;
 base = 0L;
 if (as != null) {
 for (int i = 0; i < as.length; ++i) {
 if ((a = as[i]) != null)
 a.value = 0L;
 }
 }
 }

java.util.concurrent.atomic#LongAdder:add

添加方法,

 public void add(long x) {
 Cell[] as; long b, v; int m; Cell a;
 if ((as = cells) != null || !casBase(b = base, b + x)) {
 boolean uncontended = true;
 if (as == null || (m = as.length - 1) < 0 ||
 (a = as[getProbe() & m]) == null ||
 !(uncontended = a.cas(v = a.value, v + x)))
 longAccumulate(x, null, uncontended);
 }
 }

java.util.concurrent.atomic#Striped64 : longAccumulate
final void longAccumulate(long x, LongBinaryOperator fn,
 boolean wasUncontended) {
 // 初始化当前线程的变量 threadLocalRandomProbe的值
 int h;
 if ((h = getProbe()) == 0) {
 ThreadLocalRandom.current(); // force initialization
 h = getProbe();
 wasUncontended = true;
 }
 boolean collide = false; // True if last slot nonempty
 for (;;) {
 Cell[] as; Cell a; int n; long v;
 
 if ((as = cells) != null && (n = as.length) > 0) {
 if ((a = as[(n - 1) & h]) == null) {
 if (cellsBusy == 0) { // Try to attach new Cell
 Cell r = new Cell(x); // Optimistically create
 if (cellsBusy == 0 && casCellsBusy()) {
 boolean created = false;
 try { // Recheck under lock
 Cell[] 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;
 }
 else if (!wasUncontended) // CAS already known to fail
 wasUncontended = true; // Continue after rehash
 
 //当前Cell存在，则执行CAS设置
 else if (a.cas(v = a.value, ((fn == null) ? v + x :
 fn.applyAsLong(v, x))))
 break;
 //当前Cell数纽元素个数大于CPU个数
 else if (n >= NCPU || cells != as)
 collide = false; // At max size or stale
 //是否有冲突
 else if (!collide)
 collide = true;
 //如采当前元素个数没有达到CPU个数并且有冲 突则扩容
 else if (cellsBusy == 0 && casCellsBusy()) {
 try {
 if (cells == as) { // Expand table unless stale
 Cell[] 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
 }
 //为 了能够找到一个空闲的Cell， 重新计算hash值，xorsh工ft算法生成随机数
 h = advanceProbe(h);
 }
 //初始化Cell数组
 else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
 boolean init = false;
 try { // Initialize table
 if (cells == as) {
 Cell[] rs = new Cell[2];
 rs[h & 1] = new Cell(x);
 cells = rs;
 init = true;
 }
 } finally {
 cellsBusy = 0;
 }
 if (init)
 break;
 }
 else if (casBase(v = base, ((fn == null) ? v + x :
 fn.applyAsLong(v, x))))
 break; // Fall back on using base
 }
 }

Cell 懒加载, 开始只是用 base 值. cell 并未初始化 . 当添加发生竞争的时候, 创建.

Cell 初始值为 2, 扩容为翻倍.

Cell 元素最大值为 CPU 核数.

Cell 扩容条件: casCellsBusy 为 false 没有库容并且有碰撞并且 cell 的数量没有超过 cpu 的核数.

LongAdder原子性操作类，该类通过内部 cells数组分担了高并发下多线程同时对一个原子变量进行更新时的竞争量，让多个线程可以同时对 cells 数组里面的元素进行并行的更新操作。另外，数组元素 Cell使用@sun.misc.Contended注解进行修饰，这避免了 cells数组内多个原子变量被放入同一个缓存行，也就是避免了伪共享，这对性能也是一个提升。