ConcurrentHashMap的JDK1.8實現

今天咱們介紹一下ConcurrentHashMap在JDK1.8中的實現。

基本結構

ConcurrentHashMap在1.8中的實現，相比於1.7的版本基本上所有都變掉了。首先，取消了Segment分段鎖的數據結構，取而代之的是數組+鏈表（紅黑樹）的結構。而對於鎖的粒度，調整爲對每一個數組元素加鎖（Node）。而後是定位節點的hash算法被簡化了，這樣帶來的弊端是Hash衝突會加重。所以在鏈表節點數量大於8時，會將鏈表轉化爲紅黑樹進行存儲。這樣一來，查詢的時間複雜度就會由原先的O(n)變爲O(logN)。下面是其基本結構：

相關屬性

sizeCtl

private transient volatile int sizeCtl;

sizeCtl用於table[]的初始化和擴容操做，不一樣值的表明狀態以下：

• -1：table[]正在初始化。
• -N：表示有N-1個線程正在進行擴容操做。

非負狀況：

• 若是table[]未初始化，則表示table須要初始化的大小。
• 若是初始化完成，則表示table[]擴容的閥值，默認是table[]容量的0.75 倍。

DEFAULT_CONCURRENCY_LEVEL

private static finalint DEFAULT_CONCURRENCY_LEVEL = 16;

• DEFAULT_CONCURRENCY_LEVEL：表示默認的併發級別，也就是table[]的默認大小。

LOAD_FACTOR

private static final float LOAD_FACTOR = 0.75f;

• LOAD_FACTOR：默認的負載因子。

TREEIFY_THRESHOLD

static final int TREEIFY_THRESHOLD = 8;

• TREEIFY_THRESHOLD：鏈表轉紅黑樹的閥值，當table[i]下面的鏈表長度大於8時就轉化爲紅黑樹結構。

UNTREEIFY_THRESHOLD

static final int UNTREEIFY_THRESHOLD = 6;

• UNTREEIFY_THRESHOLD：紅黑樹轉鏈表的閥值，當鏈表長度<=6時轉爲鏈表（擴容時）。

構造函數

public ConcurrentHashMap(int initialCapacity,  
                         float loadFactor, int concurrencyLevel) {  
    if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)  
        throw new IllegalArgumentException();  
    if (initialCapacity < concurrencyLevel)   // 初始化容量至少要爲concurrencyLevel  
        initialCapacity = concurrencyLevel;  
    long size = (long)(1.0 + (long)initialCapacity / loadFactor);  
    int cap = (size >= (long)MAXIMUM_CAPACITY) ?  
        MAXIMUM_CAPACITY : tableSizeFor((int)size);  
    this.sizeCtl = cap;  
}

從上面代碼能夠看出，在建立ConcurrentHashMap時，並無初始化table[]數組，只對Map容量，併發級別等作了賦值操做。

相關節點

• Node：該類用於構造table[]，只讀節點（不提供修改方法）。
• TreeBin：紅黑樹結構。
• TreeNode：紅黑樹節點。
• ForwardingNode：臨時節點（擴容時使用）。

put()操做

public V put(K key, V value) {  
    return putVal(key, value, false);  
}  
  
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)// 若table[]未建立，則初始化  
            tab = initTable();  
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
         
            // table[i]後面無節點時，直接建立Node（無鎖操做）  
            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)// 若是當前正在擴容，則幫助擴容並返回最新table[]  
            tab = helpTransfer(tab, f);  
        else {// 在鏈表或者紅黑樹中追加節點  
            V oldVal = null;  
            synchronized (f) {// 這裏並無使用ReentrantLock，說明synchronized已經足夠優化了  
                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)))) {// 找到key，替換value  
                                oldVal = e.val;  
                                if (!onlyIfAbsent)  
                                    e.val = value;  
                                break;  
                            }  
                            Node<K,V> pred = e;  
                            if ((e = e.next) == null) {// 在尾部插入Node  
                                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;  
                        }  
                    }  
                }  
            }  
            if (binCount != 0) {  
                if (binCount >= TREEIFY_THRESHOLD)// 到達閥值，變爲紅黑樹結構  
                    treeifyBin(tab, i);  
                if (oldVal != null)  
                    return oldVal;  
                break;  
            }  
        }  
    }  
     //將當前ConcurrentHashMap的元素數量+1  
    addCount(1L, binCount);  
    return null;  
}

