LongAdder是jdk8新增的用於併發環境的計數器，目的是為了在高併發情況下，代替AtomicLong/AtomicInt，成為一個用於高併發情況下的高效的通用計數器。

高併發下計數，一般最先想到的應該是AtomicLong/AtomicInt，AtmoicXXX使用硬體級別的指令 CAS 來更新計數器的值，這樣可以避免加鎖，機器直接支援的指令，效率也很高。但是AtomicXXX中的 CAS 操作在出現執行緒競爭時，失敗的執行緒會白白地迴圈一次，在併發很大的情況下，因為每次CAS都只有一個執行緒能成功，競爭失敗的執行緒會非常多。失敗次數越多，迴圈次數就越多，很多執行緒的CAS操作越來越接近 自旋鎖（spin lock）。計數操作本來是一個很簡單的操作，實際需要耗費的cpu時間應該是越少越好，AtomicXXX在高併發計數時，大量的cpu時間都浪費會在 自旋 上了，這很浪費，也降低了實際的計數效率。

// jdk1.8的AtomicLong的實現程式碼，這段程式碼在sun.misc.Unsafe中
// 當執行緒競爭很激烈時，while判斷條件中的CAS會連續多次返回false，這樣就會造成無用的迴圈，迴圈中讀取volatile變數的開銷本來就是比較高的
// 因為這樣，在高併發時，AtomicXXX並不是那麼理想的計數方式
public final long getAndAddLong(Object o, long offset, long delta) {
    long v;
    do {
        v = getLongVolatile(o, offset);
    } while (!compareAndSwapLong(o, offset, v, v + delta));
    return v;
}
 

說LongAdder比在高併發時比AtomicLong更高效，這麼說有什麼依據呢？LongAdder是根據ConcurrentHashMap這類為併發設計的類的基本原理——鎖分段，來實現的，它裡面維護一組按需分配的計數單元，併發計數時，不同的執行緒可以在不同的計數單元上進行計數，這樣減少了執行緒競爭，提高了併發效率。本質上是用空間換時間的思想，不過在實際高併發情況中消耗的空間可以忽略不計。

現在，在處理高併發計數時，應該優先使用LongAdder，而不是繼續使用AtomicLong。當然，執行緒競爭很低的情況下進行計數，使用Atomic還是更簡單更直接，並且效率稍微高一些。

其他情況，比如序號生成，這種情況下需要準確的數值，全域性唯一的AtomicLong才是正確的選擇，此時不應該使用LongAdder。

下面簡要分析下LongAdder的原始碼，有了ConcurrentHashMap（LongAdder比較像1.6和1.7的，可以看下1.7的）的基礎，這個類的原始碼看起來也不復雜。

一、類的關係

公共父類Striped64是實現中的核心，它實現一些核心操作，處理64位資料，很容易就能轉化為其他基本型別，是個通用的類。二元算術運算累積，指的是你可以給它提供一個二元算術方式，這個類按照你提供的方式進行算術計算，並儲存計算結果。二元運算中第一個運算元是累積器中某個計數單元當前的值，另外一個值是外部提供的。

舉幾個例子：

假設每次操作都需要把原來的數值加上某個值，那麼二元運算為 (x, y) -> x+y，這樣累積器每次都會加上你提供的數字y，這跟LongAdder的功能基本上是一樣的；

假設每次操作都需要把原來的數值變為它的某個倍數，那麼可以指定二元運算為 (x, y) -> x*y，累積器每次都會乘以你提供的數字y，y=2時就是通常所說的每次都翻一倍；

假設每次操作都需要把原來的數值變成它的5倍，再加上3，再除以2，再減去4，再乘以你給定的數，最後還要加上6，那麼二元運算為 (x, y) -> ((x*5+3)/2 - 4)*y +6，累積器每次累積操作都會按照你說的做；

......

LongAccumulator是標準的實現類，LongAdder是特化的實現類，它的功能等價於LongAccumulator((x, y) -> x+y, 0L)。它們的區別很簡單，前者可以進行任何二元算術操作，後者只能進行加減兩種算術操作。

