03_LongAdder 源碼分析

• AtomicLong 和 LongAdder對比
• AtomicLong
• LongAdder
• • LongAdder和AtomicLong性能測試
  • LongAdder為什麼這麼快
  • • 1、 設計思想上，LongAdder采用**"分段"的方式降低CAS失敗的頻次**
    • 2、**使用Contended注解來消除僞共享**
    • **3、惰性求值**
• LongAdder源碼剖析
• AtomicLong可以棄用了嗎？
• 總結

上文中我們分析了:CAS底層實作是在一個死循環中不斷地嘗試修改目标值，直到修改成功。如果競争不激烈的時候，修改成功率很高，否則失敗率很高。在失敗的時候，這些重複的原子性操作會耗費性能。（不停的自旋，進入一個無限重複的循環中）

那我們今天分析下LongAdder這個類

AtomicLong 和 LongAdder對比

最近阿裡巴巴釋出了Java開發手冊(泰山版) (公衆号回複： 開發手冊 可收到阿裡巴巴開發手冊(泰山版 2020.4.22釋出).pdf)，其中第17條寫到：

對于Java項目中計數統計的一些需求，如果是 JDK8，推薦使用 LongAdder 對象，比 AtomicLong 性能更好（減少樂觀鎖的重試次數）

在大多數項目及開源元件中，計數統計使用最多的仍然還是AtomicLong，雖然是阿裡巴巴這樣說，但是我們仍然要根據使用場景來決定是否使用LongAdder。

今天主要是來講講LongAdder的實作原理，還是老方式，通過圖文一步步解開LongAdder神秘的面紗，通過此篇文章你會了解到：

• 為什麼AtomicLong在高并發場景下性能急劇下降？
• LongAdder為什麼快？
• LongAdder實作原理（圖文分析）

• AtomicLong是否可以被遺棄或替換？

  本文代碼全部基于JDK 1.8，建議邊看文章邊看源碼更加利于消化

AtomicLong

上文分析cas中我們知道AtomicXX 都是利用cas原理實作的,這裡再單獨分析下AtomicLong源碼

當我們在進行計數統計的時，通常會使用AtomicLong來實作。AtomicLong能保證并發情況下計數的準确性，其内部通過CAS來解決并發安全性的問題。

AtomicLong實作原理

說到線程安全的計數統計工具類，肯定少不了Atomic下的幾個原子類。AtomicLong就是juc包下重要的原子類，在并發情況下可以對長整形類型資料進行原子操作，保證并發情況下資料的安全性。

public class AtomicLong extends Number implements java.io.Serializable {
    public final long incrementAndGet() {
        return unsafe.getAndAddLong(this, valueOffset, 1L) + 1L;
    }

    public final long decrementAndGet() {
        return unsafe.getAndAddLong(this, valueOffset, -1L) - 1L;
    }
}
           

我們在計數的過程中，一般使用incrementAndGet()和decrementAndGet()進行加一和減一操作，這裡調用了Unsafe類中的getAndAddLong()方法進行操作。

接着看看unsafe.getAndAddLong()方法：

public final class 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;
    }

    public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);
}
           

這裡直接進行CAS+自旋操作更新AtomicLong中的value值，進而保證value值的原子性更新。

AtomicLong瓶頸分析

如上代碼所示，我們在使用CAS + 自旋的過程中，在高并發環境下，N個線程同時進行自旋操作，會出現大量失敗并不斷自旋的情況，此時AtomicLong的自旋會成為瓶頸。

如上圖所示，高并發場景下AtomicLong性能會急劇下降，我們後面也會舉例說明。

那麼高并發下計數的需求有沒有更好的替代方案呢？在JDK8 中 Doug Lea大神 新寫了一個LongAdder來解決此問題，我們後面來看LongAdder是如何優化的。

LongAdder

LongAdder和AtomicLong性能測試

我們說了很多LongAdder上性能優于AtomicLong，到底是不是呢？一切還是以代碼說話：

public class testAtomicLongAdder {

