併發容器之ConcurrentHashMap(JDK 1.8版本)

時間  2019-12-01
標籤 併發 容器 concurrenthashmap jdk 1.8 版本 欄目 Java 简体版
原文   原文鏈接

1.ConcurrentHashmap簡介

在使用HashMap時在多線程狀況下擴容會出現CPU接近100%的狀況，由於hashmap並非線程安全的，一般咱們可使用在java體系中古老的hashtable類，該類基本上全部的方法都採用synchronized進行線程安全的控制，可想而知，在高併發的狀況下，每次只有一個線程可以獲取對象監視器鎖，這樣的併發性能的確不使人滿意。另一種方式經過Collections的Map<K,V> synchronizedMap(Map<K,V> m)將hashmap包裝成一個線程安全的map。好比SynchronzedMap的put方法源碼爲：java

public V put(K key, V value) {
    synchronized (mutex) {return m.put(key, value);}
}

 

實際上SynchronizedMap實現依然是採用synchronized獨佔式鎖進行線程安全的併發控制的。一樣，這種方案的性能也是使人不太滿意的。針對這種境況，Doug Lea大師竭盡全力的爲咱們創造了一些線程安全的併發容器，讓每個java開發人員倍感幸福。相對於hashmap來講，ConcurrentHashMap就是線程安全的map，其中利用了鎖分段的思想提升了併發度。node

ConcurrentHashMap在JDK1.6的版本網上資料不少，有興趣的能夠去看看。 JDK 1.6版本關鍵要素：算法

  1. segment繼承了ReentrantLock充當鎖的角色，爲每個segment提供了線程安全的保障；
  2. segment維護了哈希散列表的若干個桶，每一個桶由HashEntry構成的鏈表。

而到了JDK 1.8的ConcurrentHashMap就有了很大的變化，光是代碼量就足足增長了不少。1.8版本捨棄了segment，而且大量使用了synchronized，以及CAS無鎖操做以保證ConcurrentHashMap操做的線程安全性。數組

至於爲何不用ReentrantLock而是Synchronzied呢？實際上，synchronzied作了不少的優化，包括偏向鎖，輕量級鎖，重量級鎖，能夠依次向上升級鎖狀態，但不能降級，所以，使用synchronized相較於ReentrantLock的性能會持平甚至在某些狀況更優，具體的性能測試能夠去網上查閱一些資料。另外，底層數據結構改變爲採用數組+鏈表+紅黑樹的數據形式。安全

2.關鍵屬性及類

在瞭解ConcurrentHashMap的具體方法實現前，咱們須要系統的來看一下幾個關鍵的地方。性能優化

ConcurrentHashMap的關鍵屬性
  1. table volatile Node<K,V>[] table://裝載Node的數組，做爲ConcurrentHashMap的數據容器，採用懶加載的方式，直到第一次插入數據的時候纔會進行初始化操做，數組的大小老是爲2的冪次方。
  2. nextTable volatile Node<K,V>[] nextTable; //擴容時使用，平時爲null，只有在擴容的時候才爲非null
  3. sizeCtl volatile int sizeCtl; 該屬性用來控制table數組的大小，根據是否初始化和是否正在擴容有幾種狀況： 當值爲負數時：若是爲-1表示正在初始化，若是爲-N則表示當前正有N-1個線程進行擴容操做； 當值爲正數時：若是當前數組爲null的話表示table在初始化過程當中，sizeCtl表示爲須要新建數組的長度；

若已經初始化了，表示當前數據容器（table數組）可用容量也能夠理解成臨界值（插入節點數超過了該臨界值就須要擴容），具體指爲數組的長度n 乘以 加載因子loadFactor； 當值爲0時，即數組長度爲默認初始值。網絡

  1. sun.misc.Unsafe U 在ConcurrentHashMapde的實現中能夠看到大量的U.compareAndSwapXXXX的方法去修改ConcurrentHashMap的一些屬性。這些方法其實是利用了CAS算法保證了線程安全性，這是一種樂觀策略，假設每一次操做都不會產生衝突，當且僅當衝突發生的時候再去嘗試。

而CAS操做依賴於現代處理器指令集，經過底層CMPXCHG指令實現。CAS(V,O,N)核心思想爲：若當前變量實際值V與指望的舊值O相同，則代表該變量沒被其餘線程進行修改，所以能夠安全的將新值N賦值給變量；若當前變量實際值V與指望的舊值O不相同，則代表該變量已經被其餘線程作了處理，此時將新值N賦給變量操做就是不安全的，在進行重試。數據結構

