基於JDK1.8的ConcurrentHashMap分析

之前看過ConcurrentHashMap的分析，感覺也了解的七七八八了。但昨晚接到了面試，讓我把所知道的ConcurrentHashMap全部說出來。

然后我結結巴巴，然后應該毫無意外的話就G了，今天下定決心好好分析一下，這個萬能的並發包，ConcurrentHashMap

分一下幾個方面分析ConcurrentHashMap：

• put方法
• remove方法
• get方法

(一)put方法

public V put(K key, V value) {
    return putVal(key, value, false);
}

調用了putVal方法，傳入三個參數。第一個為key，第二個為val，第三個為onlyIfAbsent(意思為: 如果為true，當插入的key相同時，不替換val值，默認是為false，替換最新的val值)

putVal方法比較多，我們分兩個部分講：

//第一部分
final V putVal(K key, V value, boolean onlyIfAbsent) {
    //對傳入的參數進行合法性判斷
    if (key == null || value == null) throw new NullPointerException();
    //計算鍵所對應的 hash 值
    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();
        //根據鍵的 hash 值找到哈希數組相應的索引位置
        //如果為空，那么以CAS無鎖式向該位置添加一個節點
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   
        }

我們看到第四行，如果key和val都為null，直接拋出異常，所以不能傳入key和val都不能為null。

第二個注意點是 散列這個函數

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

我在HashMap里面有介紹，我就不詳細說了，反正很重要。

第三個注意點就是初始化table這個函數initTable

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    //如果表為空才進行初始化操作
    while ((tab = table) == null || tab.length == 0) {
        //sizeCtl 小於零說明已經有線程正在進行初始化操作
        //當前線程應該放棄 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) {
                    //sc 大於零說明容量已經初始化了，否則使用默認容量
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    //根據容量構建數組
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    //計算閾值，等效於 n*0.75
                    sc = n - (n >>> 2);
                }
            } finally {
                //設置閾值
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

我們看到第7，8行。如果有線程在初始化，那么那就等待，讓出cpu。

初始化只允許一個線程對表進行初始化，如果不巧有其他線程進來了，那么會讓其他線程交出 CPU 等待下次系統調度。這樣，保證了表同時只會被一個線程初始化。

默認初始化為大小為16，闋值為0.75

我們重新回到putVal這個方法中去。

//檢測到桶結點是 ForwardingNode 類型，協助擴容
else if ((fh = f.hash) == MOVED)
     tab = helpTransfer(tab, f);
//桶結點是普通的結點，鎖住該桶頭結點並試圖在該鏈表的尾部添加一個節點
else {
       V oldVal = null;
       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;
                      }
                 }
           }
           //向紅黑樹中添加元素，TreeBin 結點的hash值為TREEBIN（-2）
           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;
                }
           }
       }
   }
  //binCount != 0 說明向鏈表或者紅黑樹中添加或修改一個節點成功
  //binCount  == 0 說明 put 操作將一個新節點添加成為某個桶的首節點
  if (binCount != 0) {
         //鏈表深度超過 8 轉換為紅黑樹
         if (binCount >= TREEIFY_THRESHOLD)
             treeifyBin(tab, i);
         //oldVal != null 說明此次操作是修改操作
         //直接返回舊值即可，無需做下面的擴容邊界檢查
         if (oldVal != null)
             return oldVal;
           break;
        }
    }
}
//CAS 式更新baseCount，並判斷是否需要擴容
addCount(1L, binCount);
//程序走到這一步說明此次 put 操作是一個添加操作，否則早就 return 返回了
return null;

我們看到第三行，如果f的hash是MOVED，那么就幫助他擴容(說明至少有一個線程在擴容)

這個方法說實話我看的不太懂，我就在網上查了點資料：

首先，每個線程進來會先領取自己的任務區間，然后開始 --i 來遍歷自己的任務區間，對每個桶進行處理。如果遇到桶的頭結點是空的，那么使用 ForwardingNode 標識該桶已經被處理完成了。如果遇到已經處理完成的桶，直接跳過進行下一個桶的處理。如果是正常的桶，對桶首節點加鎖，正常的遷移即可，遷移結束后依然會將原表的該位置標識位已經處理。

當 i < 0，說明本線程處理速度夠快的，整張表的最后一部分已經被它處理完了，現在需要看看是否還有其他線程在自己的區間段還在遷移中。

putVal后面的代碼就比較清楚了。如果是鏈表，就找到尾節點，插入即可。如果是紅黑樹，童謠插入即可。

至此，對於 put 方法的源碼分析已經完全結束了，感覺真的很復雜。至此我感覺我都沒有完全理解每一行代碼是什么意思。

(二)remove方法

public V remove(Object key) {
    return replaceNode(key, null, null);
}

三個參數，第一個為key，第二個為val，刪除直接置為null，讓gc來回收。第三個是Object cv，含義還不是很清楚。先繼續看吧

我們還是分為兩部分：

//第一部分
final V replaceNode(Object key, V value, Object cv) {
    //先找到key的位置 hash散列
    int hash = spread(key.hashCode());
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        //如果表為空，直接返回null
        if (tab == null || (n = tab.length) == 0 ||
            (f = tabAt(tab, i = (n - 1) & hash)) == null)
            break;
        //如果有在擴容的線程，幫助他擴容
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);

首先遍歷整張表的桶結點，如果表還未初始化或者無法根據參數的 hash 值定位到桶結點，那么將返回 null。

如果定位到的桶結點類型是 ForwardingNode 結點，調用 helpTransfer 協助擴容。

else {
    V oldVal = null;
    boolean validated = false;
    synchronized (f) {
        if (tabAt(tab, i) == f) {
            if (fh >= 0) {
                validated = true;
                for (Node<K,V> e = f, pred = null;;) {
                    K ek;
                    if (e.hash == hash &&
                        ((ek = e.key) == key ||
                         (ek != null && key.equals(ek)))) {
                        V ev = e.val;
                        if (cv == null || cv == ev ||
                            (ev != null && cv.equals(ev))) {
                            oldVal = ev;
                            if (value != null)
                                e.val = value;
                            else if (pred != null)
                                pred.next = e.next;
                            else
                                setTabAt(tab, i, e.next);
                        }
                        break;
                    }
                    pred = e;
                    if ((e = e.next) == null)
                        break;
                }
            }
            else if (f instanceof TreeBin) {
                validated = true;
                TreeBin<K,V> t = (TreeBin<K,V>)f;
                TreeNode<K,V> r, p;
                if ((r = t.root) != null &&
                    (p = r.findTreeNode(hash, key, null)) != null) {
                    V pv = p.val;
                    if (cv == null || cv == pv ||
                        (pv != null && cv.equals(pv))) {
                        oldVal = pv;
                        if (value != null)
                            p.val = value;
                        else if (t.removeTreeNode(p))
                            setTabAt(tab, i, untreeify(t.first));
                    }
                }
            }
        }
    }
    if (validated) {
        if (oldVal != null) {
            if (value == null)
                addCount(-1L, -1);
            return oldVal;
        }
        break;
    }
}

代碼很多，但我覺得思路不難。就是找到那個桶以后，直接加鎖。判斷是鏈表還是樹，然后刪除即可。

(三)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());
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        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;
}

和HashMap的get方法大同小異。沒有涉及到並發操作。直接取到key的hash值，如果是第一個節點，直接返回。否則while循環查找