從上面代碼能夠看出，put的步驟大體以下：

1. 參數校驗。
2. 若table[]未建立，則初始化。
3. 當table[i]後面無節點時，直接建立Node（無鎖操做）。
4. 若是當前正在擴容，則幫助擴容並返回最新table[]。
5. 而後在鏈表或者紅黑樹中追加節點。
6. 最後還回去判斷是否到達閥值，如到達變爲紅黑樹結構。

除了上述步驟之外，還有一點咱們留意到的是，代碼中加鎖片斷用的是synchronized關鍵字，而不是像1.7中的ReentrantLock。這一點也說明了，synchronized在新版本的JDK中優化的程度和ReentrantLock差很少了。

helpTransfer方法

這是一個協助擴容的方法。這個方法被調用的時候，當前ConcurrentHashMap必定已經有了nextTable對象，首先拿到這個nextTable對象，調用transfer方法。回看上面的transfer方法能夠看到，當本線程進入擴容方法的時候會直接進入複製階段。

/** 
    * Helps transfer if a resize is in progress. 
    */  
   final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {  
       Node<K,V>[] nextTab; int sc;  
       if (tab != null && (f instanceof ForwardingNode) &&  
           (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {  
           int rs = resizeStamp(tab.length);//計算一個操做校驗碼  
           while (nextTab == nextTable && table == tab &&  
                  (sc = sizeCtl) < 0) {  
               if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||  
                   sc == rs + MAX_RESIZERS || transferIndex <= 0)  
                   break;  
               if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {  
                   transfer(tab, nextTab);  
                   break;  
               }  
           }  
           return nextTab;  
       }  
       return table;  
   }

get()操做

public V get(Object key) {  
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;  
    int h = spread(key.hashCode());// 定位到table[]中的i  
    if ((tab = table) != null && (n = tab.length) > 0 &&  
        (e = tabAt(tab, (n - 1) & h)) != null) {// 若table[i]存在  
        if ((eh = e.hash) == h) {// 比較鏈表頭部  
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))  
                return e.val;  
        }  
        else if (eh < 0)// 若爲紅黑樹，查找樹  
            return (p = e.find(h, key)) != null ? p.val : null;  
        while ((e = e.next) != null) {// 循環鏈表查找  
            if (e.hash == h &&  
                ((ek = e.key) == key || (ek != null && key.equals(ek))))  
                return e.val;  
        }  
    }  
    return null;// 未找到  
}

get()方法的流程相對簡單一點，從上面代碼能夠看出如下步驟：

1. 首先定位到table[]中的i。
2. 若table[i]存在，則繼續查找。
3. 首先比較鏈表頭部，若是是則返回。
4. 而後若是爲紅黑樹，查找樹。
5. 最後再循環鏈表查找。

從上面步驟能夠看出，ConcurrentHashMap的get操做上面並無加鎖。因此在多線程操做的過程當中，並不能徹底的保證一致性。這裏和1.7當中相似，是弱一致性的體現。

size()操做

// 1.2時加入  
public int size() {  
    long n = sumCount();  
    return ((n < 0L) ? 0 :  
            (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :  
            (int)n);  
}  
// 1.8加入的API  
public long mappingCount() {  
    long n = sumCount();  
    return (n < 0L) ? 0L : n; // ignore transient negative values  
}  
  
final long sumCount() {  
    CounterCell[] as = counterCells; CounterCell a;  
    long sum = baseCount;  
    if (as != null) {  
        for (int i = 0; i < as.length; ++i) {  
            if ((a = as[i]) != null)  
                sum += a.value;  
        }  
    }  
    return sum;  
}