前言

思維導圖.png

文章中所有高清無碼圖片在公眾號號回覆: 圖片666 即可查閱, 可直接關注公眾號：壹枝花算不算浪漫

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

阿里巴巴開發手冊.png

對於Java專案中計數統計的一些需求，如果是 JDK8，推薦使用 LongAdder 物件，比 AtomicLong 效能更好（減少樂觀鎖的重試次數）

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

。

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

• 為什麼AtomicLong在高併發場景下效能急劇下降？
• LongAdder為什麼快？
• LongAdder實現原理（圖文分析）
• AtomicLong是否可以被遺棄或替換？

本文程式碼全部基於JDK 1.8，建議邊看文章邊看原始碼更加利於消化

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瓶頸分析.png

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

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

LongAdder

LongAdder和AtomicLong效能測試

我們說了很多LongAdder上效能優於AtomicLong，到底是不是呢？一切還是以程式碼說話：

/**
 * Atomic和LongAdder耗時測試
 *
 * @author：一枝花算不算浪漫
 * @date：2020-05-12 7:06
 */
public class AtomicLongAdderTest {
    public static void main(String[] args) throws Exception{
        testAtomicLongAdder(1, 10000000);
        testAtomicLongAdder(10, 10000000);
        testAtomicLongAdder(100, 10000000);
    }

    static void testAtomicLongAdder(int threadCount, int times) throws Exception{
        System.out.println("threadCount: " + threadCount + ", times: " + times);
        long start = System.currentTimeMillis();
        testLongAdder(threadCount, times);
        System.out.println("LongAdder 耗時：" + (System.currentTimeMillis() - start) + "ms");
        System.out.println("threadCount: " + threadCount + ", times: " + times);
        long atomicStart = System.currentTimeMillis();
        testAtomicLong(threadCount, times);
        System.out.println("AtomicLong 耗時：" + (System.currentTimeMillis() - atomicStart) + "ms");
        System.out.println("----------------------------------------");
    }

    static void testAtomicLong(int threadCount, int times) throws Exception{
        AtomicLong atomicLong = new AtomicLong();
        List<Thread> list = Lists.newArrayList();
        for (int i = 0; i < threadCount; 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());
    }

    static void testLongAdder(int threadCount, int times) throws Exception{
        LongAdder longAdder = new LongAdder();
        List<Thread> list = Lists.newArrayList();
        for (int i = 0; i < threadCount; 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());
    }
}

執行結果：

CAS原理圖.png

這裡可以看到隨著併發的增加，AtomicLong效能是急劇下降的，耗時是LongAdder的數倍。至於原因我們還是接著往後看。

LongAdder為什麼這麼快

先看下LongAdder的操作原理圖：

YUnlDO.png

既然說到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進行累加的操作就執行完了，流程圖如下：

分段加鎖思路.png

如果要獲取真正的long值，只要將各個槽中的變數值累加返回。這種分段的做法類似於JDK7中ConcurrentHashMap的分段鎖。

2、使用Contended註解來消除偽共享

在 LongAdder 的父類 Striped64 中存在一個 volatile Cell[] cells; 陣列，其長度是2 的冪次方，每個Cell都使用 @Contended 註解進行修飾，而@Contended註解可以進行快取行填充，從而解決偽共享問題。偽共享會導致快取行失效，快取一致性開銷變大。

@sun.misc.Contended static final class Cell {

}

偽共享指的是多個執行緒同時讀寫同一個快取行的不同變數時導致的 CPU快取失效。儘管這些變數之間沒有任何關係，但由於在主記憶體中鄰近，存在於同一個快取行之中，它們的相互覆蓋會導致頻繁的快取未命中，引發效能下降。這裡對於偽共享我只是提一下概念，並不會深入去講解，大家可以自行查閱一些資料。

解決偽共享的方法一般都是使用直接填充，我們只需要保證不同執行緒的變數存在於不同的 CacheLine 即可，使用多餘的位元組來填充可以做點這一點，這樣就不會出現偽共享問題。例如在Disruptor佇列的設計中就有類似設計(可參考我之前的部落格文章：Disruptor學習筆記)：