而在大量的同步組件和併發容器的實現中使用CAS是經過sun.misc.Unsafe類實現的，該類提供了一些能夠直接操控內存和線程的底層操做，能夠理解爲java中的「指針」。該成員變量的獲取是在靜態代碼塊中：多線程

```
 static {
     try {
         U = sun.misc.Unsafe.getUnsafe();
        .......
     } catch (Exception e) {
         throw new Error(e);
     }
 }
```

 

ConcurrentHashMap中關鍵內部類
  1. Node Node類實現了Map.Entry接口，主要存放key-value對，而且具備next域
    static class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; volatile V val; volatile Node<K,V> next; ...... }

另外能夠看出不少屬性都是用volatile進行修飾的，也就是爲了保證內存可見性。併發

  1. TreeNode 樹節點，繼承於承載數據的Node類。而紅黑樹的操做是針對TreeBin類的，從該類的註釋也能夠看出，也就是TreeBin會將TreeNode進行再一次封裝
    ** * Nodes for use in TreeBins */ static final class TreeNode<K,V> extends Node<K,V> { TreeNode<K,V> parent; // red-black tree links TreeNode<K,V> left; TreeNode<K,V> right; TreeNode<K,V> prev; // needed to unlink next upon deletion boolean red; ...... }
  2. TreeBin 這個類並不負責包裝用戶的key、value信息，而是包裝的不少TreeNode節點。實際的ConcurrentHashMap「數組」中，存放的是TreeBin對象，而不是TreeNode對象。
    static final class TreeBin<K,V> extends Node<K,V> { TreeNode<K,V> root; volatile TreeNode<K,V> first; volatile Thread waiter; volatile int lockState; // values for lockState static final int WRITER = 1; // set while holding write lock static final int WAITER = 2; // set when waiting for write lock static final int READER = 4; // increment value for setting read lock ...... }
  3. ForwardingNode 在擴容時纔會出現的特殊節點，其key,value,hash所有爲null。並擁有nextTable指針引用新的table數組。
    static 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; } ..... }
CAS關鍵操做

在上面咱們說起到在ConcurrentHashMap中會大量使用CAS修改它的屬性和一些操做。所以，在理解ConcurrentHashMap的方法前咱們須要瞭解下面幾個經常使用的利用CAS算法來保障線程安全的操做。

  1. tabAt
    static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); }

該方法用來獲取table數組中索引爲i的Node元素。

  1. casTabAt
    static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); } 
    利用CAS操做設置table數組中索引爲i的元素
  2. setTabAt
    static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); } 
    該方法用來設置table數組中索引爲i的元素

3.重點方法講解

在熟悉上面的這核心信息以後，咱們接下來就來依次看看幾個經常使用的方法是怎樣實現的。

3.1 實例構造器方法

在使用ConcurrentHashMap第一件事天然而然就是new 出來一個ConcurrentHashMap對象，一共提供了以下幾個構造器方法：

// 1\. 構造一個空的map，即table數組還未初始化，初始化放在第一次插入數據時，默認大小爲16 ConcurrentHashMap() // 2\. 給定map的大小 ConcurrentHashMap(int initialCapacity) // 3\. 給定一個map ConcurrentHashMap(Map<? extends K, ? extends V> m) // 4\. 給定map的大小以及加載因子 ConcurrentHashMap(int initialCapacity, float loadFactor) // 5\. 給定map大小，加載因子以及併發度（預計同時操做數據的線程） ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)

ConcurrentHashMap一共給咱們提供了5中構造器方法，具體使用請看註釋，咱們來看看第2種構造器，傳入指定大小時的狀況，該構造器源碼爲：

public ConcurrentHashMap(int initialCapacity) {
    //1\. 小於0直接拋異常
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    //2\. 判斷是否超過了容許的最大值，超過了話則取最大值，不然再對該值進一步處理
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
               MAXIMUM_CAPACITY :
               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    //3\. 賦值給sizeCtl
    this.sizeCtl = cap;
}

 

這段代碼的邏輯請看註釋，很容易理解，若是小於0就直接拋出異常，若是指定值大於了所容許的最大值的話就取最大值，不然，在對指定值作進一步處理。最後將cap賦值給sizeCtl,關於sizeCtl的說明請看上面的說明，當調用構造器方法以後，sizeCtl的大小應該就表明了ConcurrentHashMap的大小，即table數組長度。tableSizeFor作了哪些事情了？源碼爲：