    public static void main(String[] args) throws InterruptedException {
        testAtomicLongAdder(1,10000000);
        testAtomicLongAdder(10,10000000);
        testAtomicLongAdder(100,10000000);

    }

    public static void testAtomicLongAdder(int threads, int times) throws InterruptedException {
        System.out.println("線程數:"+ threads+ "; 累加次數: " + times);
        long startAtomic = System.currentTimeMillis();
        testTtomicLong(threads,times);
        long endTomic = System.currentTimeMillis();

        long startLongadder = System.currentTimeMillis();
        testLongAdder(threads,times);
        long endLongadder = System.currentTimeMillis();
        System.out.println("AtomicLong 耗時:" + (endTomic-startAtomic) + "\n LongAdder 耗時: " + (endLongadder-startLongadder));
        System.out.println("---------------------");
    }
    private static void testTtomicLong(int threads, int times) throws InterruptedException {
        AtomicLong atomicLong = new AtomicLong(0);
        List<Thread> list = new ArrayList<>();
        for (int i = 1; i <=threads; i++) {
            list.add(new Thread(()->{
                for (int j = 0; j < times; j++) {
                    atomicLong.incrementAndGet();
                }
            }));

        }

        for (Thread thread : list) {
            thread.start();
        }

        for (Thread thread : list) {
            thread.join();
        }

        System.out.println("AtomicLong value is : " + atomicLong.get());
    }
    private static void testLongAdder(int threads, int times) throws InterruptedException {
        LongAdder longAdder = new LongAdder();
        List<Thread> list = new ArrayList<>();
        for (int i = 0; i < threads; i++) {
            list.add(new Thread(() -> {
                for (int j = 0; j < times; j++) {
                    longAdder.increment();
                }
            }));
        }

        for (Thread thread : list) {
            thread.start();
        }

        for (Thread thread : list) {
            thread.join();
        }

        System.out.println("LongAdder value is : " + longAdder.longValue());

    }


}
           

輸出結果:

這裡可以看到随着并發的增加，AtomicLong性能是急劇下降的，耗時是LongAdder的數倍。至于原因我們還是接着往後看。

LongAdder為什麼這麼快

先看下LongAdder的操作原理圖

既然說到LongAdder可以顯著提升高并發環境下的性能，那麼它是如何做到的？

1、 設計思想上，LongAdder采用**"分段"的方式降低CAS失敗的頻次**

這裡先簡單的說下LongAdder的思路，後面還會詳述LongAdder的原理。

我們知道，AtomicLong中有個内部變量value儲存着實際的long值，所有的操作都是針對該變量進行。也就是說，高并發環境下，value變量其實是一個熱點資料，也就是N個線程競争一個熱點。

LongAdder的基本思路就是分散熱點，将value值的新增操作分散到一個數組中，不同線程會命中到數組的不同槽中，各個線程隻對自己槽中的那個value值進行CAS操作，這樣熱點就被分散了，沖突的機率就小很多。

LongAdder有一個全局變量volatile long base值，當并發不高的情況下都是通過CAS來直接操作base值，如果CAS失敗，則針對LongAdder中的Cell[]數組中的Cell進行CAS操作，減少失敗的機率。

例如目前類中base = 10，有三個線程進行CAS原子性的+1操作，線程一執行成功，此時base=11，線程二、線程三執行失敗後開始針對于Cell[]數組中的Cell元素進行+1操作，同樣也是CAS操作，此時數組index=1和index=2中Cell的value都被設定為了1.

執行完成後，統計累加資料：sum = 11 + 1 + 1 = 13，利用LongAdder進行累加的操作就執行完了，流程圖如下：

如果要擷取真正的long值，隻要将各個槽中的變量值累加傳回。這種分段的做法類似于JDK7中ConcurrentHashMap的分段鎖。

2、使用Contended注解來消除僞共享

在 LongAdder 的父類 Striped64 中存在一個 volatile Cell[] cells; 數組，其長度是2 的幂次方，每個Cell都使用 @Contended 注解進行修飾，而**@Contended注解可以進行緩存行填充，進而解決僞共享問題**。僞共享會導緻緩存行失效，緩存一緻性開銷變大。