Double版本是Long版本的簡單改裝，相對Long版本，主要的變化就是用Double.longBitsToDouble 和Double.doubleToRawLongBits對底層的8位元組資料進行long <---> double轉換，儲存的時候使用long型，計算的時候轉化為double型。這是因為CAS是sun.misc.Unsafe中提供的操作，只對int、long、物件型別（引用或者指標）提供了這種操作，其他型別都需要轉化為這三種類型才能進行CAS操作。這裡的long型也可以認為是8位元組的原始型別，因為把它視為long型別是無意義的。java中沒有C語言中的 void* 無型別（或者叫原始型別），只能用最接近的long型別來代替。

四個實現類的區別就上面這兩句話，這裡只講LongAdder一個類。

二、核心實現Striped64

四個類的核心實現都在Striped64中，這個類使用分段的思想，來儘量平攤併發壓力。類似1.7及以前版本的ConcurrentHashMap.Segment，Striped64中使用了一個叫Cell的類，是一個普通的二元算術累積單元，執行緒也是通過hash取模操作對映到一個Cell上進行累積。為了加快取模運算效率，也把Cell陣列的大小設定為2^n，同時大量使用Unsafe提供的底層操作。基本的實現桶1.7的ConcurrentHashMap非常像，而且更簡單。

1、累積單元Cell

看到這裡我想了一個看似簡單的問題：既然Cell這麼簡單，只有一個long型變數，為什麼不直接用long value？

首先宣告下，Unsafe提供的操作很強大，也能對陣列的元素進行volatile讀寫，同時陣列計算某個元素的offset偏移量本身就很簡單，因此volatile、cas這種站不住腳。這個問題下面一點再進行解答。

// 很簡單的一個類，這個類可以看成是一個簡化的AtomicLong
// 通過cas操作來更新value的值
// @sun.misc.Contended是一個高階的註解，代表使用快取行填來避免偽共享，可以自己網上搜下，這個我就不細說了
@sun.misc.Contended static final class Cell {
    volatile long value;
    Cell(long x) { value = x; }
    final boolean cas(long cmp, long val) {
        return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
    }
 
    // Unsafe mechanics Unsafe相關的初始化
    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);
        }
    }
}

2、Striped64主體程式碼

abstract class Striped64 extends Number {
    @sun.misc.Contended static final class Cell { ... }
 
    /** Number of CPUS, to place bound on table size */
    static final int NCPU = Runtime.getRuntime().availableProcessors();
 
    // cell陣列，長度一樣要是2^n，可以類比為jdk1.7的ConcurrentHashMap中的segments陣列
    transient volatile Cell[] cells;
 
    // 累積器的基本值，在兩種情況下會使用：
    // 1、沒有遇到併發的情況，直接使用base，速度更快；
    // 2、多執行緒併發初始化table陣列時，必須要保證table陣列只被初始化一次，因此只有一個執行緒能夠競爭成功，這種情況下競爭失敗的執行緒會嘗試在base上進行一次累積操作
    transient volatile long base;
 
    // 自旋標識，在對cells進行初始化，或者後續擴容時，需要通過CAS操作把此標識設定為1（busy，忙標識，相當於加鎖），取消busy時可以直接使用cellsBusy = 0，相當於釋放鎖
    transient volatile int cellsBusy;
 
    Striped64() {
    }
 
    // 使用CAS更新base的值
    final boolean casBase(long cmp, long val) {
        return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
    }
 
    // 使用CAS將cells自旋標識更新為1
    // 更新為0時可以不用CAS，直接使用cellsBusy就行
    final boolean casCellsBusy() {
        return UNSAFE.compareAndSwapInt(this, CELLSBUSY, 0, 1);
    }
 
    // 下面這兩個方法是ThreadLocalRandom中的方法，不過因為包訪問關係，這裡又重新寫一遍
 
