溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇內容介紹了“并發計數類LongAdder怎么用”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！
AtomicLong是通過CAS(即Compare And Swap)原理來完成原子遞增遞減操作，在并發情況下不會出現線程不安全結果。AtomicLong中的value是使用volatile修飾，并發下各個線程對value最新值均可見。我們以incrementAndGet()方法來深入。
public final long incrementAndGet() {
return unsafe.getAndAddLong(this, valueOffset, 1L) + 1L;
}
這里是調用了unsafe的方法
public final long getAndAddLong(Object var1, long var2, long var4) {
long var6;
do {
var6 = this.getLongVolatile(var1, var2);
} while(!this.compareAndSwapLong(var1, var2, var6, var6 + var4));
return var6;
}
方法中this.compareAndSwapLong()有4個參數，var1是需要修改的類對象，var2是需要修改的字段的內存地址，var6是修改前字段的值，var6+var4是修改后字段的值。compareAndSwapLong只有該字段實際值和var6值相當的時候，才可以成功設置其為var6+var4。
再繼續往深一層去看
public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);
這里Unsafe.compareAndSwapLong是native方法，底層通過JNI(Java Native Interface)來完成調用，實際就是借助C來調用CPU指令來完成。
實現中使用了do-while循環，如果CAS失敗，則會繼續重試，直到成功為止。并發特別高的時候，雖然這里可能會有很多次循環，但是還是可以保證線程安全的。不過如果自旋CAS操作長時間不成功，競爭較大，會帶CPU帶來極大的開銷，占用更多的執行資源，可能會影響其他主業務的計算等。
LongAdder怎么優化AtomicLong
Doug Lea在jdk1.5的時候就針對HashMap進行了線程安全和并發性能的優化，推出了分段鎖實現的ConcurrentHashMap。一般Java面試，基本上離不開ConcurrentHashMap這個網紅問題。另外在ForkJoinPool中，Doug Lea在其工作竊取算法上對WorkQueue使用了細粒度鎖來較少并發的競爭,更多細節可參考我的原創文章ForkJoin使用和原理剖析。如果已經對ConcurrentHashMap有了較為深刻的理解，那么現在來看LongAdder的實現就會相對簡單了。
來看LongAdder的increase()方法實現，
public void add(long x) {
Cell[] as; long b, v; int m; Cell a;
//第一個if進行了兩個判斷，(1)如果cells不為空，則直接進入第二個if語句中。(2)同樣會先使用cas指令來嘗試add，如果成功則直接返回。如果失敗則說明存在競爭，需要重新add
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);
}
}
這里用到了Cell類對象，Cell對象是LongAdder高并發實現的關鍵。在casBase沖突嚴重的時候，就會去創建Cell對象并添加到cells中，下面會詳細分析。
@sun.misc.Contended static final class Cell {
volatile long value;
Cell(long x) { value = x; }
//提供CAS方法修改當前Cell對象上的value
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);
}
}
}
而這一句a = as[getProbe() & m]其實就是通過getProbe()拿到當前Thread的threadLocalRandomProbe的probe Hash值。這個值其實是一個隨機值，這個隨機值由當前線程ThreadLocalRandom.current()產生。不用Rondom的原因是因為這里已經是高并發了，多線程情況下Rondom會極大可能得到同一個隨機值。因此這里使用threadLocalRandomProbe在高并發時會更加隨機，減少沖突。更多ThreadLocalRandom信息想要深入了解可關注這篇文章并發包中ThreadLocalRandom類原理淺嘗。拿到as數組中當前線程的Cell對象，然后再進行CAS的更新操作,我們在源碼上進行分析。longAccumulate()是在父類Striped64.java中。
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended) {
int h;
if ((h = getProbe()) == 0) {
//如果當前線程的隨機數為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數組不為空
if ((as = cells) != null && (n = as.length) > 0) {
//如果線程隨機數對應的cells對應數組下標的Cell元素不為空，
if ((a = as[(n - 1) & h]) == null) {
//當使用到LongAdder的Cell數組相關的操作時，需要先獲取全局的cellsBusy的鎖，才可以進行相關操作。如果當前有其他線程的使用，則放棄這一步，繼續for循環重試。
if (cellsBusy == 0) { // Try to attach new Cell
//Cell的初始值是x，創建完畢則說明已經加上
Cell r = new Cell(x); // Optimistically create
//casCellsBusy獲取鎖，cellsBusy通過CAS方式獲取鎖，當成功設置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 {
//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
//如果a不為空，則對a進行cas增x操作，成功則返回
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;
//cells的長度n已經大于CPU數量，則繼續擴容沒有意義，因此直接標記為不沖突
else if (n >= NCPU || cells != as)
collide = false; // At max size or stale
else if (!collide)
collide = true;
//到這一步則說明a不為空但是a上進行CAS操作也有多個線程在競爭，因此需要擴容cells數組，其長度為原長度的2倍
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上競爭
h = advanceProbe(h);
}
//如果cells為空，則需要先創建Cell數組。初始長度為2.(個人理解這個if放在前面會比較好一點，哈哈)
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;
}
//如果在a上競爭失敗，且擴容競爭也失敗了，則在casBase上嘗試增加數量
else if (casBase(v = base, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}
最后是求LongAdder的總數，這一步就非常簡單了，把base的值和所有cells上的value值加起來就是總數了。
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;
}
思考
源碼中Cell數組的會控制在不超過NCPU的兩倍，原因是LongAdder其實在底層是依賴CPU的CAS指令來操作，如果多出太多，即使在代碼層面沒有競爭，在底層CPU的競爭會更多，所以這里會有一個數量的限制。所以在LongAdder的設計中，由于使用到CAS指令操作，瓶頸在于CPU上。
YY一下，那么有沒有方式可以突破這個瓶頸呢？我個人覺得是有的，但是有前提條件，應用場景極其有限。基于ThreadLocal的設計，假設統計只在一個固定的線程池中進行，假設線程池中的線程不會銷毀(異常補充線程的就暫時不管了)，則可以認為線程數量是固定且不變的，那么統計則可以依賴于只在當前線程中進行，那么即使是高并發，就轉化為ThreadLocal這種單線程操作了，完全可以擺脫CAS的CPU指令操作的限制，那么性能將極大提升。
“并發計數類LongAdder怎么用”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
