JUC解析-LongAdder

在JDK1.8的atomic中增长了一个原子操做类LongAddr，与AtomicLong相比，在效率方面有了不小的提高。

实现原理

AtomicLong的原子操做是经过CAS进行更新的，当有冲突的时候，会经过自旋的方式等待原子操做，具体:

public final long updateAndGet(LongUnaryOperator updateFunction) {
    long prev, next;
    do {
      prev = get();
      next = updateFunction.applyAsLong(prev);
    } while (!compareAndSet(prev, next));
    return next;
  }
复制代码

从上面的源码能够看出，在原子更新冲突比较少的时候，更新的效率仍是很快的，直接经过CAS更新，可是因为是对同一个元素进行更新，当更新操做比较频繁的时候，发生的冲突的几率就会增大，效率就会下降，如今问题就变为：

怎样减小发生冲突的几率？

在多线程环境下，线程的行为是不可预测的，所以能够在更新的元素上作文章，在AtomicLong中是经过对一个成员变量private volatile long value进行原子的更新操做的，所以冲突也是围绕该元素的，来看下Doug Lea大神是怎么作的:

1. 初始化一个volatile long base和一个int [] cell的数组，数组中的元素都初始化为0。
2. 当线程进行原子操做的时候，优先对base变量CAS操做，若是没有发生冲突，则更新成功，若是发生冲突，则放弃自旋的方案，进行步骤3。
3. 对线程经过hashcode映射到cell数组中的一个元素中，而后将原子的操做更新到该数组元素中。
4. 最后经过叠加base和cell中的每一个元素，就能够获取到原子操做的数值。

在LongAddr中，将原子更新的操做分散到了基础变量base和一个数组中，这样就下降了发生冲突的几率，提高了更新的效率。

具体实现

LongAddr继承自Striped64，具体的操做是经过Striped64实现，Striped64的成员变量以下：

public class LongAdder extends Striped64 implements Serializable {
  //对应上面的数组
  transient volatile Cell[] cells;
  //基础更新变量
  transient volatile long base;
  //同步对cells进行操做的
  transient volatile int cellsBusy;
复制代码

其中Cell的定义以下：

@sun.misc.Contended static final class Cell {
    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);
      }
    }
  }
复制代码

@sun.misc.Contended注解是为了解决伪共享的问题，具体不了解的能够看这里浅谈伪共享。

Cell中只是包含了一个cas的原子操做行为，主要是数组中的元素进行原子更新操做。

更新操做

public void add(long x) {
    Cell[] as;
    long b, v;
    int m;
    Cell a;
    //若是cell不为空，或者cell为空可是caseBase失败
    if ((as = cells) != null || !casBase(b = base, b + x)) {
      boolean uncontended = true;
      //1. cells 为空
      //2. cells不为空，可是cells中对应本线程位置为null
      //3. cells中本线程对应的位置不为空，可是对本位置操做case失败
      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);
      }
  }
复制代码

1. 若是cells数组为空，则直接对基础变量base执行cas操做casBase，若是casBase执行失败，则说明有冲突，执行步骤3。
2. 若是cells数组不为空，则此时为了下降冲突的几率，优先对cells数组进行更新操做，直接执行步骤3。
3. 执行子逻辑判断： 3.1. 若是cells为空，直接执行步骤4。 3.2. 若是cells不为空，可是本线程对应的cells数组中的元素为null,则进行步骤4。 3.3. 若是cells不为空，而且本线程对应的数组中的元素不为null，则对该元素进行cas操做，若是执行成功返回，执行失败则进行步骤4。
4. 执行更新子逻辑，具体参照源码注解