    // probe翻譯過來是探測/探測器/探針這些，不好理解，它是ThreadLocalRandom裡面的一個屬性，
    // 不過並不影響對Striped64的理解，這裡可以把它理解為執行緒本身的hash值
    static final int getProbe() {
        return UNSAFE.getInt(Thread.currentThread(), PROBE);
    }
 
    // 相當於rehash，重新算一遍執行緒的hash值
    static final int advanceProbe(int probe) {
        probe ^= probe << 13;   // xorshift
        probe ^= probe >>> 17;
        probe ^= probe << 5;
        UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
        return probe;
    }
 
    /**
     * 核心方法的實現，此方法建議在外部進行一次CAS操作（cell != null時嘗試CAS更新base值，cells != null時，CAS更新hash值取模後對應的cell.value）
     * @param x the value 前面我說的二元運算中的第二個運算元，也就是外部提供的那個運算元
     * @param fn the update function, or null for add (this convention avoids the need for an extra field or function in LongAdder).
     *     外部提供的二元算術操作，例項持有並且只能有一個，生命週期內保持不變，null代表LongAdder這種特殊但是最常用的情況，可以減少一次方法呼叫
     * @param wasUncontended false if CAS failed before call 如果為false，表明呼叫者預先呼叫的一次CAS操作都失敗了
     */
    final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) {
        int h;
        // 這個if相當於給執行緒生成一個非0的hash值
        if ((h = getProbe()) == 0) {
            ThreadLocalRandom.current(); // force initialization
            h = getProbe();
            wasUncontended = true;
        }
        boolean collide = false; // True if last slot nonempty 如果hash取模對映得到的Cell單元不是null，則為true，此值也可以看作是擴容意向，感覺這個更好理解
        for (;;) {
            Cell[] as; Cell a; int n; long v;
            if ((as = cells) != null && (n = as.length) > 0) { // cells已經被初始化了
                if ((a = as[(n - 1) & h]) == null) { // hash取模對映得到的Cell單元還為null（為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; // 對沒有使用的Cell單元進行累積操作（第一次賦值相當於是累積上一個運算元，求和時再和base執行一次運算就得到實際的結果）
                                    created = true;
                                }
                            } finally {
                                cellsBusy = 0; 清空自旋標識，釋放鎖
                            }
                            if (created) // 如果原本為null的Cell單元是由自己進行第一次累積操作，那麼任務已經完成了，所以可以退出迴圈
                                break;
                            continue;           // Slot is now non-empty 不是自己進行第一次累積操作，重頭再來
                        }
                    }
                    collide = false; // 執行這一句是因為cells被加鎖了，不能往下繼續執行第一次的賦值操作（第一次累積），所以還不能考慮擴容
                }
                else if (!wasUncontended) // CAS already known to fail 前面一次CAS更新a.value（進行一次累積）的嘗試已經失敗了，說明已經發生了執行緒競爭
                    wasUncontended = true; // Continue after rehash 情況失敗標識，後面去重新算一遍執行緒的hash值
                else if (a.cas(v = a.value, ((fn == null) ? v + x : fn.applyAsLong(v, x)))) // 嘗試CAS更新a.value（進行一次累積） ------ 標記為分支A
                    break; // 成功了就完成了累積任務，退出迴圈
                else if (n >= NCPU || cells != as) // cell陣列已經是最大的了，或者中途發生了擴容操作。因為NCPU不一定是2^n，所以這裡用 >=
                    collide = false; // At max size or stale 長度n是遞增的，執行到了這個分支，說明n >= NCPU會永遠為true，下面兩個else if就永遠不會被執行了，也就永遠不會再進行擴容
                                     // CPU能夠並行的CAS操作的最大數量是它的核心數（CAS在x86中對應的指令是cmpxchg，多核需要通過鎖快取來保證整體原子性），當n >= NCPU時，再出現幾個執行緒對映到同一個Cell導致CAS競爭的情況，那就真不關擴容的事了，完全是hash值的鍋了
                else if (!collide) // 對映到的Cell單元不是null，並且嘗試對它進行累積時，CAS競爭失敗了，這時候把擴容意向設定為true
                                   // 下一次迴圈如果還是跟這一次一樣，說明競爭很嚴重，那麼就真正擴容
                    collide = true; // 把擴容意向設定為true，只有這裡才會給collide賦值為true，也只有執行了這一句，才可能執行後面一個else if進行擴容
                else if (cellsBusy == 0 && casCellsBusy()) { // 最後再考慮擴容，能到這一步說明競爭很激烈，嘗試加鎖進行擴容 ------ 標記為分支B
                    try {
                        if (cells == as) {      // Expand table unless stale 檢查下是否被別的執行緒擴容了（CAS更新鎖標識，處理不了ABA問題，這裡再檢查一遍）
                            Cell[] rs = new Cell[n << 1]; // 執行2倍擴容
                            for (int i = 0; i < n; ++i)
                                rs[i] = as[i];
                            cells = rs;
                        }
                    } finally {
                        cellsBusy = 0; // 釋放鎖
                    }
                    collide = false; // 擴容意向為false
                    continue; // Retry with expanded table 擴容後重頭再來
                }
                h = advanceProbe(h); // 重新給執行緒生成一個hash值，降低hash衝突，減少對映到同一個Cell導致CAS競爭的情況
            }
            else if (cellsBusy == 0 && cells == as && casCellsBusy()) { // cells沒有被加鎖，並且它沒有被初始化，那麼就嘗試對它進行加鎖，加鎖成功進入這個else if
                boolean init = false;
                try {                           // Initialize table
                    if (cells == as) { // CAS避免不了ABA問題，這裡再檢測一次，如果還是null，或者空陣列，那麼就執行初始化
                        Cell[] rs = new Cell[2]; // 初始化時只建立兩個單元
                        rs[h & 1] = new Cell(x); // 對其中一個單元進行累積操作，另一個不管，繼續為null
                        cells = rs;
                        init = true;
                    }
                } finally {
                    cellsBusy = 0; // 清空自旋標識，釋放鎖
                }
                if (init) // 如果某個原本為null的Cell單元是由自己進行第一次累積操作，那麼任務已經完成了，所以可以退出迴圈
                    break;
            }
            else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x)))) // cells正在進行初始化時，嘗試直接在base上進行累加操作
                break;                          // Fall back on using base 直接在base上進行累積操作成功了，任務完成，可以退出迴圈了
        }
    }
 
    // double的不講，更long的邏輯基本上是一樣的
    final void doubleAccumulate(double x, DoubleBinaryOperator fn, boolean wasUncontended);
 
    // Unsafe mechanics Unsafe初始化
    private static final sun.misc.Unsafe UNSAFE;
    private static final long BASE;
    private static final long CELLSBUSY;
    private static final long PROBE;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> sk = Striped64.class;
            BASE = UNSAFE.objectFieldOffset
                (sk.getDeclaredField("base"));
            CELLSBUSY = UNSAFE.objectFieldOffset
                (sk.getDeclaredField("cellsBusy"));
            Class<?> tk = Thread.class;
            PROBE = UNSAFE.objectFieldOffset
                (tk.getDeclaredField("threadLocalRandomProbe"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }
 
}

