看图学源码之 Atomic 类源码浅析二（cas + 分治思想的原子累加器）

原子累加器

相较于上一节看图学源码之 Atomic 类源码浅析一（cas + 自旋操作的 AtomicXXX原子类）说的的原子类，原子累加器的效率会更高

在这里插入图片描述

XXXXAdder 和 XXXAccumulator 区别就是 Adder只有add 方法，Accumulator是可以进行自定义运算方法的

始于 Striped64

abstract class Striped64 extends Number {// cpu 运行核数, 控制数组的大小static final int NCPU = Runtime.getRuntime().availableProcessors();// 当非空时，大小是 2 的幂。 transient volatile Cell[] cells;// 表初始化竞争期间的后备值  通过  CAS 更新  就是 valuetransient volatile long base;// 锁的 标志位 调整单元大小和/或创建单元时使用自旋锁（通过 CAS 锁定）transient volatile int cellsBusy;//Base的cas 操作final boolean casBase(long cmp, long val) {return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);}// CellsBusy的cas操作final boolean casCellsBusy() {return UNSAFE.compareAndSwapInt(this, CELLSBUSY, 0, 1);}// 主要是给不同的线程找到数组中不同的下标// 当前线程的探测值static final int getProbe() {return UNSAFE.getInt(Thread.currentThread(), PROBE);}// 给定线程的给定探测值static final int advanceProbe(int probe) {probe ^= probe << 13;   // xorshiftprobe ^= probe >>> 17;probe ^= probe << 5;UNSAFE.putInt(Thread.currentThread(), PROBE, probe);return probe;}......// Unsafe mechanics// Unsafe 的获取 和 偏移量的获取private static final sun.misc.Unsafe UNSAFE;private static final long BASE;private static final long CELLSBUSY;private static final long PROBE;static {try {UNSAFE = sun.misc.Unsafe.getUnsafe();Class<?> sk = Striped64.class;BASE = UNSAFE.objectFieldOffset(sk.getDeclaredField("base"));CELLSBUSY = UNSAFE.objectFieldOffset(sk.getDeclaredField("cellsBusy"));Class<?> tk = Thread.class;PROBE = UNSAFE.objectFieldOffset(tk.getDeclaredField("threadLocalRandomProbe"));} catch (Exception e) {throw new Error(e);}}}

内部类

@sun.misc.Contended——解决伪共享，进行字节填充

@sun.misc.Contended static final class Cell {// 操作的数volatile long value;// 构造器Cell(long x) { value = x; }// 进行 cas final boolean cas(long cmp, long val) {return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);}
}

LongAdder

可能会出现一边正在进行累加操作，一边又在执行求和操作，所以就导致了不是强一致性，而是最终一致性

在这里插入图片描述

public class LongAdder extends Striped64 implements Serializable {private static final long serialVersionUID = 7249069246863182397L;public LongAdder() {}public void add(long x) {Cell[] as; long b, v; int m; Cell a;// 数组是不是 null（判断有没有发生竞争，因为只有竞争发生才会初始化数组）// 没有初始化（就是没有竞争） 直接对 base 的值+x 失败if ((as = cells) != null || !casBase(b = base, b + x)) {//有竞争的时候boolean uncontended = true;// 数组还是没有初始化 || 数组初始化，但是数组的长度  < 0 || 数组中的该位置的值是 null (表示这个下标没有初始化)|| cas的方式把当前位置的值 + x ，cas 失败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);}}public void increment() {add(1L);}public void decrement() {add(-1L);}// 返回当前总和。返回的值不是原子快照；// 在没有并发更新的情况下调用会返回准确的结果，但计算总和时发生的并发更新可能不会被合并// 所以不是 强一致性的public long sum() {Cell[] as = cells; Cell a;long sum = base;//数组不是 nullif (as != null) {//遍历数组，for (int i = 0; i < as.length; ++i) {//数组中的槽位不是 null，对槽位的数据进行运算，赋值加到base中if ((a = as[i]) != null)sum += a.value;}}//返回总的值return sum;}// 将保持总和为零的变量重置。此方法可能是创建新加法器的有用替代方法，但仅在没有并发更新时才有效。// 由于此方法本质上是活泼的，因此仅应在  已知没有线程同时更新时才使用它。public void reset() {Cell[] as = cells; Cell a;base = 0L;if (as != null) {// 数组存在，遍历数组，将数组中所有的值设置为 0 for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)a.value = 0L;}}}// 相当于sum后跟reset // 该方法可以应用于例如 多线程计算之间的静止点期间。// 如果此方法同时有更新，则不能保证返回值是重置之前发生的最终值。public long sumThenReset() {Cell[] as = cells; Cell a;long sum = base;base = 0L;if (as != null) {// 数组存在，遍历数组，先求和  后把数组中所有的值设置为 0 for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null) {sum += a.value;a.value = 0L;}}}return sum;}// 返回sum的字符串表示形式public String toString() {return Long.toString(sum());}// 返回sumpublic long longValue() {return sum();}//缩小基元转换后以 int 形式返回sum public int intValue() {return (int)sum();}// 加宽基元转换后以float形式返回sum public float floatValue() {return (float)sum();}//加宽基元转换后以 double 形式返回sum public double doubleValue() {return (double)sum();}private static class SerializationProxy implements Serializable {private static final long serialVersionUID = 7249069246863182397L;private final long value;// sum() 返回的当前值。SerializationProxy(LongAdder a) {value = a.sum();}// 返回一个LongAdder对象，其初始状态由该代理保存。private Object readResolve() {LongAdder a = new LongAdder();a.base = value;return a;}}private Object writeReplace() {return new SerializationProxy(this);}private void readObject(java.io.ObjectInputStream s)throws java.io.InvalidObjectException {throw new java.io.InvalidObjectException("Proxy required");}}

striped64中的 longAccumulate

在这里插入图片描述

final void longAccumulate(long x, LongBinaryOperator fn,boolean wasUncontended) {int h;// 拿 hash 值，拿不到强制获取if ((h = getProbe()) == 0) {ThreadLocalRandom.current(); // force initializationh = getProbe();// 将 wasUncontended 的值设为 true，表示当前线程是未争用的。wasUncontended = true;}boolean collide = false;                // True if last slot nonemptyfor (;;) {Cell[] as; Cell a; int n; long v;// 分支 1//数组已经初始化，出现了竞争if ((as = cells) != null && (n = as.length) > 0) {// 分支1.1// 当前位置的值是nullif ((a = as[(n - 1) & h]) == null) {// 锁的标志位 == 0 ，没有加锁 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;//  加锁之后再次检查指定位置是否为空 //  数组初始化过了 && 当前位置的值不是nullif ((rs = cells) != null &&(m = rs.length) > 0 &&rs[j = (m - 1) & h] == null) {// 给数组的指定位置设置为  之前设置过的cell对象rs[j] = r;// 创建成功created = true;}} finally {// 解锁cellsBusy = 0;}if (created)break;continue;           // Slot is now non-empty}}// 有人加锁了，发生了冲突 避免在当前位置发生碰撞的情况下继续进行操作，将 collide 标志位设置为 false。collide = false;}// 分支1.2// 没有发生竞争else if (!wasUncontended)       // CAS already known to fail// 此时是发生了竞争wasUncontended = true;      // Continue after rehash// 分支1.3// cas 的方式更新 此位置的值, cas 失败表示有线程正在此位置执行操作else if (a.cas(v = a.value, ((fn == null) ? v + x :fn.applyAsLong(v, x))))break;// 分支1.4// n > cpu 的个数  当前分段数组的长度是否已经达到或超过了处理器的数量。//如果是,说明分段数组已经达到了最大的容量或者已经很大了，不再需要继续进行扩容操作。// 或者 cells  发生了变化，当前线程获取到的分段数组引用是否与共享变量中的引用相等。// 如果不相等，说明在当前线程获取到分段数组的过程中，有其他线程进行了修改，即分段数组已经发生了变化。else if (n >= NCPU || cells != as)collide = false;            // At max size or stale// 分支1.5// 此时是发生了碰撞的 collide 被设置为 true else if (!collide)collide = true;// 分支1.6 // 扩容// 没有被锁 && cas  的方式 成功加锁 else if (cellsBusy == 0 && casCellsBusy()) {try {// 数组没有变化if (cells == as) {      // Expand table unless stale// 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;                   // Retry with expanded table}// 更新 hash 值h = advanceProbe(h);}
// 分支 2// 此处数组没有进行初始化，此时进行初始化// 锁的标志为 == 0  &&  数组没有改变（多线程情况下该线程没有被其他线程初始化）  && cas 成功的把锁的标志位 设置为 1（加锁流程）// 当前的 cells 数组没有被其他线程占用，并且成功获取了 cellsBusy 锁else if (cellsBusy == 0 && cells == as && casCellsBusy()) {boolean init = false;try {                           // Initialize table// 加完锁之后再次判断一次  cells 数组没有发生过变化if (cells == as) { // 数组 长度默认为2Cell[] rs = new Cell[2];// 给rs 赋值为 要加入的 xrs[h & 1] = new Cell(x);// 将 cells 数组变更为 rscells = rs;// 初始化成功init = true;}} finally {// 解锁cellsBusy = 0;}if (init)    //初始化成功break;    // 退出循环}
// 分支 3  cas 的方式 操作 base  , fn 函数式接口的方法  == null 默认加法，否则就是定义的方法else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x))))break;                          // Fall back on using base  退出循环}
}

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;}// 更新值public void accumulate(long x) {Cell[] as; long b, v, r; int m; Cell a;// 有竞争  || （ cas 运算 base 的值成功  && 对 base进行cas更新失败 ）if ((as = cells) != null ||//  function.applyAsLong  函数式接口 (r = function.applyAsLong(b = base, x)) != b && !casBase(b, r)) {boolean uncontended = true;// 出现了竞争// 数组还是没有初始化 || 数组初始化，但数组的长度 < 0 || 数组中的该位置的值是 null (表示这个下标没有初始化)|| （cas的方式运算当前位置的值 失败 && cas 更新当前位置的值也失败）if (as == null || (m = as.length - 1) < 0 ||(a = as[getProbe() & m]) == null ||!(uncontended =(r = function.applyAsLong(v = a.value, x)) == v ||a.cas(v, r)))longAccumulate(x, function, uncontended);}}// 返回当前值。返回的值不是原子快照；// 在没有并发更新的情况下调用会返回准确的结果，但计算值时发生的并发更新可能不会被合并。public long get() {Cell[] as = cells; Cell a;long result = base;// 数组存在if (as != null) {// 遍历数组for (int i = 0; i < as.length; ++i) {//数组中的槽位不是 nullif ((a = as[i]) != null)//对槽位的数据进行运算，赋值加到base中result = function.applyAsLong(result, a.value);}}// 并返回总的值return result;}// 重置变量以维护对标识值的更新。// 此方法可能是创建新更新程序的有用替代方法，但仅在没有并发更新时才有效。// 由于此方法本质上是活泼的，因此仅应在已知没有线程同时更新时才使用它。 public void reset() {Cell[] as = cells; Cell a;base = identity;// 数组存在if (as != null) {// 遍历数组for (int i = 0; i < as.length; ++i) {//数组中的槽位不是 nullif ((a = as[i]) != null)//将槽位的值设置为  identitya.value = identity;}}}// 效果相当于get后面跟着reset 。// 该方法可以应用于例如多线程计算之间的静止点期间。// 如果此方法同时有更新，则不能保证返回值是重置之前发生的最终值。public long getThenReset() {Cell[] as = cells; Cell a;long result = base;base = identity;// 数组存在if (as != null) {// 遍历数组for (int i = 0; i < as.length; ++i) {//数组中的槽位不是 nullif ((a = as[i]) != null) {// 将槽位的值设置为  identity// 对槽位的数据进行运算，赋值加到base中long v = a.value;a.value = identity;result = function.applyAsLong(result, v);}}}return result;}public String toString() {return Long.toString(get());}public long longValue() {return get();}public int intValue() {return (int)get();}public float floatValue() {return (float)get();}public double doubleValue() {return (double)get();}/*** 序列化代理，用于避免以序列化形式引用非公共 Striped64 超类*/private static class SerializationProxy implements Serializable {private static final long serialVersionUID = 7249069246863182397L;private final long value;private final LongBinaryOperator function;private final long identity;SerializationProxy(LongAccumulator a) {function = a.function;identity = a.identity;value = a.get();}private Object readResolve() {LongAccumulator a = new LongAccumulator(function, identity);a.base = value;return a;}}private Object writeReplace() {return new SerializationProxy(this);}private void readObject(java.io.ObjectInputStream s)throws java.io.InvalidObjectException {throw new java.io.InvalidObjectException("Proxy required");}}

在Double 中会有 doubleToRawLongBits的操作，主要是检查数组越界的

DoubleAdder

在这里插入图片描述

public class DoubleAdder extends Striped64 implements Serializable {private static final long serialVersionUID = 7249069246863182397L;/*
请注意，我们必须使用“long”作为底层表示，因为 double 没有compareAndSet，因为任何 CAS 实现中使用的按位等于与双精度等于不同
然而，我们仅使用 CAS 来检测和缓解争用，无论如何，按位等于效果最好。
原则上，这里使用的 longdouble 转换在大多数平台上基本上应该是免费的，因为它们只是重新解释位。*/public DoubleAdder() {}public void add(double x) {Cell[] as; long b, v; int m; Cell a;// 数组存在 || 对 base 进行 cas运算操作失败  
if ((as = cells) != null ||!casBase(b = base,Double.doubleToRawLongBits(Double.longBitsToDouble(b) + x))) {// boolean uncontended = true;// 数组为 空 || 数组的长度 < 0  || 当前位置的值为 null  ||  对该位置的值进行cas 运算失败if (as == null || (m = as.length - 1) < 0 ||(a = as[getProbe() & m]) == null ||!(uncontended = a.cas(v = a.value,Double.doubleToRawLongBits(Double.longBitsToDouble(v) + x))))doubleAccumulate(x, null, uncontended);}}/**返回当前总和。返回的值不是原子快照；在没有并发更新的情况下调用会返回准确的结果，但计算总和时发生的并发更新可能不会被合并。由于浮点算术不是严格关联的，因此返回的结果不需要与在单个变量的一系列连续更新中获得的值相同。*/public double sum() {Cell[] as = cells; Cell a;double sum = Double.longBitsToDouble(base);if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += Double.longBitsToDouble(a.value);}}return sum;}/**将保持总和为零的变量重置。此方法可能是创建新加法器的有用替代方法，但仅在没有并发更新时才有效。由于此方法本质上是活泼的，因此仅应在已知没有线程同时更新时才使用它。*/public void reset() {Cell[] as = cells; Cell a;base = 0L; // relies on fact that double 0 must have same rep as longif (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)a.value = 0L;}}}/**相当于sum后跟reset 。该方法可以应用于例如多线程计算之间的静止点期间。如果此方法同时有更新，则不能保证返回值是重置之前发生的最终值。*/public double sumThenReset() {Cell[] as = cells; Cell a;double sum = Double.longBitsToDouble(base);base = 0L;if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null) {long v = a.value;a.value = 0L;sum += Double.longBitsToDouble(v);}}}return sum;}public String toString() {return Double.toString(sum());}public double doubleValue() {return sum();}public long longValue() {return (long)sum();}public int intValue() {return (int)sum();}public float floatValue() {return (float)sum();}private static class SerializationProxy implements Serializable {private static final long serialVersionUID = 7249069246863182397L;private final double value;SerializationProxy(DoubleAdder a) {value = a.sum();}private Object readResolve() {DoubleAdder a = new DoubleAdder();a.base = Double.doubleToRawLongBits(value);return a;}}private Object writeReplace() {return new SerializationProxy(this);}private void readObject(java.io.ObjectInputStream s)throws java.io.InvalidObjectException {throw new java.io.InvalidObjectException("Proxy required");}}

striped64中的 doubleAccumulate

和上面的 striped64中的 longAccumulate 几乎一模一样，只有doubleToRawLongBits部分的细微差别

final void doubleAccumulate(double x, DoubleBinaryOperator fn,boolean wasUncontended) {int h;// 拿 hash 值，拿不到强制获取if ((h = getProbe()) == 0) {ThreadLocalRandom.current(); // force initializationh = getProbe();wasUncontended = true;}boolean collide = false;                // True if last slot nonemptyfor (;;) {Cell[] as; Cell a; int n; long v;
// 分支 1//数组已经初始化，出现了竞争if ((as = cells) != null && (n = as.length) > 0) {// 分支1.1// 当前位置的值是nullif ((a = as[(n - 1) & h]) == null) {// 锁的标志位 == 0 ，没有加锁 if (cellsBusy == 0) {       // Try to attach new CellCell r = new Cell(Double.doubleToRawLongBits(x));// 加锁if (cellsBusy == 0 && casCellsBusy()) {boolean created = false;try {               // Recheck under lockCell[] rs; int m, j;// 数组初始化过了 && 当前位置的值不是nullif ((rs = cells) != null &&(m = rs.length) > 0 &&rs[j = (m - 1) & h] == null) {// 给数组的位置设置为  之前设置过的cell对象rs[j] = r;// 创建成功created = true;}} finally {// 解锁cellsBusy = 0;}if (created)break;continue;           // Slot is now non-empty}}// 有人加锁了，发生了冲突collide = false;}// 分支1.2// 没有发生竞争else if (!wasUncontended)       // CAS already known to fail// 此时是发生了竞争wasUncontended = true;      // Continue after rehash// 分支1.3// cas 的方式更新 此位置的值, cas 失败表示有线程正在此位置执行操作else if (a.cas(v = a.value,((fn == null) ?Double.doubleToRawLongBits(Double.longBitsToDouble(v) + x) :Double.doubleToRawLongBits(fn.applyAsDouble(Double.longBitsToDouble(v), x)))))break;// 分支1.4// n > cpu 的个数 或者 cells  发生了变化，表示 之前没有发生碰撞，不能扩容else if (n >= NCPU || cells != as)collide = false;            // At max size or stale// 分支1.5// 此时是发生了碰撞的 collide 被设置为 true else if (!collide)collide = true;// 分支1.6 // 扩容// 没有被锁 && 成功加锁 else if (cellsBusy == 0 && casCellsBusy()) {try {// 数组没有变化if (cells == as) {      // Expand table unless stale// 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;                   // Retry with expanded table}// 获取hash 值h = advanceProbe(h);}
// 分支 2// 此处数组没有进行初始化，此时进行初始化// 锁的标志为 == 0  &&  数组没有改变（多线程情况下该线程没有被其他线程初始化）  && cas 成功的把锁的标志位 设置为 1（枷锁流程）else if (cellsBusy == 0 && cells == as && casCellsBusy()) {boolean init = false;try {   // Initialize table// 加完锁之后再次判断一次  cells 数组没有发生过变化if (cells == as) {// 数组 长度默认为2Cell[] rs = new Cell[2];// 给rs 赋值为 要加入的 xrs[h & 1] = new Cell(Double.doubleToRawLongBits(x));// 将 cells 数组变更为 rscells = rs;// 初始化成功init = true;}} finally {// 解锁cellsBusy = 0;}if (init)// 初始化成功break;    // 退出循环}
// 分支 3  cas 的方式 操作 base  , fn 函数式接口的方法  == null 默认加法，否则就是定义的方法else if (casBase(v = base,((fn == null) ?Double.doubleToRawLongBits(Double.longBitsToDouble(v) + x) :Double.doubleToRawLongBits(fn.applyAsDouble(Double.longBitsToDouble(v), x)))))break;                          // Fall back on using base// 退出循环}
}

DoubleAccumulator

在这里插入图片描述

public class DoubleAccumulator extends Striped64 implements Serializable {private static final long serialVersionUID = 7249069246863182397L;private final DoubleBinaryOperator function;private final long identity; // use long representationpublic DoubleAccumulator(DoubleBinaryOperator accumulatorFunction,double identity) {this.function = accumulatorFunction;base = this.identity = Double.doubleToRawLongBits(identity);}public void accumulate(double x) {Cell[] as; long b, v, r; int m; Cell a;// 数组存在 || （对 base 进行 cas的 运算操作成功  && 对base 进行cas 更新操作失败  ）if ((as = cells) != null ||(r = Double.doubleToRawLongBits(function.applyAsDouble (Double.longBitsToDouble(b = base), x))) != b  && !casBase(b, r)) {boolean uncontended = true;
// 数组为 空 || 数组被初始化但是 数组的长度 < 0  || 当前位置的值为 null  ||  (对该位置的值进行cas 运算失败  || 对该值进行cas 更新失败)if (as == null || (m = as.length - 1) < 0 ||(a = as[getProbe() & m]) == null ||!(uncontended =(r = Double.doubleToRawLongBits(function.applyAsDouble(Double.longBitsToDouble(v = a.value), x))) == v ||a.cas(v, r)))doubleAccumulate(x, function, uncontended);}}public double get() {Cell[] as = cells; Cell a;double result = Double.longBitsToDouble(base);if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)result = function.applyAsDouble(result, Double.longBitsToDouble(a.value));}}return result;}public void reset() {Cell[] as = cells; Cell a;base = identity;if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)a.value = identity;}}}public double getThenReset() {Cell[] as = cells; Cell a;double result = Double.longBitsToDouble(base);base = identity;if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null) {double v = Double.longBitsToDouble(a.value);a.value = identity;result = function.applyAsDouble(result, v);}}}return result;}public String toString() {return Double.toString(get());}public double doubleValue() {return get();}public long longValue() {return (long)get();}public int intValue() {return (int)get();}public float floatValue() {return (float)get();}private static class SerializationProxy implements Serializable {private static final long serialVersionUID = 7249069246863182397L;private final double value;private final DoubleBinaryOperator function;private final long identity;SerializationProxy(DoubleAccumulator a) {function = a.function;identity = a.identity;value = a.get();}private Object readResolve() {double d = Double.longBitsToDouble(identity);DoubleAccumulator a = new DoubleAccumulator(function, d);a.base = Double.doubleToRawLongBits(value);return a;}}private Object writeReplace() {return new SerializationProxy(this);}private void readObject(java.io.ObjectInputStream s)throws java.io.InvalidObjectException {throw new java.io.InvalidObjectException("Proxy required");}}