final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) {
  int h;
  // 初始化Thread的种子和随机数
  if ((h = getProbe()) == 0) {
      ThreadLocalRandom.current(); // force initialization
      h = getProbe();
      wasUncontended = true;
  }
  boolean collide = false;    // True if last slot nonempty
  //步骤6
  for (;;) {
    Cell[] as; Cell a; int n; long v;
    //若是cells数据不为空，即知足条件3.2和3.3时
    if ((as = cells) != null && (n = as.length) > 0) {
      //若是被hash到的cells数组位置为null(对应条件3.2)
      if ((a = as[(n - 1) & h]) == null) {
        //若是此时cells数组中同步锁为空
        if (cellsBusy == 0) {
          //初始化Cell元素，初试化为x
          Cell r = new Cell(x);
          //设置cellsBusy锁，避免其余线程同时修改cells数组
          if (cellsBusy == 0 && casCellsBusy()) {
            boolean created = false;
            try {
              //初始化cells数组中本线程对应的元素
              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锁
              cellsBusy = 0;
            }
            //若是建立成功则退出，不然continue
            if (created)
              break;
            /*若是created为false，说明上面指定的cells数组的位置 cells[m%cells.length]已经有其它线程设置了cell了， 继续从新开始循环。*/
            continue; 
          }
        }
        //若是cellsBusy=1，说明有线程正在更改cells数组，
        将collide设置为false
        collide = false;
      }
      /*若是被hash到的cells数组位置不为null(对应3.3), 说明已经发生竞争，将wasUncontended设置为true， 最后从新计算一个新的probe，而后从新执行循环。 */
      else if (!wasUncontended)       
        wasUncontended = true;        
      /*若是当前线程第一次参与cells争用的cas失败， 这里会尝试将x值加到cells[m%cells.length]的value ， 若是成功直接退出*/
      else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                  fn.applyAsLong(v, x))))
        break;
      //若是cells数组的长度已经到了最大值（大于等于cup数量），
      或者是当前cells已经作了扩容，则将collide设置为false，
      后面从新计算prob的值.*/
      else if (n >= NCPU || cells != as)
        collide = false; 
      /*若是发生了冲突collide=false，则设置其为true； 会在最后从新计算hash值后，进入下一次for循环*/
      else if (!collide)
        collide = true;
      /*扩容cells数组，新参与cell争用的线程两次均失败， 且符合库容条件，会执行该分支*/
      else if (cellsBusy == 0 && casCellsBusy()) {
        try {
            if (cells == as) {      
              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;                   
      }
      //从新计算线程的hash值
      h = advanceProbe(h);
    } 
    //若是知足3.1 和 3.2 则初始化cells数组
    else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
      boolean init = false;
      try {                           
        if (cells == as) {
          Cell[] rs = new Cell[2];
          rs[h & 1] = new Cell(x);
          cells = rs;
          init = true;
        }
      } finally {
        cellsBusy = 0;
      }
      if (init)
        break;
    }
    //若是以上操做都失败了，则尝试将值累加到base上
    else if (casBase(v = base, ((fn == null) ? v + x :
                            fn.applyAsLong(v, x))))
      break;                  
  }
}

复制代码

上面代码写的比较简单，可是逻辑比较绕，主要就是同步的在base和cells数据更新之间操做，下降发生冲突的几率，提升效率。

获取结果

因为在多线程环境下，数据的更新操做是分散到base变量和cells数组中的每一个元素中的，所以每次计算结果都要把所有的数值叠加起来

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;
}
复制代码

线程hash数值的计算

基本的实现逻辑：

1. 在Thread线程类中保存有两个变量：

@sun.misc.Contended("tlr")
int threadLocalRandomProbe;  //此线程的随机数

@sun.misc.Contended("tlr")
int threadLocalRandomSecondarySeed; //此线程的随机种子
复制代码

2. ThreadLocalRandom类会更新每一个线程类种子，而且根据种子计算出该线程的随机数，这样线程之间就不会存在随机数的同步成本，提升效率。

//判断该线程的随机数是否已经初始化，若是没有则执行localInit初始化
  public static ThreadLocalRandom current() {
    if (UNSAFE.getInt(Thread.currentThread(), PROBE) == 0)
      localInit();
    return instance;
  }
  //更新线程的种子数和随机数
  static final void localInit() {
    int p = probeGenerator.addAndGet(PROBE_INCREMENT);
    int probe = (p == 0) ? 1 : p; // skip 0
    long seed = mix64(seeder.getAndAdd(SEEDER_INCREMENT));
    Thread t = Thread.currentThread();
    UNSAFE.putLong(t, SEED, seed);
    UNSAFE.putInt(t, PROBE, probe);
  }
复制代码

3. 若是发生冲突，能够对Thread进行rehash，具体以下：

static final int advanceProbe(int probe) {
    probe ^= probe << 13;   // xorshift
    probe ^= probe >>> 17;
    probe ^= probe << 5;
    UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
    return probe;
  }
复制代码

完。