/**
 * Returns a power of two table size for the given desired capacity.
 * See Hackers Delight, sec 3.2
 */
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;
}

 

經過註釋就很清楚了，該方法會將調用構造器方法時指定的大小轉換成一個2的冪次方數，也就是說ConcurrentHashMap的大小必定是2的冪次方，好比，當指定大小爲18時，爲了知足2的冪次方特性，實際上concurrentHashMapd的大小爲2的5次方（32）。另外，須要注意的是，調用構造器方法的時候並未構造出table數組（能夠理解爲ConcurrentHashMap的數據容器），只是算出table數組的長度，當第一次向ConcurrentHashMap插入數據的時候才真正的完成初始化建立table數組的工做。

3.2 initTable方法

直接上源碼：

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        if ((sc = sizeCtl) < 0)
            // 1\. 保證只有一個線程正在進行初始化操做
            Thread.yield(); // lost initialization race; just spin
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    // 2\. 得出數組的大小
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    // 3\. 這裏才真正的初始化數組
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    // 4\. 計算數組中可用的大小：實際大小n*0.75（加載因子）
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

 

代碼的邏輯請見註釋，有可能存在一個狀況是多個線程同時走到這個方法中，爲了保證可以正確初始化，在第1步中會先經過if進行判斷，若當前已經有一個線程正在初始化即sizeCtl值變爲-1，這個時候其餘線程在If判斷爲true從而調用Thread.yield()讓出CPU時間片。正在進行初始化的線程會調用U.compareAndSwapInt方法將sizeCtl改成-1即正在初始化的狀態。

另外還須要注意的事情是，在第四步中會進一步計算數組中可用的大小即爲數組實際大小n乘以加載因子0.75.能夠看看這裏乘以0.75是怎麼算的，0.75爲四分之三，這裏n - (n >>> 2)是否是恰好是n-(1/4)n=(3/4)n，挺有意思的吧:)。若是選擇是無參的構造器的話，這裏在new Node數組的時候會使用默認大小爲DEFAULT_CAPACITY（16），而後乘以加載因子0.75爲12，也就是說數組的可用大小爲12。

3.3 put方法

使用ConcurrentHashMap最長用的也應該是put和get方法了吧，咱們先來看看put方法是怎樣實現的。調用put方法時實際具體實現是putVal方法，源碼以下：

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    //1\. 計算key的hash值
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        //2\. 若是當前table尚未初始化先調用initTable方法將tab進行初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        //3\. tab中索引爲i的位置的元素爲null，則直接使用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;                   // no lock when adding to empty bin
        }
        //4\. 當前正在擴容
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    //5\. 當前爲鏈表，在鏈表中插入新的鍵值對
                    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;
                            }
                        }
                    }
                    // 6.當前爲紅黑樹，將新的鍵值對插入到紅黑樹中
                    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;
                        }
                    }
                }
            }
            // 7.插入完鍵值對後再根據實際大小看是否須要轉換成紅黑樹
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    //8.對當前容量大小進行檢查，若是超過了臨界值（實際大小*加載因子）就須要擴容 
    addCount(1L, binCount);
    return null;
}

 

put方法的代碼量有點長，咱們按照上面的分解的步驟一步步來看。從總體而言，爲了解決線程安全的問題，ConcurrentHashMap使用了synchronzied和CAS的方式。在以前瞭解過HashMap以及1.8版本以前的ConcurrenHashMap都應該知道ConcurrentHashMap結構圖，爲了方面下面的講解這裏先直接給出，若是對這有疑問的話，能夠在網上隨便搜搜便可。

[圖片上傳中...(image-326780-1575107646328-1)]

<figcaption></figcaption>

如圖（圖片摘自網絡），ConcurrentHashMap是一個哈希桶數組，若是不出現哈希衝突的時候，每一個元素均勻的分佈在哈希桶數組中。當出現哈希衝突的時候，是標準的鏈地址的解決方式，將hash值相同的節點構成鏈表的形式，稱爲「拉鍊法」，另外，在1.8版本中爲了防止拉鍊過長，當鏈表的長度大於8的時候會將鏈表轉換成紅黑樹。table數組中的每一個元素其實是單鏈表的頭結點或者紅黑樹的根節點。當插入鍵值對時首先應該定位到要插入的桶，即插入table數組的索引i處。那麼，怎樣計算得出索引i呢？固然是根據key的hashCode值。

  1. spread()重哈希，以減少Hash衝突

 