@sun.misc.Contended static final class Cell {

}
           

其中 Cell 即為累加單元

這裡說的僞共享是指的緩存僞共享 ,這個是什麼概念?

我們不得不從緩存說起: 大家都知道緩存的速度比記憶體的速度快很多

因為 CPU 與 記憶體的速度差異很大，需要靠預讀資料至緩存來提升效率。

而緩存以緩存行為機關，每個緩存行對應着一塊記憶體，一般是 64 byte（8 個 long）

緩存的加入會造成資料副本的産生，即同一份資料會緩存在不同核心的緩存行中

CPU 要保證資料的一緻性，如果某個 CPU 核心更改了資料，其它 CPU 核心對應的整個緩存行必須失效

個人簡單總結下: @sun.misc.Contended 就是讓緩存行隻存儲一個對象或者字段,避免一個失效後影響另外一個對象或者字段

3、惰性求值

LongAdder隻有在使用longValue()擷取目前累加值時才會真正的去結算計數的資料，longValue()方法底層就是調用sum()方法，對base和Cell數組的資料累加然後傳回，做到資料寫入和讀取分離。

而AtomicLong使用incrementAndGet()每次都會傳回long類型的計數值，每次遞增後還會伴随着資料傳回，增加了額外的開銷。

LongAdder實作原理

之前說了，AtomicLong是多個線程針對單個熱點值value進行原子操作。而LongAdder是每個線程擁有自己的槽，各個線程一般隻對自己槽中的那個值進行CAS操作。

比如有三個線程同時對value增加1，那麼value = 1 + 1 + 1 = 3

但是對于LongAdder來說，内部有一個base變量，一個Cell[]數組。

base變量：非競态條件下，直接累加到該變量上

Cell[]數組：競态條件下，累加個各個線程自己的槽Cell[i]中

最終結果的計算是下面這個形式：

value = base +

LongAdder源碼剖析

LongAdder 父類中有這麼幾個關鍵的屬性

LongAdder源碼

public void increment() {
        add(1L);
    }
 public void add(long x) {
        Cell[] as; long b, v; int m; Cell a;
        //進入if的兩個條件
        //1.as有值,表示已經發生過競争
        //2.casBase() 失敗,給base累加失敗,表示存在競争
        if ((as = cells) != null || !casBase(b = base, b + x)) {
        	//uncontended 表示沒有競争
            boolean uncontended = true;
            if (
            //cell[] 還沒建立
            as == null || (m = as.length - 1) < 0 ||
            	//目前線程的對應cell還沒有
                (a = as[getProbe() & m]) == null ||
                //目前線程的cell累加失敗 a為目前線程的cell
                !(uncontended = a.cas(v = a.value, v + x)))
                //進入cell數組建立,cell建立
                longAccumulate(x, null, uncontended);
        }
    }   
           

一般我們進行計數時都會使用increment()方法，每次進行+1操作，increment()會直接調用add()方法。

變量說明：

• as 表示cells引用
• b 表示擷取的base值
• v 表示 期望值,
• m 表示 cells 數組的長度
• a 表示目前線程命中的cell單元格

條件分析：

條件一：as == null || (m = as.length - 1) < 0

此條件成立說明cells數組未初始化。如果不成立則說明cells數組已經完成初始化，對應的線程需要找到Cell數組中的元素去寫值。

條件二：(a = as[getProbe() & m]) == null

getProbe()擷取目前線程的hash值，m表示cells長度-1，cells長度是2的幂次方數，原因之前也講到過，與數組長度取模可以轉化為按位與運算，提升計算性能。

當條件成立時說明目前線程通過hash計算出來數組位置處的cell為空，進一步去執行longAccumulate()方法。如果不成立則說明對應的cell不為空，下一步将要将x值通過CAS操作添加到cell中。