看完這個在來看看第一點中我提的問題：既然Cell這麼簡單，為什麼不直接用long value？

先看看我特別標明的兩個分支：分支A是用CAS更新對應的cell.value，是個寫操作，分支B是進行擴容。

ConcurrentHashMap中，擴容和寫操作是會嚴格處理的，在一個分段鎖管轄區內，不會出現擴容和寫操作併發：1.6和1.7的擴容操作都是在put內部執行的，put本身就會加鎖，因此擴容進行時會阻塞對同一個Segment的寫操作；1.8中擴容時，put/remove等方法如果碰見正在其他執行緒正在執行擴容，會去幫助擴容，擴容完成了之後才會去嘗試加鎖執行真正的寫操作。

雖然B分支會進行”加鎖“，但是A操作跟cellsBusy無關，”加鎖“並不禁止A操作的執行。AB兩個分支是不互斥的， 因此Striped64這裡會出現A分支的寫操作，和B分支擴容操作併發執行的情況。

那麼問題是：為什麼這麼併發執行沒問題？

仔細看看A操作，就明白了。A操作使用CAS更新Cell物件中的某個屬性，並不改變陣列持有的Cell物件的引用，擴容操作進行的是陣列持有的Cell物件引用的複製，複製後引用指向的還是原來的那個Cell物件。