快取行填充程式碼.png 快取行填充.png

在Striped64類中我們可以看看Doug Lea在Cell上加的註釋也有說明這一點：

Cell註釋.png

紅框中的翻譯如下：

Cell類是AtomicLong添加了padded（[email protected])來消除偽共享的變種版本。快取行填充對於大多數原子來說是繁瑣的，因為它們通常不規則地分散在記憶體中，因此彼此之間不會有太大的干擾。但是，駐留在陣列中的原子物件往往彼此相鄰，因此在沒有這種預防措施的情況下，通常會共享快取行資料（對效能有巨大的負面影響）。

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 class LongAdder extends Striped64 implements Serializable {
    public void increment() {
        add(1L);
    }

    public void add(long x) {
        Cell[] as; long b, v; int m; Cell a;
        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);
        }
    }

    final boolean casBase(long cmp, long val) {
        return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
    }
}

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

變數說明：

• as 表示cells引用
• b 表示獲取的base值
• v 表示 期望值,
• m 表示 cells 陣列的長度
• a 表示當前執行緒命中的cell單元格

條件分析：

條件一：as == null || (m = as.length - 1) < 0
此條件成立說明cells陣列未初始化。如果不成立則說明cells陣列已經完成初始化，對應的執行緒需要找到Cell陣列中的元素去寫值。

條件一.png

條件二：(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()方法。

條件二/條件三.png

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

java.util.concurrent.atomic.Striped64.:

final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) {
    int h;
    if ((h = getProbe()) == 0) {
        ThreadLocalRandom.current();
        h = getProbe();
        wasUncontended = true;
    }
    boolean collide = false;
    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) {
                    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;
            }
            else if (!wasUncontended)
                wasUncontended = true;
            else if (a.cas(v = a.value, ((fn == null) ? v + x : fn.applyAsLong(v, x))))
                break;
            else if (n >= NCPU || cells != as)
                collide = false;
            else if (!collide)
                collide = true;
            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;
            }
            h = advanceProbe(h);
        }
        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;
        }
        else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x))))
            break;                          
    }
}

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

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

1. cells陣列沒有初始化
2. cells陣列已經初始化，但是當前執行緒對應的cell資料為空
3. cells陣列已經初始化， 當前執行緒對應的cell資料為空，且CAS操作+1失敗

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值

接著開始正式解析longAccumulate()原始碼：

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。

可以畫圖理解為：

wasUncontended設定說明.png

接著執行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的實現細節程式碼較多，放到最後面講解。

CASE2執行條件和實現原理：

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

CASE2 標識cells陣列還未初始化，因為判斷cells == as，這個代表當前執行緒到了這裡獲取的cells還是之前的一致。我們可以先看這個case，最後再回頭看最為麻煩的CASE1實現邏輯。

cellsBusy上面說了是加鎖的狀態，初始化cells陣列和擴容的時候都要獲取加鎖的狀態，這個是通過CAS來實現的，為0代表無鎖狀態，為1代表其他執行緒已經持有鎖了。cells==as代表當前執行緒持有的陣列未進行修改過，casCellsBusy()通過CAS操作去獲取鎖。但是裡面的if條件又再次判斷了cell==as，這一點是不是很奇怪？通過畫圖來說明下問題：

cells==as雙重判斷說明.png

如果上面條件都執行成功就會執行陣列的初始化及賦值操作， 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.1.png

繼續往下看程式碼，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成功則直接跳出迴圈。

CASE1.3.png

接著看CASE1.4：

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陣列元素沒有太大作用。

多執行緒更新Cell.png

接著看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()所有的邏輯，邏輯分支挺多，仔細分析看看其實還是挺清晰的，流程圖如下：

流程圖.png

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

多執行緒執行示例.png

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

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的原理及實現都很簡單，但其設計的思想值得我們品味和學習。

相關推薦