條件三：!(uncontended = a.cas(v = a.value, v + x)

主要看a.cas(v = a.value, v + x)，接着條件二，說明目前線程hash與數組長度取模計算出的位置的cell有值，此時直接嘗試一次CAS操作，如果成功則退出if條件，失敗則繼續往下執行longAccumulate()方法。

接着往下看核心的longAccumulate()方法，代碼很長，後面會一步步分析，先上代碼：

final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
        int h;
        //目前線程沒有對應的cell,需要随機生成一個h值,綁定到目前線程的cell上
        if ((h = getProbe()) == 0) {
        	//初始化 probe
            ThreadLocalRandom.current(); // force initialization
            //h對象新的probe值,用來對應cell
            h = getProbe();
            wasUncontended = true;
        }
        //collide 為true 說明cells 需要擴容
        boolean collide = false;                // True if last slot nonempty
        for (;;) {  //這是一個死循環
            Cell[] as; Cell a; int n; long v;
            if ((as = cells) != null && (n = as.length) > 0) {
                if ((a = as[(n - 1) & h]) == null) {
                    if (cellsBusy == 0) {       // Try to attach new Cell
                        Cell r = new Cell(x);   // Optimistically create
                        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 {
                                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
                else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                             fn.applyAsLong(v, x))))
                    break;
                else if (n >= NCPU || cells != as)
                    collide = false;            // At max size or stale
                else if (!collide)
                    collide = true;
                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
                }
                h = advanceProbe(h);
            }
            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;
            }
            else if (casBase(v = base, ((fn == null) ? v + x :
                                        fn.applyAsLong(v, x))))
                break;                          // Fall back on using base
        }
    }
           

代碼很長，if else分支很多，除此看肯定會很頭疼。這裡一點點分析，然後結合畫圖一步步了解其中實作原理。

我們首先要清楚執行這個方法的前置條件，它們是或的關系，如上面條件一、二、三

• cells數組已經初始化
• cells數組未初始化，嘗試給CellsBusy加鎖
• cells數組未經初始化，嘗試給CellsBusy加鎖失敗,嘗試給base加鎖

longAccumulate()方法的入參：

long x 需要增加的值，一般預設都是1

LongBinaryOperator fn 預設傳遞的是null

wasUncontended競争辨別，如果是false則代表有競争。隻有cells初始化之後，并且目前線程CAS競争修改失敗，才會是false

然後再看下Striped64中一些變量或者方法的定義：

base: 類似于AtomicLong中全局的value值。在沒有競争情況下資料直接累加到base上，或者cells擴容時，也需要将資料寫入到base上

collide：表示擴容意向，false 一定不會擴容，true可能會擴容。

cellsBusy：初始化cells或者擴容cells需要擷取鎖, 0:表示無鎖狀态 1:表示其他線程已經持有了鎖

casCellsBusy(): 通過CAS操作修改cellsBusy的值，CAS成功代表擷取鎖，傳回true

NCPU：目前計算機CPU數量，Cell數組擴容時會使用到

getProbe(): 擷取目前線程的hash值

advanceProbe(): 重置目前線程的hash值

private static final long PROBE;

if ((h = getProbe()) == 0) {
    ThreadLocalRandom.current();
    h = getProbe();
    wasUncontended = true;
}

static final int getProbe() {
    return UNSAFE.getInt(Thread.currentThread(), PROBE);
}
           

我們上面說過getProbe()方法是為了擷取目前線程的hash值，具體實作是通過UNSAFE.getInt()實作的，PROBE是在初始化時候擷取目前線程threadLocalRandomProbe的值。

注：Unsafe.getInt()有三個重載方法getInt(Object o, long offset)、getInt(long address) 和getIntVolatile(long address)，都是從指定的位置擷取變量的值，隻不過第一個的offset是相對于對象o的相對偏移量，第二個address是絕對位址偏移量。如果第一個方法中o為null是，offset也會被作為絕對偏移量。第三個則是帶有volatile語義的load讀操作。