咱們知道對於一個hash表來講，hash值分散的不夠均勻的話會大大增長哈希衝突的機率，從而影響到hash表的性能。所以經過spread方法進行了一次重hash從而大大減少哈希衝突的可能性。spread方法爲：

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

 

該方法主要是將key的hashCode的低16位於高16位進行異或運算，這樣不只可以使得hash值可以分散可以均勻減少hash衝突的機率，另外只用到了異或運算，在性能開銷上也能兼顧，作到平衡的trade-off。

2.初始化table

緊接着到第2步，會判斷當前table數組是否初始化了，沒有的話就調用initTable進行初始化，該方法在上面已經講過了。

3.可否直接將新值插入到table數組中

從上面的結構示意圖就能夠看出存在這樣一種狀況，若是插入值待插入的位置恰好所在的table數組爲null的話就能夠直接將值插入便可。那麼怎樣根據hash肯定在table中待插入的索引i呢？很顯然能夠經過hash值與數組的長度取模操做，從而肯定新值插入到數組的哪一個位置。而以前咱們提過ConcurrentHashMap的大小老是2的冪次方，(n - 1) & hash運算等價於對長度n取模，也就是hash%n，可是位運算比取模運算的效率要高不少，Doug lea大師在設計併發容器的時候也是將性能優化到了極致，使人欽佩。

肯定好數組的索引i後，就能夠能夠tabAt()方法（該方法在上面已經說明了，有疑問能夠回過頭去看看）獲取該位置上的元素，若是當前Node f爲null的話，就能夠直接用casTabAt方法將新值插入便可。

4.當前是否正在擴容

若是當前節點不爲null，且該節點爲特殊節點（forwardingNode）的話，就說明當前concurrentHashMap正在進行擴容操做，關於擴容操做，下面會做爲一個具體的方法進行講解。那麼怎樣肯定當前的這個Node是否是特殊的節點了？是經過判斷該節點的hash值是否是等於-1（MOVED）,代碼爲(fh = f.hash) == MOVED，對MOVED的解釋在源碼上也寫的很清楚了：

static final int MOVED     = -1; // hash for forwarding nodes

 

5.當table[i]爲鏈表的頭結點，在鏈表中插入新值

在table[i]不爲null而且不爲forwardingNode時，而且當前Node f的hash值大於0（fh >= 0）的話說明當前節點f爲當前桶的全部的節點組成的鏈表的頭結點。那麼接下來，要想向ConcurrentHashMap插入新值的話就是向這個鏈表插入新值。經過synchronized (f)的方式進行加鎖以實現線程安全性。往鏈表中插入節點的部分代碼爲：

if (fh >= 0) {
    binCount = 1;
    for (Node<K,V> e = f;; ++binCount) {
        K ek;
        // 找到hash值相同的key,覆蓋舊值便可
        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;
        }
    }
}

 

這部分代碼很好理解，就是兩種狀況：1. 在鏈表中若是找到了與待插入的鍵值對的key相同的節點，就直接覆蓋便可；2. 若是直到找到了鏈表的末尾都沒有找到的話，就直接將待插入的鍵值對追加到鏈表的末尾便可

6.當table[i]爲紅黑樹的根節點，在紅黑樹中插入新值

按照以前的數組+鏈表的設計方案，這裏存在一個問題，即便負載因子和Hash算法設計的再合理，也免不了會出現拉鍊過長的狀況，一旦出現拉鍊過長，甚至在極端狀況下，查找一個節點會出現時間複雜度爲O(n)的狀況，則會嚴重影響ConcurrentHashMap的性能，因而，在JDK1.8版本中，對數據結構作了進一步的優化，引入了紅黑樹。而當鏈表長度太長（默認超過8）時，鏈表就轉換爲紅黑樹，利用紅黑樹快速增刪改查的特色提升ConcurrentHashMap的性能，其中會用到紅黑樹的插入、刪除、查找等算法。當table[i]爲紅黑樹的樹節點時的操做爲：

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中經過f instanceof TreeBin判斷當前table[i]是不是樹節點，這下也正好驗證了咱們在最上面介紹時說的TreeBin會對TreeNode作進一步封裝，對紅黑樹進行操做的時候針對的是TreeBin而不是TreeNode。這段代碼很簡單，調用putTreeVal方法完成向紅黑樹插入新節點，一樣的邏輯，若是在紅黑樹中存在於待插入鍵值對的Key相同（hash值相等而且equals方法判斷爲true）的節點的話，就覆蓋舊值，不然就向紅黑樹追加新節點。

