一、 LongAdder簡(jiǎn)介

LongAdder類(lèi)是JDK1.8新增的一個(gè)原子性操作類(lèi)。上一節(jié)說(shuō)到，AtomicLong通過(guò)CAS提供了非阻塞的原子性操作，相比用阻塞算法的synchronized來(lái)說(shuō)性能已經(jīng)得到了很大提升。在高并發(fā)下大量線程會(huì)同時(shí)競(jìng)爭(zhēng)更新同一個(gè)原子變量，但由于只有一個(gè)線程的CAS操作會(huì)成功，這就造成了大量線程競(jìng)爭(zhēng)失敗后，會(huì)通過(guò)無(wú)限循環(huán)不斷進(jìn)行自旋嘗試CAS操作，這會(huì)白白浪費(fèi)CPU資源。

為了解決AtomicLong在高并發(fā)下的缺點(diǎn)，LongAdder應(yīng)運(yùn)而生。LongAdder采用的思路是：既然AtomicLong由于過(guò)多線程同時(shí)去競(jìng)爭(zhēng)同一個(gè)變量的更新而產(chǎn)生性能瓶頸，那么把一個(gè)變量分解為多個(gè)變量，讓同樣多的線程去競(jìng)爭(zhēng)多個(gè)資源，就解決了性能問(wèn)題。

如圖4-1：

使用AtomicLong時(shí)，是多個(gè)線程同時(shí)競(jìng)爭(zhēng)同一個(gè)原子變量。

如圖4-2：

使用LongAdder時(shí)，則是內(nèi)部維護(hù)多個(gè)Cell變量，每個(gè)Cell里面有一個(gè)初始值為0的long型變量，在同等并發(fā)量的情況下，爭(zhēng)奪單個(gè)變量的線程會(huì)減少，這是變相地減少了爭(zhēng)奪共享資源的并發(fā)量。另外，多個(gè)線程在爭(zhēng)奪同一個(gè)Cell原子變量時(shí)候，如果失敗并不是自旋CAS重試，而是嘗試獲取在其他Cell原子變量上進(jìn)行CAS嘗試，這增加了當(dāng)前線程重試CAS成功的可能性。最后，在獲取LongAdder當(dāng)前值時(shí)，是把所有Cell變量的value值累加后再加上base值返回的。

LongAdder維護(hù)了一個(gè)Cells數(shù)組和一個(gè)基值變量base，Cells數(shù)組的特點(diǎn)如下：

• 延遲初始化（默認(rèn)情況下Cells為null）。因?yàn)镃ells占用內(nèi)存相對(duì)較大，故惰性加載。Cells為null時(shí)且并發(fā)線程較少時(shí)，所有的累加操作都是對(duì)base變量進(jìn)行的。
• 初始化時(shí)，Cells數(shù)組中Cell元素個(gè)數(shù)為2；同時(shí)，Cell數(shù)組中元素個(gè)數(shù)保存為2的N次方。

Cell 類(lèi)是AtomicLong的一個(gè)改進(jìn)，用來(lái)減少緩存的爭(zhēng)用，即解決偽共享問(wèn)題。對(duì)于大多數(shù)孤立的多個(gè)原子操作進(jìn)行字節(jié)填充是浪費(fèi)的，因?yàn)樵硬僮鞫际菬o(wú)規(guī)律地分散在內(nèi)存中進(jìn)行的，多個(gè)原子變量被放入同一個(gè)緩存行的可能性很小。但是原子性數(shù)組元素的內(nèi)存地址是連續(xù)的，故數(shù)組內(nèi)的多個(gè)元素能經(jīng)常共享緩存行（偽共享），因此Cell類(lèi)使用了@sun.misc.Contended注解進(jìn)行字節(jié)填充，這是為了防止數(shù)組中多個(gè)元素共享一個(gè)緩存行，從而提升性能。

二、LongAdder代碼分析

