1、LongAdder由來(lái)

LongAdder類是JDK1.8新增的一個(gè)原子性操作類。AtomicLong通過(guò)CAS算法提供了非阻塞的原子性操作，相比受用阻塞算法的同步器來(lái)說(shuō)性能已經(jīng)很好了，但是JDK開發(fā)組并不滿足于此，因?yàn)榻?jīng)常搞并發(fā)的請(qǐng)求下AtomicLong的性能是不能讓人接受的。

如下AtomicLong 的incrementAndGet的代碼，雖然AtomicLong使用CAS算法，但是CAS失敗后還是通過(guò)無(wú)限循環(huán)的自旋鎖不多的嘗試，這就是高并發(fā)下CAS性能低下的原因所在。源碼如下：

public final long incrementAndGet() {
        for (;;) {
            long current = get();
            long next = current + 1;
            if (compareAndSet(current, next))
                return next;
        }
    }

高并發(fā)下N多線程同時(shí)去操作一個(gè)變量會(huì)造成大量線程CAS失敗，然后處于自旋狀態(tài)，導(dǎo)致嚴(yán)重浪費(fèi)CPU資源，降低了并發(fā)性。

2、LongAdder與AtomicLong的簡(jiǎn)單介紹

我們知道，volatile關(guān)鍵字是輕量級(jí)鎖，可以解決多線程內(nèi)存不可見問(wèn)題。對(duì)于一寫多讀，可以解決變量同步問(wèn)題，但是如果是多寫，volatile無(wú)法解決線程安全問(wèn)題的。例如，count++操作，就應(yīng)該使用如下方式： AtomicInteger count = new AtomicInteger(); 、count.addAndGet(1);而如果是JDK8及以上，推薦使用LongAdder對(duì)象替代，因?yàn)樗男阅鼙?code>AtomicLong 更好（減少樂(lè)觀鎖的重試次數(shù)）。

LongAdder其他應(yīng)用場(chǎng)景：

對(duì)于Java項(xiàng)目中計(jì)數(shù)統(tǒng)計(jì)的一些需求，如果是 JDK8，推薦使用 LongAdder 對(duì)象，比 AtomicLong 性能更好（減少樂(lè)觀鎖的重試次數(shù)）

在大多數(shù)項(xiàng)目及開源組件中，計(jì)數(shù)統(tǒng)計(jì)使用最多的仍然還是AtomicLong，雖然是阿里巴巴這樣說(shuō)，但是我們?nèi)匀灰鶕?jù)使用場(chǎng)景來(lái)決定是否使用LongAdder。

今天主要是來(lái)講講LongAdder的實(shí)現(xiàn)原理，還是老方式，通過(guò)圖文一步步解開LongAdder神秘的面紗，通過(guò)此篇文章你會(huì)了解到：

• 為什么AtomicLong在高并發(fā)場(chǎng)景下性能急劇下降？
• LongAdder為什么快？
• LongAdder實(shí)現(xiàn)原理（圖文分析）
• AtomicLong是否可以被遺棄或替換？

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

3、AtomicLong

當(dāng)我們?cè)谶M(jìn)行計(jì)數(shù)統(tǒng)計(jì)的時(shí)，通常會(huì)使用AtomicLong來(lái)實(shí)現(xiàn)。AtomicLong能保證并發(fā)情況下計(jì)數(shù)的準(zhǔn)確性，其內(nèi)部通過(guò)CAS來(lái)解決并發(fā)安全性的問(wèn)題。

3.1 AtomicLong實(shí)現(xiàn)原理

說(shuō)到線程安全的計(jì)數(shù)統(tǒng)計(jì)工具類，肯定少不了Atomic下的幾個(gè)原子類。AtomicLong就是juc包下重要的原子類，在并發(fā)情況下可以對(duì)長(zhǎng)整形類型數(shù)據(jù)進(jìn)行原子操作，保證并發(fā)情況下數(shù)據(jù)的安全性。

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

我們?cè)谟?jì)數(shù)的過(guò)程中，一般使用incrementAndGet()和decrementAndGet()進(jìn)行加一和減一操作，這里調(diào)用了Unsafe類中的getAndAddLong()方法進(jìn)行操作。

接著看看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);
}

這里直接進(jìn)行CAS+自旋操作更新AtomicLong中的value值，進(jìn)而保證value值的原子性更新。

3.2 AtomicLong瓶頸分析

如上代碼所示，我們?cè)谑褂肅AS + 自旋的過(guò)程中，在高并發(fā)環(huán)境下，N個(gè)線程同時(shí)進(jìn)行自旋操作，會(huì)出現(xiàn)大量失敗并不斷自旋的情況，此時(shí)AtomicLong的自旋會(huì)成為瓶頸。

如上圖所示，高并發(fā)場(chǎng)景下AtomicLong性能會(huì)急劇下降，我們后面也會(huì)舉例說(shuō)明。