7.根據當前節點個數進行調整

當完成數據新節點插入以後，會進一步對當前鏈表大小進行調整，這部分代碼爲：

if (binCount != 0) {
    if (binCount >= TREEIFY_THRESHOLD)
        treeifyBin(tab, i);
    if (oldVal != null)
        return oldVal;
    break;
}

 

很容易理解，若是當前鏈表節點個數大於等於8（TREEIFY_THRESHOLD）的時候，就會調用treeifyBin方法將tabel[i]（第i個散列桶）拉鍊轉換成紅黑樹。

至此，關於Put方法的邏輯就基本說的差很少了，如今來作一些總結：

總體流程：

  1. 首先對於每個放入的值，首先利用spread方法對key的hashcode進行一次hash計算，由此來肯定這個值在 table中的位置；
  2. 若是當前table數組還未初始化，先將table數組進行初始化操做；
  3. 若是這個位置是null的，那麼使用CAS操做直接放入；
  4. 若是這個位置存在結點，說明發生了hash碰撞，首先判斷這個節點的類型。若是該節點fh==MOVED(表明forwardingNode,數組正在進行擴容)的話，說明正在進行擴容；
  5. 若是是鏈表節點（fh>0）,則獲得的結點就是hash值相同的節點組成的鏈表的頭節點。須要依次向後遍歷肯定這個新加入的值所在位置。若是遇到key相同的節點，則只須要覆蓋該結點的value值便可。不然依次向後遍歷，直到鏈表尾插入這個結點；
  6. 若是這個節點的類型是TreeBin的話，直接調用紅黑樹的插入方法進行插入新的節點；
  7. 插入完節點以後再次檢查鏈表長度，若是長度大於8，就把這個鏈表轉換成紅黑樹；
  8. 對當前容量大小進行檢查，若是超過了臨界值（實際大小*加載因子）就須要擴容。

3.4 get方法

