深入淺出ConcurrentHashMap1.8

簡書 占小狼
轉(zhuǎn)載請注明原創(chuàng)出處，謝謝！

JDK1.8分析

1.8的實現(xiàn)已經(jīng)拋棄了Segment分段鎖機制，利用CAS+Synchronized來保證并發(fā)更新的安全，底層采用數(shù)組+鏈表+紅黑樹的存儲結(jié)構(gòu)。

重要概念

在開始之前，有些重要的概念需要介紹一下：

  • table：默認為null，初始化發(fā)生在第一次插入操作，默認大小為16的數(shù)組，用來存儲Node節(jié)點數(shù)據(jù)，擴容時大小總是2的冪次方。
  • nextTable：默認為null，擴容時新生成的數(shù)組，其大小為原數(shù)組的兩倍。
  • sizeCtl ：默認為0，用來控制table的初始化和擴容操作，具體應用在后續(xù)會體現(xiàn)出來。
    • -1 代表table正在初始化
    • -N 表示有N-1個線程正在進行擴容操作

其余情況：
1、如果table未初始化，表示table需要初始化的大小。
2、2、如果table初始化完成，表示table的容量，默認是table大小的0.75倍，居然用這個公式算0.75（n - (n >>> 2)）。

  • Node：保存key，value及key的hash值的數(shù)據(jù)結(jié)構(gòu)。
class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;
    ... 省略部分代碼
}

其中value和next都用volatile修飾，保證并發(fā)的可見性。

ForwardingNode：一個特殊的Node節(jié)點，hash值為-1，其中存儲nextTable的引用。

final class ForwardingNode<K,V> extends Node<K,V> {
    final Node<K,V>[] nextTable;
    ForwardingNode(Node<K,V>[] tab) {
        super(MOVED, null, null, null);
        this.nextTable = tab;
    }
}

只有table發(fā)生擴容的時候，F(xiàn)orwardingNode才會發(fā)揮作用，作為一個占位符放在table中表示當前節(jié)點為null或則已經(jīng)被移動。

實例初始化

實例化ConcurrentHashMap時帶參數(shù)時，會根據(jù)參數(shù)調(diào)整table的大小，假設(shè)參數(shù)為100，最終會調(diào)整成256，確保table的大小總是2的冪次方，算法如下：

ConcurrentHashMap<String, String> hashMap = new ConcurrentHashMap<>(100);
private static final int tableSizeFor(int c) {
    int n = c - 1;
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}

注意，ConcurrentHashMap在構(gòu)造函數(shù)中只會初始化sizeCtl值，并不會直接初始化table，而是延緩到第一次put操作。

table初始化

前面已經(jīng)提到過，table初始化操作會延緩到第一次put行為。但是put是可以并發(fā)執(zhí)行的，Doug Lea是如何實現(xiàn)table只初始化一次的？讓我們來看看源碼的實現(xiàn)。

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
//如果一個線程發(fā)現(xiàn)sizeCtl<0，意味著另外的線程執(zhí)行CAS操作成功，當前線程只需要讓出cpu時間片
        if ((sc = sizeCtl) < 0) 
            Thread.yield(); // lost initialization race; just spin
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

sizeCtl默認為0，如果ConcurrentHashMap實例化時有傳參數(shù)，sizeCtl會是一個2的冪次方的值。所以執(zhí)行第一次put操作的線程會執(zhí)行Unsafe.compareAndSwapInt方法修改sizeCtl為-1，有且只有一個線程能夠修改成功，其它線程通過Thread.yield()讓出CPU時間片等待table初始化完成。

put操作

假設(shè)table已經(jīng)初始化完成，put操作采用CAS+synchronized實現(xiàn)并發(fā)插入或更新操作，具體實現(xiàn)如下。

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        ...省略部分代碼
    }
    addCount(1L, binCount);
    return null;
}

1、hash算法

static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;}

2、table中定位索引位置，n是table的大小

int index = (n - 1) & hash

3、獲取table中對應索引的元素f

Doug Lea采用Unsafe.getObjectVolatile來獲取，也許有人質(zhì)疑，直接table[index]不可以么，為什么要這么復雜？

在java內(nèi)存模型中，我們已經(jīng)知道每個線程都有一個工作內(nèi)存，里面存儲著table的副本，雖然table是volatile修飾的，但不能保證線程每次都拿到table中的最新元素，Unsafe.getObjectVolatile可以直接獲取指定內(nèi)存的數(shù)據(jù)，保證了每次拿到數(shù)據(jù)都是最新的。

4、如果f為null，說明table中這個位置第一次插入元素，利用- Unsafe.compareAndSwapObject方法插入Node節(jié)點。

  • 如果CAS成功，說明Node節(jié)點已經(jīng)插入，隨后addCount(1L, binCount)方法會檢查當前容量是否需要進行擴容。
  • 如果CAS失敗，說明有其它線程提前插入了節(jié)點，自旋重新嘗試在這個位置插入節(jié)點。