為了解決高并發(fā)下多線程對(duì)一個(gè)變量 CAS 爭(zhēng)奪失敗后進(jìn)行無(wú)限自旋而造成的降低并發(fā)性能的問(wèn)題，LongAdder在內(nèi)部維護(hù)了一個(gè)動(dòng)態(tài)的Cell數(shù)組來(lái)分擔(dān)對(duì)單個(gè)變量進(jìn)行爭(zhēng)奪的開(kāi)銷(xiāo)。

這一節(jié)我們來(lái)圍繞以下話題對(duì)LongAdder的實(shí)現(xiàn)進(jìn)行分析：

• （1）LongAdder的結(jié)構(gòu)
• （2）當(dāng)前線程應(yīng)該訪問(wèn)Cells數(shù)組里的哪一個(gè)Cell元素
• （3）如何初始化Cells數(shù)組
• （4）Cells數(shù)組的擴(kuò)容機(jī)制
• （5）線程訪問(wèn)所分配的Cell元素有沖突后如何處理
• （6）如何保證線程操作被分配的Cell元素的原子性

（1）LongAdder的結(jié)構(gòu)

如圖，LongAdder類(lèi)繼承自Striped64類(lèi)：

先混個(gè)眼熟，Striped64類(lèi)是一個(gè)高并發(fā)累加的工具類(lèi)。其特點(diǎn)為：

• 設(shè)計(jì)核心思路就是通過(guò)內(nèi)部的分散計(jì)算來(lái)避免競(jìng)爭(zhēng)。
• 內(nèi)部包含一個(gè)base和一個(gè)Cells數(shù)組
• 沒(méi)有競(jìng)爭(zhēng)的情況下，要累加的數(shù)通過(guò)CAS累加到base上；如果有競(jìng)爭(zhēng)的話，會(huì)將Cells數(shù)組中的每個(gè)Cell的value值累加，再加上base值返回。

Striped64類(lèi)內(nèi)部維護(hù)三個(gè)重要的變量：

• transient volatile Cell[] cells; // 存放Cell的數(shù)組，大小為2的N次方
• transient volatile long base; // 基礎(chǔ)值，默認(rèn)為0
• transient volatile int cellsBusy; // 用來(lái)實(shí)現(xiàn)自旋鎖，狀態(tài)值只有0和1，通過(guò)CAS操作該變量來(lái)保證只有一個(gè)線程可以創(chuàng)建Cell元素、初始化Cells數(shù)組、對(duì)Cells數(shù)組擴(kuò)容

上文中出現(xiàn)頻率很高的Cell長(zhǎng)啥樣？下面我們來(lái)揭秘一下。

Cell結(jié)構(gòu)如下：

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

簡(jiǎn)單分析一下：

• 為提高性能，使用注解@sun.misc.Contended修飾，用來(lái)避免偽共享。
• value變量被聲明為volatile，為了保證變量的內(nèi)存可見(jiàn)性。
• cas方法通過(guò)CAS操作，保證了當(dāng)前線程更新時(shí)被分配的Cell元素中value值的原子性。

（2）add方法實(shí)現(xiàn)

從add方法的代碼中，我們可以找到開(kāi)頭里很多問(wèn)題的答案。

  	public void add(long x) {
        Cell[] as; long b, v; int m; Cell a;
        if ((as = cells) != null || !casBase(b = base, b + x)) {//(1)
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 || //(2)
                (a = as[getProbe() & m]) == null ||//(3)
                !(uncontended = a.cas(v = a.value, v + x)))//(4)
                longAccumulate(x, null, uncontended);//(5)
        }
    }
	final boolean casBase(long cmp, long val) {
        return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
    }

• (1)處代碼表明：如果cells為null，則add方法實(shí)現(xiàn)的效果是在base變量上累加x，這是與AtomicLong的操作類(lèi)似。
• 當(dāng)cells不為null或者線程執(zhí)行代碼(1)處的CAS操作失敗了，就會(huì)執(zhí)行代碼(2)；代碼(2)(3)決定了當(dāng)前線程應(yīng)該訪問(wèn)Cells數(shù)組里哪一個(gè)Cell元素，如果當(dāng)前線程映射的Cell元素存在則執(zhí)行代碼(4)，使用CAS操作去更新Cell元素的value值
• 如果當(dāng)前線程映射的Cell元素不存在或者存在，但是CAS操作失敗則執(zhí)行代碼(5)。