那么高并發(fā)下計(jì)數(shù)的需求有沒(méi)有更好的替代方案呢？在JDK8 中 Doug Lea大神新寫了一個(gè)LongAdder來(lái)解決此問(wèn)題，我們后面來(lái)看LongAdder是如何優(yōu)化的。

4、LongAdder

4.1 LongAdder和AtomicLong性能測(cè)試

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

/**
 * Atomic和LongAdder耗時(shí)測(cè)試
 */
 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 耗時(shí)：" + (System.currentTimeMillis() - start) + "ms");
        System.out.println("threadCount: " + threadCount + ", times: " + times);
        long atomicStart = System.currentTimeMillis();
        testAtomicLong(threadCount, times);
        System.out.println("AtomicLong 耗時(shí)：" + (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());
    }
}

執(zhí)行結(jié)果：

這里可以看到隨著并發(fā)的增加，AtomicLong性能是急劇下降的，耗時(shí)是LongAdder的數(shù)倍。至于原因我們還是接著往后看。

4.2 LongAdder為什么這么快

先看下LongAdder的操作原理圖：

既然說(shuō)到LongAdder可以顯著提升高并發(fā)環(huán)境下的性能，那么它是如何做到的？

1、 設(shè)計(jì)思想上，LongAdder采用"分段"的方式降低CAS失敗的頻次

這里先簡(jiǎn)單的說(shuō)下LongAdder的思路，后面還會(huì)詳述LongAdder的原理。

我們知道，AtomicLong中有個(gè)內(nèi)部變量value保存著實(shí)際的long值，所有的操作都是針對(duì)該變量進(jìn)行。也就是說(shuō)，高并發(fā)環(huán)境下，value變量其實(shí)是一個(gè)熱點(diǎn)數(shù)據(jù)，也就是N個(gè)線程競(jìng)爭(zhēng)一個(gè)熱點(diǎn)。

LongAdder的基本思路就是分散熱點(diǎn)，將value值的新增操作分散到一個(gè)數(shù)組中，不同線程會(huì)命中到數(shù)組的不同槽中，各個(gè)線程只對(duì)自己槽中的那個(gè)value值進(jìn)行CAS操作，這樣熱點(diǎn)就被分散了，沖突的概率就小很多。

LongAdder有一個(gè)全局變量volatile long base值，當(dāng)并發(fā)不高的情況下都是通過(guò)CAS來(lái)直接操作base值，如果CAS失敗，則針對(duì)LongAdder中的Cell[]數(shù)組中的Cell進(jìn)行CAS操作，減少失敗的概率。

例如當(dāng)前類中base = 10，有三個(gè)線程進(jìn)行CAS原子性的**+1操作**，線程一執(zhí)行成功，此時(shí)base=11，線程二、線程三執(zhí)行失敗后開始針對(duì)于Cell[]數(shù)組中的Cell元素進(jìn)行**+1操作**，同樣也是CAS操作，此時(shí)數(shù)組index=1和index=2中Cell的value都被設(shè)置為了1.

執(zhí)行完成后，統(tǒng)計(jì)累加數(shù)據(jù)：sum = 11 + 1 + 1 = 13，利用LongAdder進(jìn)行累加的操作就執(zhí)行完了，流程圖如下：

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

2、使用Contended注解來(lái)消除偽共享

在 LongAdder 的父類 Striped64 中存在一個(gè) volatile Cell[] cells; 數(shù)組，其長(zhǎng)度是2 的冪次方，每個(gè)Cell都使用 @Contended 注解進(jìn)行修飾，而@Contended注解可以進(jìn)行緩存行填充，從而解決偽共享問(wèn)題。偽共享會(huì)導(dǎo)致緩存行失效，緩存一致性開銷變大。

@sun.misc.Contended static final class Cell {
}

偽共享指的是多個(gè)線程同時(shí)讀寫同一個(gè)緩存行的不同變量時(shí)導(dǎo)致的 CPU緩存失效。盡管這些變量之間沒(méi)有任何關(guān)系，但由于在主內(nèi)存中鄰近，存在于同一個(gè)緩存行之中，它們的相互覆蓋會(huì)導(dǎo)致頻繁的緩存未命中，引發(fā)性能下降。這里對(duì)于偽共享我只是提一下概念，并不會(huì)深入去講解，大家可以自行查閱一些資料。

解決偽共享的方法一般都是使用直接填充，我們只需要保證不同線程的變量存在于不同的 CacheLine 即可，使用多余的字節(jié)來(lái)填充可以做點(diǎn)這一點(diǎn)，這樣就不會(huì)出現(xiàn)偽共享問(wèn)題。例如在Disruptor隊(duì)列的設(shè)計(jì)中就有類似設(shè)計(jì)。

在Striped64類中我們可以看看Doug Lea在Cell上加的注釋也有說(shuō)明這一點(diǎn)：

框中的翻譯如下：

