LongAdder

LongAdder

在使用AtomicLong时，如果是在高并发场景下去同时竞争修改同一个原子变量，由于内部使用的是CAS，只会有一个线程修改成功，这就造成了大量的线程竞争失败后，通过无限循环来不断的进行CAS操作，白白的浪费了CPU资源，在JDK8中为了解决这种问题，提供了LongAdder来进行原子性递增递减

LongAdder将一个变量分解为多个变量，在内部维护了多个Cell变量，每个Cell变量中有一个初始值为0的long类型变量。当多线程争夺同一个Cell原子变量时如果失败，并不是在当前Cell变量上一直自旋CAS重试，而是会尝试在其他Cell变量上进行CAS尝试，增加了CAS成功的可能性

最终，获取LongAdder的当前值时，会把所有Cell变量的value值累加后再加上base值

// 真实值为base值与cells数组中所有Cell元素中value值的累加
// 在LongAdder中维护了一个Cell数组，分担对单个变量进行竞争的开销
transient volatile Cell[] cells;
// 基值变量
transient volatile long base;
// 用来实现自旋锁，状态值只有0和1，当创建Cell元素，扩容Cell数组和初始化Cell数组时，使用CAS操作该变量来保证同时只有一个线程可以进行其中一个Cell的操作
transient volatile int cellsBusy;

@sun.misc.Contended static final class Cell {
              // 每个Cell中有一个初始值为0的long类型变量value
        volatile long value;
        Cell(long x) { value = x; }
        final boolean cas(long cmp, long val) {
            return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
        }

        // Unsafe mechanics
        private static final sun.misc.Unsafe UNSAFE;
        private static final long valueOffset;
        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> ak = Cell.class;
                valueOffset = UNSAFE.objectFieldOffset
                    (ak.getDeclaredField("value"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    }


// 进行加一减一操作
public void add(long x) {
  Cell[] as; long b, v; int m; Cell a;
  // 如果cells为null的话在base的基础上进行累加
  if ((as = cells) != null || !casBase(b = base, b + x)) {
    boolean uncontended = true;
    // 选择其中一个Cell进行操作
    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);
  }
}


final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
  int h;
  // 初始化当前线程的变量h，用于计算当前线程应该被分到cells数组中的哪个元素
  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;
    // cells如果为null，说明还没有初始化
    // 不为null，说明已经初始化过了
    if ((as = cells) != null && (n = as.length) > 0) {
      // 数组cells已经扩容，但是有的Cell还没有填充，此时为null，需要新增一个Cell元素到cells数组中
      if ((a = as[(n - 1) & h]) == null) {
        if (cellsBusy == 0) {       // Try to attach new Cell
          Cell r = new Cell(x);   // Optimistically create
          // 添加新元素需要将cellsBusy置为1
          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
      // 是否有冲突(多个线程访问了cells数组中的同一个Cell)
      else if (!collide)
        collide = true;
      // 只有当前元素个数没有达到CPU个数，并且有冲突，会进行扩容
      // 进行扩容时casCellsBusy方法来将cellsBusy设置为1
      else if (cellsBusy == 0 && casCellsBusy()) {
        try {
          
          if (cells == as) {      // Expand table unless stale
            // 扩容为原来的2倍
            Cell[] rs = new Cell[n << 1];
            // 复制原数组中的Cell元素到新的cells数组中
            for (int i = 0; i < n; ++i)
              rs[i] = as[i];
            cells = rs;
          }
        } finally {
          cellsBusy = 0;
        }
        collide = false;
        continue;                   // Retry with expanded table
      }
      // 重新计算h，找到空闲的Cell
      h = advanceProbe(h);
    }
    // 初始化Cell数组，cellsBusy为0表示当前cells数组没有在被初始化或者扩容，也没有在新建Cell元素；为1表示当前cells数组在被初始化或者扩容，或者在新建Cell元素
    // casCellsBusy来切换cellsBusy的状态
    else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
      boolean init = false;
      try {                           // Initialize table
        if (cells == as) {
          // 初始化的个数为2个
          Cell[] rs = new Cell[2];
          // 使用h&1来计算当前线程应该访问哪个Cell
          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
  }
}