並發編程——多線程計數的更優解:LongAdder原理分析

2020-12-22 探險家之指路明燈

前言

最近在學習ConcurrentHashMap的源碼,發現它採用了一種比較獨特的方式對map中的元素數量進行統計,自然是要好好研究一下其原理思想,同時也能更好地理解ConcurrentHashMap本身。

本文主要思路分為以下4個部分

1.計數的使用效果

2.原理的直觀圖解

3.源碼的細節分析

4.與AtomicInteger的比較

5.思想的抽象

學習的入口自然是map的put方法

public V put(K key, V value) { return putVal(key, value, false);}查看putVal方法

這裡並不對ConcurrentHashMap本身的原理作過多討論,因此我們直接跳到計數部分

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;@Testpublic 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:100000num: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欄位的相關註解)

如果覺得本文對你有幫助,可以轉發關注支持一下

作者:tera原文連結:https://www.cnblogs.com/tera/p/13886856.html

相關焦點

  • 「原創」Java並發編程系列07|synchronized原理
    synchronized是實現同步加鎖的原理?synchronized解決了並發編程的哪些問題?1.synchronized使用1.1 線程安全問題並發編程中,當多個線程同時訪問同一個資源的時候,就會存在線程安全問題。由於每個線程執行的過程是不可控的,所以很可能導致最終的結果與實際期望的結果相違背或者直接導致程序出錯。
  • 阿里P9都窺視已久的「Java並發實現原理:JDK源碼剖析」
    前言並發編程可選擇的方式有多進程、多線程和多協程。對於Java來說,它既不像C++那樣,在運行中調用Linux的系統API去「fork」出多個進程;也不像Go那樣,在語言層面原生提供多協程。在Java中,並發就是多線程模式。
  • 原創】Java並發編程系列01|開篇獲獎感言
    所以,並發編程已經成為一項必備技能。並發編程是Java語言的重要特性之一,它能使複雜的代碼變得更簡單,從而極大地簡化複雜系統的開發。並發編程可以充分發揮多處理器系統的強大計算能力,隨著處理器數量的持續增長,如何高效的並發變得越來越重要。
  • Java多線程並發容器之並發倒計時器
    從火箭發場景來學習Java多線程並發閉鎖對象倒計時器場景在我們開發過程中,有時候會使用到倒計時計數器。最簡單的是:int size = 5; 執行後,size—這種方式來實現。但是在多線程並發的情況下,這種操作會不安全的。舉個現實中最典型的一個例子:火箭發射的案例。
  • JAVA並發編程:並發問題的根源及主要解決方法
    計算機的發展有一部分就是如何重複利用資源,解決硬體資源之間效率的不平衡,而後就有了多進程,多線程的發展。ThreadLocal會為每個線程提供一個本地副本,每個線程都只會修改自己的ThreadLocal變量。這樣一來就不會出現共享變量,也就不會出現衝突了。其實現原理是在ThreadLocal內部維護一個ThreadLocalMap,每次有線程要獲取對應變量的時候,先獲取當前線程,然後根據不同線程取不同的值,典型的以空間換時間。
  • Python並發編程很簡單,一文幫你搞清如何創建線程類
    對於Python的並發編程相關的東東,相信通過上次咱們的探討,大家已經比較清楚了,對於Python創建線程的方式主要有兩種,這個上次咱們也已經說過了哦,第一種是使用threading模塊的Thread類的構造器來創建線程,這種方式上次咱們已經詳細討論過了哦,這次呢,咱們就重點和大家來聊聊第二種方式吧
  • 「原創」Java並發編程系列02|並發編程三大核心問題
    要快速準確的發現並解決這些問題,首先就是要弄清並發編程的本質,並發編程要解決的是什麼問題。本文將帶你深入理解並發編程要解決的三大問題:原子性、可見性、有序性。補充知識硬體的發展中,一直存在一個矛盾,CPU、內存、I/O設備的速度差異。
  • Java並發編程:多線程如何實現阻塞與喚醒
    線程的阻塞和喚醒在多線程並發過程中是一個關鍵點,當線程數量達到很大的數量級時,並發可能帶來很多隱蔽的問題。如何正確暫停一個線程,暫停後又如何在一個要求的時間點恢復,這些都需要仔細考慮的細節。
  • Java並發編程:如何防止在線程阻塞與喚醒時死鎖
    Java並發編程:多線程如何實現阻塞與喚醒說到suspend與resume組合有死鎖傾向,一不小心將導致很多問題,甚至導致整個系統崩潰。接著看另外一種解決方案,我們可以使用以對象為目標的阻塞,即利用Object類的wait()和notify()方法實現線程阻塞。
  • 「原創」Java並發編程系列09|基礎乾貨
    本文轉載自【微信公眾號:java進階架構師,ID:java_jiagoushi】經微信公眾號授權轉載,如需轉載與原文作者聯繫本文為何適原創並發編程系列第9篇。現在,我們進入正題:介紹並發編程的基礎性概念。
  • JAVA高並發網絡編程之BIO堵塞網絡編程
    上次說了網絡編程都是有作業系統統一的API的,每個語言有對它的實現,這次來一起說說通過java原生的socket編程完成BIO的網絡編程。
  • Java並發編程系列23|循環屏障CyclicBarrier
    本文轉載自【微信公眾號:java進階架構師,ID:java_jiagoushi】經微信公眾號授權轉載,如需轉載與原文作者聯繫本篇介紹第二個並發工具類CyclicBarrier,CyclicBarrier的字面意思是可循環使用(Cyclic)的屏障(Barrier),分以下部分介紹:CyclicBarrier的使用CyclicBarrier與CountDownLatch
  • 多線程有哪些優點?-Python每日3題(多線程專題)
    這裡是Python7編程挑戰-多線程專題! 每天學習3個問題,包括初級,中級,高級問題各1個。 今天是第2天!一起來呀,就7天! 參加方法:關注麥叔編程公眾號,回復「7天」入群學習和討論。
  • 新手編程:Java多線程中Thread與Runnable的區別
    Java多線程中Thread與Runnable的區別定義extends Thread子類繼承Thread具備多線程能力,可以實現多線程;啟動線程的方法:①創建子類對象②對象名.start();不建議使用:避免多線程OOP單繼承的局限性(OOP:Object Oriented Programming,面向對象的編程、類似的還有OOD(面向對象的設計),OOA(面向對象的分析));implements Runnable
  • 「原創」Java並發編程系列18|讀寫鎖(下)
    本文轉載自【微信公眾號:java進階架構師,ID:java_jiagoushi】經微信公眾號授權轉載,如需轉載與原文作者聯繫本文為何適原創並發編程系列第 18 篇,文末有本系列文章匯總。上篇為【原創】Java並發編程系列17 | 讀寫鎖八講(上),沒看過的可以先看看。本文是下篇,從「源碼分析寫鎖的獲取與釋放」開始。7.
  • 「原創」Java並發編程系列14|AQS源碼分析
    本文轉載自【微信公眾號:java進階架構師,ID:java_jiagoushi】經微信公眾號授權轉載,如需轉載與原文作者聯繫本文為何適原創並發編程系列第 14 篇,文末有本系列文章匯總。AbstractQueuedSynchronizer是Java並發包java.util.concurrent的核心基礎組件,是實現Lock的基礎。
  • 「原創」Java並發編程系列06|你不知道的final
    那麼fianl在並發編程中有什麼作用呢?本文就在對final常見應用總結基礎上,講解final並發編程中的應用。1. final基礎應用final變量final變量只能被賦值一次,賦值後值不再改變。final變量可以安全的在多線程環境下進行共享,而不需要額外的同步開銷。2.
  • Java項目實踐,CountDownLatch實現多線程閉鎖
    摘要本文主要介紹Java多線程並發中閉鎖(Latch)的基本概念、原理、示例代碼、應用場景,通過學習,可以掌握多線程並發時閉鎖(Latch)的使用方法。概念「閉鎖」就是指一個被鎖住了的門將線程a擋在了門外(等待執行),只有當門打開後(其他線程執行完畢),門上的鎖才會被打開,a才能夠繼續執行。
  • Java 並發底層知識,鎖獲取超時機制知多少?
    當我們在使用Java進行網絡編程時經常會遇到很多超時的概念,比如一個瀏覽器請求過程就可能會產生很多超時的地方,當我們在瀏覽器發起一個請求後,網絡socket讀寫可能會超時,web伺服器響應可能會超時,資料庫查詢可能會超時。
  • Go 語言並發 M-P-G 模型
    Go 語言自從誕生以來就自帶「高並發」的buff,而並發編程也是當今開發環境的一個大方向了。目前的各大語言基本上也都提供了原生多線程的並發編程模式。但是 Go 語言的並發模型和 C++,Java 的並發模型是有區別的。