看完了put方法再來看get方法就很容易了，用逆向思惟去看就好，這樣存的話我反過來這麼取就行了。get方法源碼爲：

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // 1\. 重hash
    int h = spread(key.hashCode());
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        // 2\. table[i]桶節點的key與查找的key相同，則直接返回
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        // 3\. 當前節點hash小於0說明爲樹節點，在紅黑樹中查找便可
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        while ((e = e.next) != null) {
        //4\. 從鏈表中查找，查找到則返回該節點的value，不然就返回null便可
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

 

代碼的邏輯請看註釋，首先先看當前的hash桶數組節點即table[i]是否爲查找的節點，如果則直接返回；若不是，則繼續再看當前是否是樹節點？經過看節點的hash值是否爲小於0，若是小於0則爲樹節點。若是是樹節點在紅黑樹中查找節點；若是不是樹節點，那就只剩下爲鏈表的形式的一種可能性了，就向後遍歷查找節點，若查找到則返回節點的value便可，若沒有找到就返回null。

3.5 transfer方法

當ConcurrentHashMap容量不足的時候，須要對table進行擴容。這個方法的基本思想跟HashMap是很像的，可是因爲它是支持併發擴容的，因此要複雜的多。緣由是它支持多線程進行擴容操做，而並無加鎖。我想這樣作的目的不只僅是爲了知足concurrent的要求，而是但願利用併發處理去減小擴容帶來的時間影響。transfer方法源碼爲：

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range
    //1\. 新建Node數組，容量爲以前的兩倍
    if (nextTab == null) {            // initiating
        try {
            @SuppressWarnings("unchecked")
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        nextTable = nextTab;
        transferIndex = n;
    }
    int nextn = nextTab.length;
    //2\. 新建forwardingNode引用，在以後會用到
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    boolean advance = true;
    boolean finishing = false; // to ensure sweep before committing nextTab
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // 3\. 肯定遍歷中的索引i
        while (advance) {
            int nextIndex, nextBound;
            if (--i >= bound || finishing)
                advance = false;
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                bound = nextBound;
                i = nextIndex - 1;
                advance = false;
            }
        }
        //4.將原數組中的元素複製到新數組中去
        //4.5 for循環退出，擴容結束脩改sizeCtl屬性
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            if (finishing) {
                nextTable = null;
                table = nextTab;
                sizeCtl = (n << 1) - (n >>> 1);
                return;
            }
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                    return;
                finishing = advance = true;
                i = n; // recheck before commit
            }
        }
        //4.1 當前數組中第i個元素爲null，用CAS設置成特殊節點forwardingNode(能夠理解成佔位符)
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        //4.2 若是遍歷到ForwardingNode節點  說明這個點已經被處理過了 直接跳過  這裏是控制併發擴容的核心
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        else {
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;
                    if (fh >= 0) {
                        //4.3 處理當前節點爲鏈表的頭結點的狀況，構造兩個鏈表，一個是原鏈表  另外一個是原鏈表的反序排列
                        int runBit = fh & n;
                        Node<K,V> lastRun = f;
                        for (Node<K,V> p = f.next; p != null; p = p.next) {
                            int b = p.hash & n;
                            if (b != runBit) {
                                runBit = b;
                                lastRun = p;
                            }
                        }
                        if (runBit == 0) {
                            ln = lastRun;
                            hn = null;
                        }
                        else {
                            hn = lastRun;
                            ln = null;
                        }
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
                            int ph = p.hash; K pk = p.key; V pv = p.val;
                            if ((ph & n) == 0)
                                ln = new Node<K,V>(ph, pk, pv, ln);
                            else
                                hn = new Node<K,V>(ph, pk, pv, hn);
                        }
                       //在nextTable的i位置上插入一個鏈表
                       setTabAt(nextTab, i, ln);
                       //在nextTable的i+n的位置上插入另外一個鏈表
                       setTabAt(nextTab, i + n, hn);
                       //在table的i位置上插入forwardNode節點  表示已經處理過該節點
                       setTabAt(tab, i, fwd);
                       //設置advance爲true 返回到上面的while循環中 就能夠執行i--操做
                       advance = true;
                    }
                    //4.4 處理當前節點是TreeBin時的狀況，操做和上面的相似
                    else if (f instanceof TreeBin) {
                        TreeBin<K,V> t = (TreeBin<K,V>)f;
                        TreeNode<K,V> lo = null, loTail = null;
                        TreeNode<K,V> hi = null, hiTail = null;
                        int lc = 0, hc = 0;
                        for (Node<K,V> e = t.first; e != null; e = e.next) {
                            int h = e.hash;
                            TreeNode<K,V> p = new TreeNode<K,V>
                                (h, e.key, e.val, null, null);
                            if ((h & n) == 0) {
                                if ((p.prev = loTail) == null)
                                    lo = p;
                                else
                                    loTail.next = p;
                                loTail = p;
                                ++lc;
                            }
                            else {
                                if ((p.prev = hiTail) == null)
                                    hi = p;
                                else
                                    hiTail.next = p;
                                hiTail = p;
                                ++hc;
                            }
                        }
                        ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                            (hc != 0) ? new TreeBin<K,V>(lo) : t;
                        hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                            (lc != 0) ? new TreeBin<K,V>(hi) : t;
                        setTabAt(nextTab, i, ln);
                        setTabAt(nextTab, i + n, hn);
                        setTabAt(tab, i, fwd);
                        advance = true;
                    }
                }
            }
        }
    }
}

 

代碼邏輯請看註釋,整個擴容操做分爲兩個部分：

第一部分是構建一個nextTable,它的容量是原來的兩倍，這個操做是單線程完成的。新建table數組的代碼爲:Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1],在原容量大小的基礎上右移一位。

第二個部分就是將原來table中的元素複製到nextTable中，主要是遍歷複製的過程。 根據運算獲得當前遍歷的數組的位置i，而後利用tabAt方法得到i位置的元素再進行判斷：

  1. 若是這個位置爲空，就在原table中的i位置放入forwardNode節點，這個也是觸發併發擴容的關鍵點；
  2. 若是這個位置是Node節點（fh>=0），若是它是一個鏈表的頭節點，就構造一個反序鏈表，把他們分別放在nextTable的i和i+n的位置上
  3. 若是這個位置是TreeBin節點（fh<0），也作一個反序處理，而且判斷是否須要untreefi，把處理的結果分別放在nextTable的i和i+n的位置上
  4. 遍歷過全部的節點之後就完成了複製工做，這時讓nextTable做爲新的table，而且更新sizeCtl爲新容量的0.75倍 ，完成擴容。設置爲新容量的0.75倍代碼爲 sizeCtl = (n << 1) - (n >>> 1)，仔細體會下是否是很巧妙，n<<1至關於n右移一位表示n的兩倍即2n,n>>>1左右一位至關於n除以2即0.5n,而後二者相減爲2n-0.5n=1.5n,是否是恰好等於新容量的0.75倍即2n*0.75=1.5n。