總結(jié)來(lái)看就是，(2)(3)(4)處的代碼就是獲取當(dāng)前線程應(yīng)該訪問(wèn)的Cells數(shù)組中的Cell元素，然后進(jìn)行CAS更新操作，在獲取期間如果有條件不滿足就會(huì)跳到代碼(5)執(zhí)行。

另外，當(dāng)前線程應(yīng)該訪問(wèn)Cells數(shù)組的哪一個(gè)Cell元素是通過(guò)getProbe() & m 進(jìn)行計(jì)算的，其中m是當(dāng)前Cells數(shù)組元素個(gè)數(shù)-1，getProbe()則用于獲取當(dāng)前線程中變量threadLocalRandomProbe的值（默認(rèn)為0，代碼(5)將其初始化），并且當(dāng)前線程通過(guò)分配的Cell元素的cas函數(shù)來(lái)保證對(duì)Cell元素更新的原子性。到此，我們解決了問(wèn)題當(dāng)前線程應(yīng)該訪問(wèn)Cells數(shù)組里的哪一個(gè)Cell元素和如何保證線程操作被分配的Cell元素的原子性。

（3）add方法中l(wèi)ongAccumulate方法的實(shí)現(xiàn)

longAccumulate方法是cells數(shù)組被初始化和擴(kuò)容的地方。

	final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
        int h;//(6)
        if ((h = getProbe()) == 0) {
            ThreadLocalRandom.current(); // force initialization
            h = getProbe();
            wasUncontended = true;
        }
        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) {//(7)
                if ((a = as[(n - 1) & h]) == null) {//(8)
                    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
				//(9)               
                else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                             fn.applyAsLong(v, x))))
                    break;
                //(10)
                else if (n >= NCPU || cells != as)
                    collide = false;            // At max size or stale
                //(11)
                else if (!collide)
                    collide = true;
                //(12)
                else if (cellsBusy == 0 && casCellsBusy()) {
                    try {
                        if (cells == as) {      // Expand table unless stale
                            //(12.1)
                            Cell[] rs = new Cell[n << 1];
                            for (int i = 0; i < n; ++i)
                                rs[i] = as[i];
                            cells = rs;
                        }
                    } finally {
                    	//(12.2)
                        cellsBusy = 0;
                    }
                    //(12.3)
                    collide = false;
                    continue;                   // Retry with expanded table
                }
                //(13)為了能找到一個(gè)空閑的Cell，重新計(jì)算hash值，xorshift算法生成隨機(jī)數(shù)。
                h = advanceProbe(h);
            }
            //(14)
            else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
                boolean init = false;
                try {                           // Initialize table
                    if (cells == as) {
                    	//(14.1)
                        Cell[] rs = new Cell[2];
                        //(14.2)
                        rs[h & 1] = new Cell(x);
                        cells = rs;
                        init = true;
                    }
                } finally {
                	//(14.3)
                    cellsBusy = 0;
                }
                if (init)
                    break;
            }
            else if (casBase(v = base, ((fn == null) ? v + x :
                                        fn.applyAsLong(v, x))))
                break;                          // Fall back on using base
        }
    }

• 當(dāng)線程第一次執(zhí)行代碼(6)時(shí)，會(huì)初始化當(dāng)前線程變量threadLocalRandomProbe的值，此變量在計(jì)算當(dāng)前線程被分配到Cells數(shù)組的哪一個(gè)Cell元素時(shí)會(huì)使用。
• 當(dāng)前線程執(zhí)行到代碼(7)(8)時(shí)，會(huì)根據(jù)當(dāng)前線程的threadLocalRandomProbe和cells元素個(gè)數(shù)計(jì)算訪問(wèn)的Cell元素下標(biāo)，如果發(fā)現(xiàn)對(duì)應(yīng)下標(biāo)元素的值為null，則新增一個(gè)Cell元素添加到Cells數(shù)組中。但在添加的時(shí)候，需要競(jìng)爭(zhēng)并設(shè)置cellsBusy為1。
• 當(dāng)前線程對(duì)應(yīng)的Cell元素存在，執(zhí)行代碼(9)，進(jìn)行累加操作。

