併發編程——多線程計數的更優解：LongAdder原理分析
前言
最近在學習ConcurrentHashMap的源碼，發現它採用了一種比較獨特的方式對map中的元素數量進行統計，天然是要好好研究一下其原理思想，同時也能更好地理解ConcurrentHashMap自己。java
本文主要思路分爲如下4個部分算法
1.計數的使用效果編程
2.原理的直觀圖解數組
3.源碼的細節分析安全
4.與AtomicInteger的比較多線程
5.思想的抽象併發
學習的入口天然是map的put方法app
public V put(K key, V value) {
return putVal(key, value, false);
}
查看putVal方法dom
這裏並不對ConcurrentHashMap自己的原理做過多討論，所以咱們直接跳到計數部分ide
final V putVal(K key, V value, boolean onlyIfAbsent) {
...
addCount(1L, binCount);
return null;
}
每當成功添加一個元素以後，都會調用addCount方法進行數量的累加1的操做，這就是咱們研究的目標
由於ConcurrentHashMap的設計初衷就是爲了解決多線程併發場景下的map操做，所以在做數值累加的時候天然也要考慮線程安全
固然，多線程數值累加通常是學習併發編程的第一課，自己並不是很複雜，能夠採用AtomicInteger或者鎖等等方式來解決該問題
然而若是咱們查看該方法，就會發現，一個想來應該比較簡單的累加方法，其邏輯看上去卻至關複雜
這裏我只貼出了累加算法的核心部分
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
...
}
咱們就來研究一下該邏輯的實現思路。而這個思路實際上是照搬了LongAdder類的邏輯，所以咱們直接查看該算法的原始類
1.LongAdder類的使用
咱們先看下LongAdder的使用效果
LongAdder adder = new LongAdder();
int num = 0;
@Test
public void test5() throws InterruptedException {
Thread[] threads = new Thread[10];
for (int i = 0; i < 10; i++) {
threads[i] = new Thread(() -> {
for (int j = 0; j < 10000; j++) {
adder.add(1);
num += 1;
}
});
threads[i].start();
}
for (int i = 0; i < 10; i++) {
threads[i].join();
}
System.out.println("adder:" + adder);
System.out.println("num:" + num);
}
輸出結果
adder:100000 num:40982
能夠看到adder在使用效果上是能夠保證累加的線程安全的
2.LongAdder原理的直觀理解
爲了更好地對源碼進行分析，咱們須要先從直覺上理解它的原理，不然直接看代碼的話會一臉懵逼
LongAdder的計數主要分爲2個對象
一個long類型的字段：base
一個Cell對象數組，Cell對象中就維護了一個long類型的字段value，用來計數
/** * Table of cells. When non-null, size is a power of 2. */ transient volatile Cell[] cells; /** * Base value, used mainly when there is no contention, but also as * a fallback during table initialization races. Updated via CAS. */ transient volatile long base;
當沒有發生線程競爭的時候，累加都會發生在base字段上，這就至關因而一個單線程累加2次，只不過base的累加是一個cas操做
當發生線程競爭的時候，必然有一個線程對base的cas累加操做失敗，因而它先去判斷Cell是否已經被初始化了，若是沒有則初始化一個長度爲2的數組，並根據線程的hash值找到對應的數組索引，並對該索引的Cell對象中的value值進行累加（這個累加也是cas的操做）
若是一共有3個線程發生了競爭，那麼其中第一個線程對base的cas累加成功，剩下2個線程都須要去對Cell數組中的元素進行累加。由於對Cell中value值的累加也是一個cas操做，若是第二個線程和第三個線程的hash值對應的數組下標是同一個，那麼一樣會發生競爭，若是第二個線程成功了，第三個線程就會去rehash本身的hash值，若是獲得的新的hash值對應的是另外一個元素爲null的數組下標，那麼就new一個Cell對象並對value值進行累加
若是此時有線程4同時參與競爭，那麼對於線程4來講，即便rehash後仍是可能在和線程3的競爭過程當中cas失敗，此時若是當前數組的容量小於系統可用的cpu的數量，那麼它就會對數組進行擴容，以後再次rehash，重複嘗試對Cell數組中某個下標對象的累加
以上就是總體直覺上的理解，然而代碼中還有不少細節的設計很是值得學習，因此咱們就開始進入源碼分析的環節
3.源碼分析
入口方法是add
public void add(long x) {
Cell[] as; long b, v; int m; Cell a;
/**
* 這裏優先判斷了cell數組是否爲空，以後才判斷base字段的cas累加
* 意味着若是線程不發生競爭，cell數組一直爲空，那麼全部的累加操做都會累加到base上
* 而一旦發生過一次競爭致使cell數組不爲空，那麼全部的累加操做都會優先做用於數組中的對象上
*/
if ((as = cells) != null || !casBase(b = base, b + x)) {
/**
* 這個字段是用來標識在對cell數組中的對象進行累加操做時是否發生了競爭
* 若是發生了競爭，那麼在longAccumulate方法中會多進行一次rehash的自旋
* 這個在後面的方法中詳細說明，這裏先有個印象
* true表示未發生競爭
*/
boolean uncontended = true;
/**
* 若是cell數組爲空或者長度爲0則直接進入主邏輯方法
*/
if (as == null || (m = as.length - 1) < 0 ||
/**
* 這裏的getProbe()方法能夠認爲就是獲取線程的hash值
* hash值與(數組長度-1)進行位與操做後獲得對應的數組下標
* 判斷該元素是否爲空，若是不爲空那麼就會嘗試累加
* 不然進入主邏輯方法
*/
(a = as[getProbe() & m]) == null ||
/**
* 對數組下標的元素進行cas累加，若是成功了，那麼就能夠直接返回
* 不然進入主邏輯方法
*/
!(uncontended = a.cas(v = a.value, v + x)))
longAccumulate(x, null, uncontended);
}
}
當不發生線程競爭的時候，那累加操做就會由第一個if中的casBase負責，對應以前圖解的狀況一
當發生線程競爭以後，累加操做就會由cell數組負責，對應以前圖解的狀況二（數組的初始化在longAccumulate方法中）
接着咱們查看主邏輯方法，由於方法比較長，因此我會一段一段拿出來解析
longAccumulate方法
簽名中的參數
x表示須要累加的值
fn表示須要如何累加，通常傳null就行，不重要
wasUncontended表示是否在外層方法遇到了競爭失敗的狀況，由於外層的判斷邏輯是多個「或」(as == null || (m = as.length - 1) < 0 || (a = as[getProbe() & m]) == null)，因此若是數組爲空或者相應的下標元素還未初始化，這個字段就會保持false
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended) {
...
}
首先判斷線程的hash值是否爲0，若是爲0則須要作一個初始化，即rehash
以後會將wasUncontended置爲true，由於即便以前是衝突過的，通過rehash後就會先假設它能找到一個元素不衝突的數組下標
int h;//線程的hash值，在後面的邏輯中會用到
if ((h = getProbe()) == 0) {
ThreadLocalRandom.current(); // force initialization
h = getProbe();
wasUncontended = true;
}
以後是一個死循環，死循環中有3個大的if分支，這3個分支的邏輯做用於數組未初始化的時候，一旦數組初始化完成，那麼就都會進入主邏輯了，所以我這裏把主邏輯抽取出來放到後面單獨說，也能夠避免外層分支對思路的影響
/**
* 用來標記某個線程在上一次循環中找到的數組下標是否已經有Cell對象了
* 若是爲true，則表示數組下標爲空
* 在主邏輯的循環中會用到
*/
boolean collide = false;
/**
* 死循環，提供自旋操做
*/
for (; ; ) {
Cell[] as;
Cell a;
int n;//cell數組長度
long v;//須要被累積的值
/**
* 若是cells數組不爲空，且已經被某個線程初始化成功，那麼就會進入主邏輯，這個後面詳細解釋
*/
if ((as = cells) != null && (n = as.length) > 0) {
...
/**
* 若是數組爲空，那麼就須要初始化一個Cell數組
* cellsBusy用來標記cells數組是否能被操做，做用至關於一個鎖
* cells == as 判斷是否有其餘線程在當前線程進入這個判斷以前已經初始化了一個數組
* casCellsBusy 用一個cas操做給cellsBusy字段賦值爲1，若是成功能夠認爲拿到了操做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賦值
* 若是又沒競爭過（感受有點可憐），那麼就自旋
* 另外提一下方法簽名中的LongBinaryOperator對象就是用在這裏的，不影響邏輯
*/
} else if (casBase(v = base, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
接着就看對cell數組元素進行累加的主邏輯
/**
* 若是cells數組不爲空，且已經被某個線程初始化成功，進入主邏輯
*/
if ((as = cells) != null && (n = as.length) > 0) {
/**
* 若是當前線程的hash值對應的數組元素爲空
*/
if ((a = as[(n - 1) & h]) == null) {
/**
* Cell數組並未被其餘線程操做
*/
if (cellsBusy == 0) {
/**
* 這裏沒有理解做者爲何會在這裏初始化單個Cell
* 做者這裏的註釋是Optimistically create，若是有理解的同窗能夠說一下
*/
Cell r = new Cell(x);
/**
* 在此判斷cell鎖的狀態，並嘗試加鎖
*/
if (cellsBusy == 0 && casCellsBusy()) {
boolean created = false;
try {
/**
* 這裏對數組是否爲空等狀態再次進行校驗
* 若是校驗經過，那麼就將以前new的Cell對象放到Cell數組的該下標處
*/
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;
/**
* 走到這裏說明在判空和拿到鎖之間正好有其餘線程在該下標處建立了一個Cell
* 所以直接continue，不rehash，下次就不會進入到該分支了
*/
continue;
}
}
/**
* 當執行到這裏的時候，由於是在 if ((a = as[(n - 1) & h]) == null) 這個判斷邏輯中
* 就說明在第一個if判斷的時候該下標處沒有元素，因此賦值爲false
* collide的意義是：上一次循環中找到的數組下標是否已經有Cell對象了
* True if last slot nonempty
*/
collide = false;
/**
* 這個字段若是爲false，說明以前已經和其餘線程發過了競爭
* 即便此時能夠直接取嘗試cas操做，可是在高併發場景下
* 這2個線程以後依然可能發生競爭，而每次競爭都須要自旋的話會很浪費cpu資源
* 所以在這裏先直接增長自旋一次，在for的最後會作一次rehash
* 使得線程儘快地找到本身獨佔的數組下標
*/
} else if (!wasUncontended)
wasUncontended = true;
/**
* 嘗試給hash對應的Cell累加，若是這一步成功了，那麼就返回
* 若是這一步依然失敗了，說明此時總體的併發競爭很是激烈
* 那就可能須要考慮擴容數組了
* （由於數組初始化容量爲2，若是此時有10個線程在併發運行，那就很難避免競爭的發生了）
*/
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;
/**
* 這裏判斷下cpu的核數，由於即便有100個線程
* 能同時並行運行的線程數等於cpu數
* 所以若是數組的長度已經大於cpu數目了，那就不該當再擴容了
*/
else if (n >= NCPU || cells != as)
collide = false;
/**
* 走到這裏，說明當前循環中根據線程hash值找到的數組下標已經有元素了
* 若是此時collide爲false，說明上一次循環中找到的下邊是沒有元素的
* 那麼就自旋一次並rehash
* 若是再次運行到這裏，而且collide爲true，就說明明競爭很是激烈，應當擴容了
*/
else if (!collide)
collide = true;
/**
* 能運行到這裏，說明須要擴容數組了
* 判斷鎖狀態並嘗試獲取鎖
*/
else if (cellsBusy == 0 && casCellsBusy()) {
/**
* 擴容數組的邏輯，這個擴容比較簡單，就不解釋了
* 擴容大小爲2倍
*/
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，由於擴容過了，就先不rehash了
*/
continue;
}
/**
* 作一個rehash，使得線程在下一個循環中可能找到獨佔的數組下標
*/
h = advanceProbe(h);
}
到這裏LongAdder的源碼其實就分析結束了，其實代碼並很少，可是他的思想很是值得咱們去學習。
4.與AtomicInteger的比較
光分析源碼其實還差一些感受，咱們尚未搞懂爲什麼做者要在已經有AtomicInteger的狀況下，再設計這麼一個看上去很是複雜的類。
那麼首先咱們先分析下AtomicInteger保證線程安全的原理
查看最基本的getAndIncrement方法
public final int getAndIncrement() {
return unsafe.getAndAddInt(this, valueOffset, 1);
}
調用了Unsafe類的getAndAddInt方法，繼續往下看
public final int getAndAddInt(Object var1, long var2, int var4) {
int var5;
do {
var5 = this.getIntVolatile(var1, var2);
} while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
return var5;
}
這裏咱們再也不深究getIntVolatile和compareAndSwapInt方法具體實現，由於其已是native的方法了
能夠看到，AtomicInteger底層是使用了cas+自旋的方式解決原子性問題的，即若是一次賦值不成功，那麼就自旋，直到賦值成功爲止
那麼由此能夠推斷，當出現大量線程併發，競爭很是激烈的時候，AtomicInteger就有可能致使有些線程不斷地競爭失敗，不斷自旋從而影響任務的吞吐量
爲了解決高併發下的自旋問題，LongAdder的做者在設計的時候就經過增長一個數組的方式，使得競爭的對象從一個值變成多個值，從而使得發生競爭的頻率下降，從而緩解了自旋的問題，固然付出的代價就是額外的存儲空間。
最後我簡單作了個測試，比較2種計數方法的耗時
經過原理可知，只有當線程競爭很是激烈的時候，LongAdder的優點纔會比較明顯，所以這裏我用了100個線程，每個線程對同一個數累加1000000次，獲得結果以下，差距很是巨大，達到15倍！
LongAdder耗時：104292242nanos
AtomicInteger耗時：1583294474nanos
固然這只是一個簡單測試，包含了不少隨機性，有興趣的同窗能夠嘗試不一樣的競爭程度屢次測試
5.思想的抽象
最後咱們須要將做者的具體代碼和實現邏輯抽象一下，理清思考的過程
1）AtomicInteger遇到的問題：單個資源的競爭致使自旋的發生
2）解決的思路：將單個對象的競爭擴展爲多個對象的競爭（有那麼一些分治的思想）
3）擴展的可控性：多個競爭對象須要付出額外的存儲空間，所以不能無腦地擴展（極端狀況是一個線程一個計數的對象，這明顯不合理）
4）問題的分層：由於使用類的時候的場景是不可控的，所以須要根據併發的激烈程度動態地擴展額外的存儲空間（相似於synchronized的膨脹）
5）3個分層策略：當不發生競爭時，那麼用一個值累加便可；當發生必定程度的競爭時，建立一個容量爲2的數組，使得競爭的資源擴展爲3個；當競爭更加激烈時，則繼續擴展數組（對應圖解中的1個線程到4個線程的過程）
6）策略細節：在自旋的時候增長rehash，此時雖然付出了必定的運算時間計算hash、比較數組對象等，可是這會使得併發的線程儘快地找到專屬於本身的對象，在以後就不會再發生任何競爭（磨刀不誤砍柴工，特別注意wasUncontended字段的相關注解）
- 1. 多線程編程:併發加/減操做優化, LongAdder原理,與AtomicLong比較
- 2. 初識多線程和原理分析(併發編程一)
- 3. 併發編程之LongAdder
- 4. Java多線程——AtomicLong LongAdder源碼解析
- 5. 併發編程之美-深刻理解java多線程原理
- 6. 1-4併發編程之線程池原理分析
- 7. Java併發——線程池原理解析
- 8. 多線程併發編程分享
- 9. 多線程併發編程
- 10. 併發編程-多線程
- 更多相關文章...
- • Rust 併發編程 - RUST 教程
- • C# 多線程 - C#教程
- • Git五分鐘教程
- • TiDB 在摩拜單車在線數據業務的應用和實踐
-
每一个你不满意的现在，都有一个你没有努力的曾经。