3.6 與size相關的一些方法

對於ConcurrentHashMap來講，這個table裏到底裝了多少東西實際上是個不肯定的數量，由於不可能在調用size()方法的時候像GC的「stop the world」同樣讓其餘線程都停下來讓你去統計，所以只能說這個數量是個估計值。對於這個估計值，ConcurrentHashMap也是大費周章才計算出來的。

爲了統計元素個數，ConcurrentHashMap定義了一些變量和一個內部類

/**
 * A padded cell for distributing counts.  Adapted from LongAdder
 * and Striped64\.  See their internal docs for explanation.
 */
@sun.misc.Contended static final class CounterCell {
    volatile long value;
    CounterCell(long x) { value = x; }
}

/******************************************/ 

/**
 * 實際上保存的是hashmap中的元素個數  利用CAS鎖進行更新
 但它並不用返回當前hashmap的元素個數 

 */
private transient volatile long baseCount;
/**
 * Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
 */
private transient volatile int cellsBusy;

/**
 * Table of counter cells. When non-null, size is a power of 2.
 */
private transient volatile CounterCell[] counterCells;

 

mappingCount與size方法

mappingCount與size方法的相似 從給出的註釋來看，應該使用mappingCount代替size方法 兩個方法都沒有直接返回basecount 而是統計一次這個值，而這個值其實也是一個大概的數值，所以可能在統計的時候有其餘線程正在執行插入或刪除操做。

public int size() {
    long n = sumCount();
    return ((n < 0L) ? 0 :
            (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
            (int)n);
}
 /**
 * Returns the number of mappings. This method should be used
 * instead of {@link #size} because a ConcurrentHashMap may
 * contain more mappings than can be represented as an int. The
 * value returned is an estimate; the actual count may differ if
 * there are concurrent insertions or removals.
 *
 * @return the number of mappings
 * @since 1.8
 */
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;//全部counter的值求和
        }
    }
    return sum;
}

 

addCount方法

在put方法結尾處調用了addCount方法，把當前ConcurrentHashMap的元素個數+1這個方法一共作了兩件事,更新baseCount的值，檢測是否進行擴容。

private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    //利用CAS方法更新baseCount的值 
    if ((as = counterCells) != null ||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell a; long v; int m;
        boolean uncontended = true;
        if (as == null || (m = as.length - 1) < 0 ||
            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
            !(uncontended =
              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
            fullAddCount(x, uncontended);
            return;
        }
        if (check <= 1)
            return;
        s = sumCount();
    }
    //若是check值大於等於0 則須要檢驗是否須要進行擴容操做
    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);
            }
            //當前線程是惟一的或是第一個發起擴容的線程  此時nextTable=null
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}

 

4. 總結

JDK6,7中的ConcurrentHashmap主要使用Segment來實現減少鎖粒度，分割成若干個Segment，在put的時候須要鎖住Segment，get時候不加鎖，使用volatile來保證可見性，當要統計全局時（好比size），首先會嘗試屢次計算modcount來肯定，這幾回嘗試中，是否有其餘線程進行了修改操做，若是沒有，則直接返回size。若是有，則須要依次鎖住全部的Segment來計算。

1.8以前put定位節點時要先定位到具體的segment，而後再在segment中定位到具體的桶。而在1.8的時候摒棄了segment臃腫的設計，直接針對的是Node[] tale數組中的每個桶，進一步減少了鎖粒度。而且防止拉鍊過長致使性能降低，當鏈表長度大於8的時候採用紅黑樹的設計。

主要設計上的變化有如下幾點:

  1. 不採用segment而採用node，鎖住node來實現減少鎖粒度。
  2. 設計了MOVED狀態 當resize的中過程當中 線程2還在put數據，線程2會幫助resize。
  3. 使用3個CAS操做來確保node的一些操做的原子性，這種方式代替了鎖。
  4. sizeCtl的不一樣值來表明不一樣含義，起到了控制的做用。
  5. 採用synchronized而不是ReentrantLock
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程