5、如果f的hash值為-1，說明當前f是ForwardingNode節(jié)點，意味有其它線程正在擴容，則一起進行擴容操作。

6、其余情況把新的Node節(jié)點按鏈表或紅黑樹的方式插入到合適的位置，這個過程采用同步內(nèi)置鎖實現(xiàn)并發(fā)，代碼如下:

synchronized (f) {
    if (tabAt(tab, i) == f) {
        if (fh >= 0) {
            binCount = 1;
            for (Node<K,V> e = f;; ++binCount) {
                K ek;
                if (e.hash == hash &&
                    ((ek = e.key) == key ||
                     (ek != null && key.equals(ek)))) {
                    oldVal = e.val;
                    if (!onlyIfAbsent)
                        e.val = value;
                    break;
                }
                Node<K,V> pred = e;
                if ((e = e.next) == null) {
                    pred.next = new Node<K,V>(hash, key,
                                              value, null);
                    break;
                }
            }
        }
        else if (f instanceof TreeBin) {
            Node<K,V> p;
            binCount = 2;
            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                           value)) != null) {
                oldVal = p.val;
                if (!onlyIfAbsent)
                    p.val = value;
            }
        }
    }
}

在節(jié)點f上進行同步，節(jié)點插入之前，再次利用tabAt(tab, i) == f判斷，防止被其它線程修改。

  • 1、如果f.hash >= 0，說明f是鏈表結(jié)構(gòu)的頭結(jié)點，遍歷鏈表，如果找到對應的node節(jié)點，則修改value，否則在鏈表尾部加入節(jié)點。

  • 2 、如果f是TreeBin類型節(jié)點，說明f是紅黑樹根節(jié)點，則在樹結(jié)構(gòu)上遍歷元素，更新或增加節(jié)點。

  • 3、 如果鏈表中節(jié)點數(shù)binCount >= TREEIFY_THRESHOLD(默認是8)，則把鏈表轉(zhuǎn)化為紅黑樹結(jié)構(gòu)。

table擴容

當table容量不足的時候，即table的元素數(shù)量達到容量閾值sizeCtl，需要對table進行擴容。
整個擴容分為兩部分：

  • 1、構(gòu)建一個nextTable，大小為table的兩倍。
  • 2、把table的數(shù)據(jù)復制到nextTable中。

這兩個過程在單線程下實現(xiàn)很簡單，但是ConcurrentHashMap是支持并發(fā)插入的，擴容操作自然也會有并發(fā)的出現(xiàn)，這種情況下，第二步可以支持節(jié)點的并發(fā)復制，這樣性能自然提升不少，但實現(xiàn)的復雜度也上升了一個臺階。

先看第一步，構(gòu)建nextTable，毫無疑問，這個過程只能只有單個線程進行nextTable的初始化，具體實現(xiàn)如下：

private final void addCount(long x, int check) {
    ... 省略部分代碼
    if (check >= 0) {
        Node<K,V>[] tab, nt; int n, sc;
        while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
               (n = tab.length) < MAXIMUM_CAPACITY) {
            int rs = resizeStamp(n);
            if (sc < 0) {
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                    transferIndex <= 0)
                    break;
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}

通過Unsafe.compareAndSwapInt修改sizeCtl值，保證只有一個線程能夠初始化nextTable，擴容后的數(shù)組長度為原來的兩倍，但是容量是原來的1.5。

節(jié)點從table移動到nextTable，大體思想是遍歷、復制的過程。

  • 1、首先根據(jù)運算得到需要遍歷的次數(shù)i，然后利用tabAt方法獲得i位置的元素f，初始化一個forwardNode實例fwd。

  • 2、如果f == null，則在table中的i位置放入fwd，這個過程是采用Unsafe.compareAndSwapObjectf方法實現(xiàn)的，很巧妙的實現(xiàn)了節(jié)點的并發(fā)移動。

  • 3、如果f是鏈表的頭節(jié)點，就構(gòu)造一個反序鏈表，把他們分別放在nextTable的i和i+n的位置上，移動完成，采用Unsafe.putObjectVolatile方法給table原位置賦值fwd。

  • 4、如果f是TreeBin節(jié)點，也做一個反序處理，并判斷是否需要untreeify，把處理的結(jié)果分別放在nextTable的i和i+n的位置上，移動完成，同樣采用Unsafe.putObjectVolatile方法給table原位置賦值fwd。

遍歷過所有的節(jié)點以后就完成了復制工作，把table指向nextTable，并更新sizeCtl為新數(shù)組大小的0.75倍 ，擴容完成。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容