ConcurrentHashMap 1.8原理解析

• 2019 年 11 月 27 日
• 筆記

JDK 1.8中，Hash家族有這麼一些存在，HashMap,HashTable,LinkedHashMap,ConcurrentHashMap。這裡面支援執行緒安全的有HashTable以及ConcurrentHashMap。對Hash有一個基本了解可以參考本人的從Hash到一致性Hash原理(深度好文) 。

那既然說到ConcurrentHashMap，自然要討論的就是它的執行緒安全性和效能。我們先來看一下HashTable的執行緒安全性以及效能的低下。

public synchronized V put(K key, V value) {      // Make sure the value is not null      if (value == null) {          throw new NullPointerException();      }        // Makes sure the key is not already in the hashtable.      Entry<?,?> tab[] = table;      int hash = key.hashCode();      int index = (hash & 0x7FFFFFFF) % tab.length;      @SuppressWarnings("unchecked")      Entry<K,V> entry = (Entry<K,V>)tab[index];      for(; entry != null ; entry = entry.next) {          if ((entry.hash == hash) && entry.key.equals(key)) {              V old = entry.value;              entry.value = value;              return old;          }      }        addEntry(hash, key, value, index);      return null;  }

public synchronized V get(Object key) {      Entry<?,?> tab[] = table;      int hash = key.hashCode();      int index = (hash & 0x7FFFFFFF) % tab.length;      for (Entry<?,?> e = tab[index] ; e != null ; e = e.next) {          if ((e.hash == hash) && e.key.equals(key)) {              return (V)e.value;          }      }      return null;  }

我們對比一下HashMap 1.7的這兩個方法(因為1.8點HashMap會生成紅黑樹，我們暫時先不考慮紅黑樹的問題)

public V put(K key, V value) {          if (key == null)              return putForNullKey(value);          int hash = hash(key);          int i = indexFor(hash, table.length);          for (Entry<K,V> e = table[i]; e != null; e = e.next) {              Object k;              if (e.hash == hash && ((k = e.key) == key || key.equals(k))) {                  V oldValue = e.value;                  e.value = value;                  e.recordAccess(this);                  return oldValue;              }          }            modCount++;          addEntry(hash, key, value, i);          return null;      }

public V get(Object key) {          if (key == null)              return getForNullKey();          Entry<K,V> entry = getEntry(key);            return null == entry ? null : entry.getValue();      }

final int hash(Object k) {          int h = 0;          if (useAltHashing) {              if (k instanceof String) {                  return sun.misc.Hashing.stringHash32((String) k);              }              h = hashSeed;          }            h ^= k.hashCode();            // This function ensures that hashCodes that differ only by          // constant multiples at each bit position have a bounded          // number of collisions (approximately 8 at default load factor).          //一種演算法，進行4次位移，得到相對比較分散的鏈表          h ^= (h >>> 20) ^ (h >>> 12);          return h ^ (h >>> 7) ^ (h >>> 4);      }

在這裡我們可以看到，他們除了key的hash演算法不同，HashTable的比較簡單，只是用key的哈希值與0x7FFFFFFF(十進位2147483647，二進位1111111111111111111111111111111)進行一次與運算，即為只要相同二進位位上不論0，1全部都變成1，再對table數組的長度取模。而HashMap 1.7則為key的哈希值進行各位無符號位移加異或運算，取得最終哈希值。然後是HashTable不接收null的key,而HashMap接受。他們最大的區別就在於synchronized顯示器鎖了。

synchronized的本質是所有對象的位元組碼中有一個monitor的對象頭，任何執行緒拿到了這個monitor的對象頭就可以對這個對象進行操作，而拿不到monitor對象頭的執行緒就只能等待，直到拿到了monitor的執行緒放棄，其他執行緒才能爭奪這個對象頭來對對象進行操作。那麼問題來了，當大量執行緒高並發的時候，只要有一個執行緒拿到了這個對象頭，其他執行緒對這個對象是既不能讀也不能寫。而對於HashMap來說，如果多執行緒對其進行操作，那麼任意執行緒都可以胡亂修改裡面值的內容造成臟讀，所以HashMap是執行緒不安全的。

那麼我們今天的主角登場了ConcurrentHashMap。我們同樣來看一下這兩個方法。以下是1.8的源碼，首先我們要清楚的是1.8跟1.7已經完全不同，1.7是使用segements(16個segement)，每個segement都有一個table(Map.Entry數組)，相當於16個HashMap，同步機製為分段鎖，每個segment繼承ReentrantLock；而1.8隻有1個table(Map.Entry數組)，同步機製為CAS + synchronized保證並發更新。如果不搞清楚這個問題，那麼你看1.8的源碼可能會很懵逼。

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

final V putVal(K key, V value, boolean onlyIfAbsent) {      //無論key還是value,不允許空      if (key == null || value == null) throw new NullPointerException();      //此處獲取hash值的方法與HashTable類似      int hash = spread(key.hashCode());      int binCount = 0;      //無限循環      for (Node<K,V>[] tab = table;;) {          Node<K,V> f; int n, i, fh;          //如果節點數組為null，或者長度為0，初始化節點數組          if (tab == null || (n = tab.length) == 0)              tab = initTable();          //如果節點數組的某個節點為null，則put的時候就會採用無鎖競爭來獲取該節點的頭把交椅          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);          else { //如果hash衝突的時候，即多執行緒操作時，大家都有一樣的hash值              V oldVal = null;              synchronized (f) { //鎖定節點數組的該節點                  if (tabAt(tab, i) == f) {                      //如果當前該節點為鏈表形態                      if (fh >= 0) {                          binCount = 1;                          for (Node<K,V> e = f;; ++binCount) {                              K ek;                              //找鏈表中找到相同的key，把新value替代老value                              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;                              //如果找不到key，就添加到鏈表到末尾                              if ((e = e.next) == null) {                                  pred.next = new Node<K,V>(hash, key,                                                            value, null);                                  break;                              }                          }                      }                      //如果當前為紅黑樹形態，進行紅黑樹到查找和替代(存在相同的key)，或者放入紅黑樹到新葉節點上(key不存在)                      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) {                  //如果鏈表長度超過了8，鏈錶轉紅黑樹                  if (binCount >= TREEIFY_THRESHOLD)                      treeifyBin(tab, i);                  if (oldVal != null)                      return oldVal;                  break;              }          }      }      //統計節點個數，檢查是否需要擴容      addCount(1L, binCount);      return null;  }

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

讀取的時候，我們沒有看見鎖到存在，說明讀不受多執行緒影響。

對比ConcurrentHashMap和HashTable,我們可以明顯的看到，ConcurrentHashMap在寫的時候，並沒有鎖住整個節點數組，在新節點上使用的是無鎖競爭，在老節點上鎖住的僅僅是一個節點，讀的時候如果不是恰好讀到寫執行緒寫入相同Hash值的位置，不受影響(可以認為我們的操作一般是讀多寫少，這種幾率也比較低)。而HashTable是對整個節點數組進行鎖定，讀到時候不能寫，寫的時候不能讀，這麼一對比就可以明顯感覺到性能差距是巨大的。

雖然ConcurrentHashMap的並發性能還算比較優異，但在億級計算中，卻依然會成為性能瓶頸，具體可以參考本人的Fork/Join框架原理和使用探秘

至於這裡為什麼會慢，我認為在這種超高並發下，節點數組的單節點的的寫寫競爭是互斥的，其次，由於紅黑樹具有讀快寫慢的特性，它要不斷保持樹的平衡而不斷返轉，所以才會使得高並發寫的性能急劇下降。