看圖學源碼之 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");}}