舉個例子就是，舊的cell陣列，叫作old，old[1] = cellA，cellA.value = 1，擴容後的新陣列，叫作new，任然有new[1] = cellA。A分支實際上執行的是cellA.value = 2，無論分支A和B怎麼併發執行，執行完成後新陣列都能看到分支A對Cell的改變，擴容前後實際上陣列持有的是同一群Cell物件。

這下就知道為什麼不直接用long變數代替Cell物件了吧。long[]進行復制時，兩個陣列完完全全分離了，A分支直接作用在舊陣列上，B分支擴容後，看不到串行復制執行後對舊陣列同一位置的改變。舉個例子就是，old[1]=10，A分支要把old[1]更新為11，這時候B分支已經複製到old[5]了，A分支執行完成後，B分支建立的新陣列new[1]可能還是10（不管是多少，反正沒記錄A分支的操作），這樣A分支的操作就被遺失了，程式會有問題。

下面簡單畫了個示意圖，可以看看。

三、LongAdder

看完了Striped64的講解，這部分就很簡單了，只是一些簡單的封裝。

public class LongAdder extends Striped64 implements Serializable {
 
    // 構造方法，什麼也不做，直接使用預設值，base = 0, cells = null
    public LongAdder() {
    }
 
    // add方法，根據父類的longAccumulate方法的要求，這裡要進行一次CAS操作
    // （雖然這裡有兩個CAS，但是第一個CAS成功了就不會執行第二個，要執行第二個，第一個就被“短路”了不會被執行）
    // 線上程競爭不激烈時，這樣做更快
    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);
        }
    }
 
    public void increment() {
        add(1L);
    }
 
    public void decrement() {
        add(-1L);
    }
 
    // 返回累加的和，也就是“當前時刻”的計數值
    // 此返回值可能不是絕對準確的，因為呼叫這個方法時還有其他執行緒可能正在進行計數累加，
    //     方法的返回時刻和呼叫時刻不是同一個點，在有併發的情況下，這個值只是近似準確的計數值
    // 高併發時，除非全域性加鎖，否則得不到程式執行中某個時刻絕對準確的值，但是全域性加鎖在高併發情況下是下下策
    // 在很多的併發場景中，計數操作並不是核心，這種情況下允許計數器的值出現一點偏差，此時可以使用LongAdder
    // 在必須依賴準確計數值的場景中，應該自己處理而不是使用通用的類
    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;
    }
 
    // 重置計數器，只應該在明確沒有併發的情況下呼叫，可以用來避免重新new一個LongAdder
    public void reset() {
        Cell[] as = cells; Cell a;
        base = 0L;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    a.value = 0L;
            }
        }
    }
 
    // 相當於sum()後再呼叫reset()
    public long sumThenReset() {
        Cell[] as = cells; Cell a;
        long sum = base;
        base = 0L;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null) {
                    sum += a.value;
                    a.value = 0L;
                }
            }
        }
        return sum;
    }
 
    // 其他的不說了
}

簡單總結下：

這個類是jdk1.8新增的類，目的是為了提供一個通用的，更高效的用於併發場景的計數器。

jdk1.8的ConcurrentHashMap中，沒有再使用Segment，使用了一個簡單的仿造LongAdder實現的計數器，這樣能夠保證計數效率不低於使用Segment的效率。