如果目前線程的hash值h=getProbe()為0，0與任何數取模都是0，會固定到數組第一個位置，是以這裡做了優化，使用ThreadLocalRandom為目前線程重新計算一個hash值。最後設定wasUncontended = true，這裡含義是重新計算了目前線程的hash後認為此次不算是一次競争。hash值被重置就好比一個全新的線程一樣，是以設定了競争狀态為true。

可以畫圖了解為：

接着執行for循環，我們可以把for循環代碼拆分一下，每個if條件算作一個CASE來分析：

final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) {

    for (;;) {
        Cell[] as; Cell a; int n; long v;
        if ((as = cells) != null && (n = as.length) > 0) {

        }
        else if (cellsBusy == 0 && cells == as && casCellsBusy()) {

        }
        else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x))))

    }
}
           

如上所示，第一個if語句代表CASE1，裡面再有if判斷會以CASE1.1這種形式來講解，下面接着的else if為CASE2， 最後一個為CASE3

CASE1執行條件：

if ((as = cells) != null && (n = as.length) > 0) {

}
           

cells數組不為空，且數組長度大于0的情況會執行CASE1，CASE1的實作細節代碼較多，放到最後面講解。

我們這裡先分析

else if (cellsBusy == 0 && cells == as && casCellsBusy())

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;
            }
           

CASE2 辨別cells數組還未初始化，因為判斷cells == as，這個代表目前線程到了這裡擷取的cells還是之前的一緻。我們可以先看這個case，最後再回頭看最為麻煩的CASE1實作邏輯。

cellsBusy上面說了是加鎖的狀态，初始化cells數組和擴容的時候都要擷取加鎖的狀态，這個是通過CAS來實作的，為0代表無鎖狀态，為1代表其他線程已經持有鎖了。cellsas代表目前線程持有的數組未進行修改過，casCellsBusy()通過CAS操作去擷取鎖。但是裡面的if條件又再次判斷了cellas，這一點是不是很奇怪？通過畫圖來說明下問題：

如果上面條件都執行成功就會執行數組的初始化及指派操作， Cell[] rs = new Cell[2]表示數組的長度為2，rs[h & 1] = new Cell(x) 表示建立一個新的Cell元素，value是x值，預設為1。

h & 1類似于我們之前HashMap或者ThreadLocal裡面經常用到的計算散列桶index的算法，通常都是hash & (table.len - 1)，這裡就不做過多解釋了。 執行完成後直接退出for循環。

CASE3執行條件和實作原理：

else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x))))
    break;
           

進入到這裡說明cells正在或者已經初始化過了，執行caseBase()方法，通過CAS操作來修改base的值，如果修改成功則跳出循環，這個CASE隻有在初始化Cell數組的時候，多個線程嘗試CAS修改cellsBusy加鎖的時候，失敗的線程會走到這個分支，然後直接CAS修改base資料。

CASE1 實作原理：

分析完了CASE2和CASE3，我們再折頭回看一下CASE1，進入CASE1的前提是：cells數組不為空，已經完成了初始化指派操作。

接着還是一點點往下拆分代碼，首先看第一個判斷分支CASE1.1：