如果Cells數(shù)組不存在時(shí)，會(huì)發(fā)生什么？

• cells數(shù)組初始化是在代碼(14)中進(jìn)行的，cellsBusy是一個(gè)標(biāo)識(shí)（0：當(dāng)前數(shù)組沒(méi)有在被初始化或者擴(kuò)容，也未新建Cell元素。1：cells數(shù)組在被初始化或擴(kuò)容、正在創(chuàng)建新的cell元素），通過(guò)CAS操作來(lái)進(jìn)行0或1狀態(tài)的切換，使用casCellsBusy方法。
• (14.1)初始化cells數(shù)組，長(zhǎng)度為2，然后(14.2)使用當(dāng)前線程的threadLocalRandomProbe值&(cells元素個(gè)數(shù)-1)來(lái)計(jì)算當(dāng)前線程應(yīng)該訪問(wèn)cells數(shù)組的哪個(gè)位置。最后(14.3)重置cellsBusy為0，表示初始化完成。這里雖未使用CAS，但卻線程安全，因?yàn)閏ellsBusy是volatile的，保證了內(nèi)存可見(jiàn)性。
• 初始化完的cells數(shù)組，里面的元素還是為null的。創(chuàng)建Cell元素是在(7)(8)中。

Cells數(shù)組擴(kuò)容是怎么實(shí)現(xiàn)的？

• 代碼(12)中，擴(kuò)容之前需要進(jìn)行(10)和(11)的判斷，當(dāng)cells元素個(gè)數(shù)小于當(dāng)前機(jī)器CPU個(gè)數(shù)且當(dāng)前多個(gè)線程訪問(wèn)了cells中同一個(gè)元素時(shí)，從而導(dǎo)致沖突使其中一個(gè)線程CAS失敗時(shí)才會(huì)進(jìn)行擴(kuò)容操作。
• 涉及CPU個(gè)數(shù)的原因是，只有每個(gè)CPU都運(yùn)行一個(gè)線程時(shí)，多線程效果最佳。即當(dāng)Cells元素個(gè)數(shù)等于CPU個(gè)數(shù)時(shí)，每個(gè)Cell都使用一個(gè)CPU進(jìn)行處理，效果最佳。
• 代碼(12)中，擴(kuò)容操作是先通過(guò)CAS設(shè)置cellsBusy為1，然后才擴(kuò)容。假設(shè)CAS成功則執(zhí)行代碼(12.1)擴(kuò)容，容量為之前的2倍，并復(fù)制Cell元素到擴(kuò)容后數(shù)組，并復(fù)制Cell元素到擴(kuò)容后的數(shù)組。

線程訪問(wèn)所分配的Cell元素有沖突后如何處理?

• 代碼(13)已經(jīng)給了我們答案，對(duì)CAS失敗的線程重新計(jì)算當(dāng)前線程的threadLocalRandomProbe值，以減少下次訪問(wèn)cells元素時(shí)的沖突機(jī)會(huì)。

三、總結(jié)

最后再問(wèn)一下自己，看完了源碼分析后，能回答出這六個(gè)問(wèn)題嗎？如果能，那么恭喜你，LongAdder的原理已經(jīng)掌握得差不多了。

• （1）LongAdder的結(jié)構(gòu)
• （2）當(dāng)前線程應(yīng)該訪問(wèn)Cells數(shù)組里的哪一個(gè)Cell元素
• （3）如何初始化Cells數(shù)組
• （4）Cells數(shù)組的擴(kuò)容機(jī)制
• （5）線程訪問(wèn)所分配的Cell元素有沖突后如何處理
• （6）如何保證線程操作被分配的Cell元素的原子性

由于篇幅有限，下一期我們?cè)賮?lái)看看LongAccumulator類(lèi)的原理，畢竟這期的內(nèi)容有點(diǎn)多了，講太多也不好消化。這期就到這吧~更多關(guān)于java高并發(fā)AtomicLong LongAdder的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論