Cell類是AtomicLong添加了padded（via@sun.misc.compended)來(lái)消除偽共享的變種版本。緩存行填充對(duì)于大多數(shù)原子來(lái)說(shuō)是繁瑣的，因?yàn)樗鼈兺ǔ２灰?guī)則地分散在內(nèi)存中，因此彼此之間不會(huì)有太大的干擾。但是，駐留在數(shù)組中的原子對(duì)象往往彼此相鄰，因此在沒(méi)有這種預(yù)防措施的情況下，通常會(huì)共享緩存行數(shù)據(jù)（對(duì)性能有巨大的負(fù)面影響）。

3、惰性求值

LongAdder只有在使用longValue()獲取當(dāng)前累加值時(shí)才會(huì)真正的去結(jié)算計(jì)數(shù)的數(shù)據(jù)，longValue()方法底層就是調(diào)用sum()方法，對(duì)base和Cell數(shù)組的數(shù)據(jù)累加然后返回，做到數(shù)據(jù)寫入和讀取分離。

而AtomicLong使用incrementAndGet()每次都會(huì)返回long類型的計(jì)數(shù)值，每次遞增后還會(huì)伴隨著數(shù)據(jù)返回，增加了額外的開銷。

4.3 LongAdder實(shí)現(xiàn)原理

之前說(shuō)了，AtomicLong是多個(gè)線程針對(duì)單個(gè)熱點(diǎn)值value進(jìn)行原子操作。而LongAdder是每個(gè)線程擁有自己的槽，各個(gè)線程一般只對(duì)自己槽中的那個(gè)值進(jìn)行CAS操作。

比如有三個(gè)線程同時(shí)對(duì)value增加1，那么value = 1 + 1 + 1 = 3

但是對(duì)于LongAdder來(lái)說(shuō)，內(nèi)部有一個(gè)base變量，一個(gè)Cell[]數(shù)組。
base變量：非競(jìng)爭(zhēng)條件下，直接累加到該變量上
Cell[]數(shù)組：競(jìng)爭(zhēng)條件下，累加個(gè)各個(gè)線程自己的槽Cell[i]中
最終結(jié)果的計(jì)算是下面這個(gè)形式：

4.4 ongAdder源碼剖析

前面已經(jīng)用圖分析了LongAdder高性能的原理，我們繼續(xù)看下LongAdder實(shí)現(xiàn)的源碼：

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

一般我們進(jìn)行計(jì)數(shù)時(shí)都會(huì)使用increment()方法，每次進(jìn)行**+1操作**，increment()會(huì)直接調(diào)用add()方法。

變量說(shuō)明：

• as 表示cells引用
• b 表示獲取的base值
• v 表示 期望值,
• m 表示 cells 數(shù)組的長(zhǎng)度
• a 表示當(dāng)前線程命中的cell單元格

條件分析：

條件一：as == null || (m = as.length - 1) < 0
此條件成立說(shuō)明cells數(shù)組未初始化。如果不成立則說(shuō)明cells數(shù)組已經(jīng)完成初始化，對(duì)應(yīng)的線程需要找到Cell數(shù)組中的元素去寫值。

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

getProbe()獲取當(dāng)前線程的hash值，m表示cells長(zhǎng)度-1，cells長(zhǎng)度是2的冪次方數(shù)，原因之前也講到過(guò)，與數(shù)組長(zhǎng)度取?？梢赞D(zhuǎn)化為按位與運(yùn)算，提升計(jì)算性能。

當(dāng)條件成立時(shí)說(shuō)明當(dāng)前線程通過(guò)hash計(jì)算出來(lái)數(shù)組位置處的cell為空，進(jìn)一步去執(zhí)行longAccumulate()方法。如果不成立則說(shuō)明對(duì)應(yīng)的cell不為空，下一步將要將x值通過(guò)CAS操作添加到cell中。

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

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

接著往下看核心的longAccumulate()方法，代碼很長(zhǎng)，后面會(huì)一步步分析，先上代碼：

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

代碼很長(zhǎng)，if else分支很多，除此看肯定會(huì)很頭疼。這里一點(diǎn)點(diǎn)分析，然后結(jié)合畫圖一步步了解其中實(shí)現(xiàn)原理。

我們首先要清楚執(zhí)行這個(gè)方法的前置條件，它們是或的關(guān)系，如上面條件一、二、三

• cells數(shù)組沒(méi)有初始化
• cells數(shù)組已經(jīng)初始化，但是當(dāng)前線程對(duì)應(yīng)的cell數(shù)據(jù)為空
• cells數(shù)組已經(jīng)初始化， 當(dāng)前線程對(duì)應(yīng)的cell數(shù)據(jù)為空，且CAS操作+1失敗

longAccumulate()方法的入?yún)ⅲ?/strong>

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

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 ((as = cells) != null && (n = as.length) > 0) {
}

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

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

else if (!wasUncontended)
    wasUncontended = true;
h = advanceProbe(h);

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

break;

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

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

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