if ((a = as[(n - 1) & h]) == null) {
    if (cellsBusy == 0) {
        Cell r = new Cell(x);
        if (cellsBusy == 0 && casCellsBusy()) {
            boolean created = false;
            try {
                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;
            continue;
        }
    }
    collide = false;
}
           

這個if條件中(a = as[(n - 1) & h]) == null代表目前線程對應的數組下标位置的cell資料為null，代表沒有線程在此處建立Cell對象。

接着判斷cellsBusy==0，代表目前鎖未被占用。然後新建立Cell對象，接着又判斷了一遍cellsBusy == 0，然後執行casCellsBusy()嘗試通過CAS操作修改cellsBusy=1，加鎖成功後修改擴容意向collide = false;

for (;;) {
    if ((rs = cells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) {
        rs[j] = r;
        created = true;
    }

    if (created)
        break;
    continue;
}
           

上面代碼判斷目前線程hash後指向的資料位置元素是否為空，如果為空則将cell資料放入數組中，跳出循環。如果不為空則繼續循環。

繼續往下看代碼，CASE1.2：

else if (!wasUncontended)
    wasUncontended = true;

h = advanceProbe(h);
           

wasUncontended表示cells初始化後，目前線程競争修改失敗wasUncontended =false，這裡隻是重新設定了這個值為true，緊接着執行advanceProbe(h)重置目前線程的hash，重新循環。

接着看CASE1.3：

else if (a.cas(v = a.value, ((fn == null) ? v + x : fn.applyAsLong(v, x))))
    break;
           

進入CASE1.3說明目前線程對應的數組中有了資料，也重置過hash值，這時通過CAS操作嘗試對目前數中的value值進行累加x操作，x預設為1，如果CAS成功則直接跳出循環。

else if (n >= NCPU || cells != as)
    collide = false;   
           

如果cells數組的長度達到了CPU核心數，或者cells擴容了，設定擴容意向collide為false并通過下面的h = advanceProbe(h)方法修改線程的probe再重新嘗試

至于這裡為什麼要提出和CPU數量做判斷的問題：每個線程會通過線程對cells[threadHash%cells.length]位置的Cell對象中的value做累加，這樣相當于将線程綁定到了cells中的某個cell對象上，如果超過CPU數量的時候就不再擴容是因為CPU的數量代表了機器處理能力，當超過CPU數量時，多出來的cells數組元素沒有太大作用。

接着看CASE1.5：

else if (!collide)
   collide = true;
           

如果擴容意向collide是false則修改它為true，然後重新計算目前線程的hash值繼續循環，在CASE1.4中，如果目前數組的長度已經大于了CPU的核數，就會再次設定擴容意向collide=false，這裡的意義是保證擴容意向為false後不再繼續往後執行CASE1.6的擴容操作。

接着看CASE1.6分支：

else if (cellsBusy == 0 && casCellsBusy()) {
    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;
}
           

這裡面執行的其實是擴容邏輯，首先是判斷通過CAS改變cellsBusy來嘗試加鎖，如果CAS成功則代表擷取鎖成功，繼續向下執行，判斷目前的cells數組和最先指派的as是同一個，代表沒有被其他線程擴容過，然後進行擴容，擴容大小為之前的容量的兩倍，這裡用的按位左移1位來操作的。

Cell[] rs = new Cell[n << 1];

擴容後再将之前數組的元素拷貝到新數組中，釋放鎖設定cellsBusy = 0，設定擴容狀态，然後繼續循環執行。

到了這裡，我們已經分析完了longAccumulate()所有的邏輯，邏輯分支挺多，仔細分析看看其實還是挺清晰的，流程圖如下：

我們再舉一些線程執行的例子裡面場景覆寫不全，大家可以按照這種模式自己模拟場景分析代碼流程：

如有問題也請及時指出，我會第一時間更正，不勝感激！

LongAdder的sum方法

當我們最終擷取計數器值時，我們可以使用LongAdder.longValue()方法，其内部就是使用sum方法來彙總資料的。

java.util.concurrent.atomic.LongAdder.sum():

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;
}
           

實作很簡單，base + ，周遊cells數組中的值，然後累加。

AtomicLong可以棄用了嗎？

看上去LongAdder的性能全面超越了AtomicLong，而且阿裡巴巴開發手冊也提及到 推薦使用 LongAdder 對象，比 AtomicLong 性能更好（減少樂觀 鎖的重試次數），但是我們真的就可以舍棄掉LongAdder了嗎？

當然不是，我們需要看場景來使用，如果是并發不太高的系統，使用AtomicLong可能會更好一些，而且記憶體需求也會小一些。

我們看過sum()方法後可以知道LongAdder在統計的時候如果有并發更新，可能導緻統計的資料有誤差。

而在高并發統計計數的場景下，才更适合使用LongAdder。

總結

LongAdder中最核心的思想就是利用空間來換時間，将熱點value分散成一個Cell清單來承接并發的CAS，以此來提升性能。

LongAdder的原理及實作都很簡單，但其設計的思想值得我們品味和學習。