1.ConcurrentHashmap簡介

在使用HashMap時在多線程情況下擴(kuò)容會出現(xiàn)CPU接近100%的情況暇屋，因?yàn)閔ashmap并不是線程安全的蟀悦，通常我們可以使用在java體系中古老的hashtable類仁期，該類基本上所有的方法都采用synchronized進(jìn)行線程安全的控制计螺，可想而知蕾各，在高并發(fā)的情況下，每次只有一個線程能夠獲取對象監(jiān)視器鎖，這樣的并發(fā)性能的確不令人滿意遏乔。另外一種方式通過Collections的Map<K,V> synchronizedMap(Map<K,V> m)將hashmap包裝成一個線程安全的map义矛。比如SynchronzedMap的put方法源碼為：

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

實(shí)際上SynchronizedMap實(shí)現(xiàn)依然是采用synchronized獨(dú)占式鎖進(jìn)行線程安全的并發(fā)控制的。同樣盟萨，這種方案的性能也是令人不太滿意的凉翻。針對這種境況，Doug Lea大師不遺余力的為我們創(chuàng)造了一些線程安全的并發(fā)容器捻激，讓每一個java開發(fā)人員倍感幸福制轰。相對于hashmap來說，ConcurrentHashMap就是線程安全的map胞谭，其中利用了鎖分段的思想提高了并發(fā)度垃杖。

ConcurrentHashMap在JDK1.6的版本網(wǎng)上資料很多，有興趣的可以去看看丈屹。

JDK 1.6版本關(guān)鍵要素：

segment繼承了ReentrantLock充當(dāng)鎖的角色调俘，為每一個segment提供了線程安全的保障；
segment維護(hù)了哈希散列表的若干個桶泉瞻，每個桶由HashEntry構(gòu)成的鏈表脉漏。

而到了JDK 1.8的ConcurrentHashMap就有了很大的變化苞冯，光是代碼量就足足增加了很多袖牙。1.8版本舍棄了segment，并且大量使用了synchronized舅锄，以及CAS無鎖操作以保證ConcurrentHashMap操作的線程安全性鞭达。至于為什么不用ReentrantLock而是Synchronzied呢？實(shí)際上皇忿，synchronzied做了很多的優(yōu)化畴蹭，包括偏向鎖，輕量級鎖鳍烁，重量級鎖叨襟，可以依次向上升級鎖狀態(tài)，但不能降級（關(guān)于synchronized可以看這篇文章）幔荒，因此糊闽，使用synchronized相較于ReentrantLock的性能會持平甚至在某些情況更優(yōu)，具體的性能測試可以去網(wǎng)上查閱一些資料爹梁。另外右犹，底層數(shù)據(jù)結(jié)構(gòu)改變?yōu)椴捎脭?shù)組+鏈表+紅黑樹的數(shù)據(jù)形式。

2.關(guān)鍵屬性及類

在了解ConcurrentHashMap的具體方法實(shí)現(xiàn)前姚垃，我們需要系統(tǒng)的來看一下幾個關(guān)鍵的地方念链。

ConcurrentHashMap的關(guān)鍵屬性

table

volatile Node<K,V>[] table://裝載Node的數(shù)組，作為ConcurrentHashMap的數(shù)據(jù)容器，采用懶加載的方式掂墓，直到第一次插入數(shù)據(jù)的時候才會進(jìn)行初始化操作谦纱，數(shù)組的大小總是為2的冪次方。
nextTable

volatile Node<K,V>[] nextTable; //擴(kuò)容時使用君编，平時為null服协，只有在擴(kuò)容的時候才為非null
sizeCtl

volatile int sizeCtl; //該屬性用來控制table數(shù)組的大小，根據(jù)是否初始化和是否正在擴(kuò)容有幾種情況：當(dāng)值為負(fù)數(shù)時：如果為-1表示正在初始化啦粹，如果為-N則表示當(dāng)前正有N-1個線程進(jìn)行擴(kuò)容操作偿荷；當(dāng)值為正數(shù)時：如果當(dāng)前數(shù)組為null的話表示table在初始化過程中需要新建數(shù)組的長度，若已經(jīng)初始化了唠椭，表示當(dāng)前數(shù)據(jù)容器（table數(shù)組）可用容量也可以理解成臨界值（插入節(jié)點(diǎn)數(shù)超過了該臨界值就需要擴(kuò)容）跳纳，具體指為數(shù)組的長度n*加載因子；當(dāng)值為0時贪嫂，即為默認(rèn)初始值寺庄。
sun.misc.Unsafe U

在ConcurrentHashMapde的實(shí)現(xiàn)中可以看到大量的U.compareAndSwapXXXX的方法去修改ConcurrentHashMap的一些屬性。這些方法實(shí)際上是利用了CAS算法保證了線程安全性力崇，這是一種樂觀策略斗塘，假設(shè)每一次操作都不會產(chǎn)生沖突，當(dāng)且僅當(dāng)沖突發(fā)生的時候再去嘗試亮靴。而CAS操作依賴于現(xiàn)代處理器指令集馍盟，通過底層CMPXCHG指令實(shí)現(xiàn)。CAS(V,O,N)核心思想為：若當(dāng)前變量實(shí)際值V與期望的舊值O相同茧吊，則表明該變量沒被其他線程進(jìn)行修改贞岭，因此可以安全的將新值N賦值給變量；若當(dāng)前變量實(shí)際值V與期望的舊值O不相同搓侄，則表明該變量已經(jīng)被其他線程做了處理瞄桨，此時將新值N賦給變量操作就是不安全的，在進(jìn)行重試讶踪。而在大量的同步組件和并發(fā)容器的實(shí)現(xiàn)中使用CAS是通過sun.misc.Unsafe類實(shí)現(xiàn)的芯侥，該類提供了一些可以直接操控內(nèi)存和線程的底層操作，可以理解為java中的“指針”乳讥。該成員變量的獲取是在靜態(tài)代碼塊中：
```
 static {
     try {
         U = sun.misc.Unsafe.getUnsafe();
         .......
     } catch (Exception e) {
         throw new Error(e);
     }
 }
```

ConcurrentHashMap中關(guān)鍵內(nèi)部類

Node

Node類實(shí)現(xiàn)了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進(jìn)行修飾的雏婶，也就是為了保證內(nèi)存可見性物赶。

TreeNode

樹節(jié)點(diǎn)，繼承于承載數(shù)據(jù)的Node類留晚。而紅黑樹的操作是針對TreeBin類的酵紫，從該類的注釋也可以看出告嘲，也就是TreeBin會將TreeNode進(jìn)行再一次封裝

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

TreeBin

這個類并不負(fù)責(zé)包裝用戶的key、value信息奖地，而是包裝的很多TreeNode節(jié)點(diǎn)橄唬。實(shí)際的ConcurrentHashMap“數(shù)組”中，存放的是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
         ......
 }

ForwardingNode

在擴(kuò)容時才會出現(xiàn)的特殊節(jié)點(diǎn)，其key,value,hash全部為null犬庇。并擁有nextTable指針引用新的table數(shù)組僧界。

 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關(guān)鍵操作

在上面我們提及到在ConcurrentHashMap中會大量使用CAS修改它的屬性和一些操作。因此臭挽，在理解ConcurrentHashMap的方法前我們需要了解下面幾個常用的利用CAS算法來保障線程安全的操作捂襟。

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數(shù)組中索引為i的Node元素。

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操作設(shè)置table數(shù)組中索引為i的元素

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

該方法用來設(shè)置table數(shù)組中索引為i的元素

3.重點(diǎn)方法講解

在熟悉上面的這核心信息之后欢峰，我們接下來就來依次看看幾個常用的方法是怎樣實(shí)現(xiàn)的葬荷。

3.1 實(shí)例構(gòu)造器方法

在使用ConcurrentHashMap第一件事自然而然就是new 出來一個ConcurrentHashMap對象，一共提供了如下幾個構(gòu)造器方法：

// 1\. 構(gòu)造一個空的map纽帖，即table數(shù)組還未初始化宠漩，初始化放在第一次插入數(shù)據(jù)時，默認(rèn)大小為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大小懊直，加載因子以及并發(fā)度（預(yù)計同時操作數(shù)據(jù)的線程）
ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)

ConcurrentHashMap一共給我們提供了5中構(gòu)造器方法扒吁，具體使用請看注釋，我們來看看第2種構(gòu)造器吹截，傳入指定大小時的情況瘦陈，該構(gòu)造器源碼為：

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

這段代碼的邏輯請看注釋，很容易理解蛾默，如果小于0就直接拋出異常懦铺，如果指定值大于了所允許的最大值的話就取最大值，否則支鸡，在對指定值做進(jìn)一步處理冬念。最后將cap賦值給sizeCtl,關(guān)于sizeCtl的說明請看上面的說明，當(dāng)調(diào)用構(gòu)造器方法之后牧挣，sizeCtl的大小應(yīng)該就代表了ConcurrentHashMap的大小急前，即table數(shù)組長度。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;
}

通過注釋就很清楚了裆针，該方法會將調(diào)用構(gòu)造器方法時指定的大小轉(zhuǎn)換成一個2的冪次方數(shù)，也就是說ConcurrentHashMap的大小一定是2的冪次方，比如世吨，當(dāng)指定大小為18時澡刹，為了滿足2的冪次方特性，實(shí)際上concurrentHashMapd的大小為2的5次方（32）耘婚。另外罢浇，需要注意的是，調(diào)用構(gòu)造器方法的時候并未構(gòu)造出table數(shù)組（可以理解為ConcurrentHashMap的數(shù)據(jù)容器）沐祷，只是算出table數(shù)組的長度嚷闭，當(dāng)?shù)谝淮蜗駽oncurrentHashMap插入數(shù)據(jù)的時候才真正的完成初始化創(chuàng)建table數(shù)組的工作。

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

代碼的邏輯請見注釋凌受，有可能存在一個情況是多個線程同時走到這個方法中，為了保證能夠正確初始化思杯，在第1步中會先通過if進(jìn)行判斷胜蛉，若當(dāng)前已經(jīng)有一個線程正在初始化即sizeCtl值變?yōu)?1，這個時候其他線程在If判斷為true從而調(diào)用Thread.yield()讓出CPU時間片色乾。正在進(jìn)行初始化的線程會調(diào)用U.compareAndSwapInt方法將sizeCtl改為-1即正在初始化的狀態(tài)誊册。另外還需要注意的事情是，在第四步中會進(jìn)一步計算數(shù)組中可用的大小即為數(shù)組實(shí)際大小n乘以加載因子0.75.可以看看這里乘以0.75是怎么算的暖璧，0.75為四分之三案怯，這里n - (n >>> 2)是不是剛好是n-(1/4)n=(3/4)n，挺有意思的吧:)澎办。如果選擇是無參的構(gòu)造器的話嘲碱，這里在new Node數(shù)組的時候會使用默認(rèn)大小為DEFAULT_CAPACITY（16），然后乘以加載因子0.75為12局蚀，也就是說數(shù)組的可用大小為12麦锯。

3.3 put方法

使用ConcurrentHashMap最長用的也應(yīng)該是put和get方法了吧，我們先來看看put方法是怎樣實(shí)現(xiàn)的琅绅。調(diào)用put方法時實(shí)際具體實(shí)現(xiàn)是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\. 如果當(dāng)前table還沒有初始化先調(diào)用initTable方法將tab進(jìn)行初始化
        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\. 當(dāng)前正在擴(kuò)容
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    //5\. 當(dāng)前為鏈表千扶，在鏈表中插入新的鍵值對
                    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.當(dāng)前為紅黑樹料祠，將新的鍵值對插入到紅黑樹中
                    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.插入完鍵值對后再根據(jù)實(shí)際大小看是否需要轉(zhuǎn)換成紅黑樹
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    //8.對當(dāng)前容量大小進(jìn)行檢查，如果超過了臨界值（實(shí)際大小*加載因子）就需要擴(kuò)容 
    addCount(1L, binCount);
    return null;
}

put方法的代碼量有點(diǎn)長澎羞，我們按照上面的分解的步驟一步步來看髓绽。從整體而言，為了解決線程安全的問題妆绞，ConcurrentHashMap使用了synchronzied和CAS的方式顺呕。在之前了解過HashMap以及1.8版本之前的ConcurrenHashMap都應(yīng)該知道ConcurrentHashMap結(jié)構(gòu)圖接谨，為了方面下面的講解這里先直接給出，如果對這有疑問的話塘匣，可以在網(wǎng)上隨便搜搜即可脓豪。

image

如圖（圖片摘自網(wǎng)絡(luò)），ConcurrentHashMap是一個哈希桶數(shù)組忌卤，當(dāng)不出現(xiàn)哈希沖突的時候均勻的出現(xiàn)在數(shù)組的每個元素扫夜。當(dāng)出現(xiàn)哈希沖突的時候，是標(biāo)準(zhǔn)的鏈地址的解決方式驰徊，將hash值相同的節(jié)點(diǎn)構(gòu)成鏈表的形式笤闯，稱為“拉鏈法”，另外棍厂，在1.8版本中為了防止拉鏈過長颗味，當(dāng)鏈表的長度大于8的時候會將鏈表轉(zhuǎn)換成紅黑樹。table數(shù)組中的每個元素實(shí)際上是單鏈表的頭結(jié)點(diǎn)或者紅黑樹的根節(jié)點(diǎn)牺弹。當(dāng)插入鍵值對時首先應(yīng)該定位到要插入的桶浦马，即插入table數(shù)組的索引i處。那么张漂，怎樣計算得出索引i呢晶默？當(dāng)然是根據(jù)key的hashCode值。

spread()重哈希航攒，以減小Hash沖突

我們知道對于一個hash表來說磺陡，hash值分散的不夠均勻的話會大大增加哈希沖突的概率，從而影響到hash表的性能漠畜。因此通過spread方法進(jìn)行了一次重hash從而大大減小哈希沖突的可能性币他。spread方法為：

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

該方法主要是將key的hashCode的低16位于高16位進(jìn)行異或運(yùn)算，這樣不僅能夠使得hash值能夠分散能夠均勻減小hash沖突的概率憔狞，另外另外只用到了異或運(yùn)算蝴悉，在性能開銷上也能兼顧，做到平衡的trade-off躯喇。

2.初始化table

緊接著到第2步辫封，會判斷當(dāng)前table數(shù)組是否初始化了，沒有的話就調(diào)用initTable進(jìn)行初始化廉丽，該方法在上面已經(jīng)講過了。

3.能否直接將新值插入到table數(shù)組中

從上面的結(jié)構(gòu)示意圖就可以看出存在這樣一種情況妻味，如果插入值待插入的位置剛好所在的table數(shù)組為null的話就可以直接將值插入即可正压。那么怎樣根據(jù)hash確定在table中待插入的索引i呢？很顯然可以通過hash值與數(shù)組的長度取模操作责球，從而確定新值插入到數(shù)組的哪個位置焦履。而之前我們提過ConcurrentHashMap的大小總是2的冪次方拓劝，(n - 1) & hash運(yùn)算等價于對長度n取模，也就是hash%n嘉裤，但是位運(yùn)算比取模運(yùn)算的效率要高很多郑临，Doug lea大師在設(shè)計并發(fā)容器的時候也是將性能優(yōu)化到了極致，令人欽佩屑宠。

確定好數(shù)組的索引i后厢洞，就可以可以tabAt()方法（該方法在上面已經(jīng)說明了，有疑問可以回過頭去看看）獲取該位置上的元素典奉，如果當(dāng)前Node f為null的話躺翻，就可以直接用casTabAt方法將新值插入即可。

4.當(dāng)前是否正在擴(kuò)容

如果當(dāng)前節(jié)點(diǎn)不為null卫玖，且該節(jié)點(diǎn)為特殊節(jié)點(diǎn)（forwardingNode）的話公你，就說明當(dāng)前concurrentHashMap正在進(jìn)行擴(kuò)容操作，關(guān)于擴(kuò)容操作假瞬，下面會作為一個具體的方法進(jìn)行講解陕靠。那么怎樣確定當(dāng)前的這個Node是不是特殊的節(jié)點(diǎn)了？是通過判斷該節(jié)點(diǎn)的hash值是不是等于-1（MOVED）,代碼為(fh = f.hash) == MOVED脱茉，對MOVED的解釋在源碼上也寫的很清楚了：

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

5.當(dāng)table[i]為鏈表的頭結(jié)點(diǎn)懦傍，在鏈表中插入新值

在table[i]不為null并且不為forwardingNode時，并且當(dāng)前Node f的hash值大于0（fh >= 0）的話說明當(dāng)前節(jié)點(diǎn)f為當(dāng)前桶的所有的節(jié)點(diǎn)組成的鏈表的頭結(jié)點(diǎn)芦劣。那么接下來粗俱，要想向ConcurrentHashMap插入新值的話就是向這個鏈表插入新值。通過synchronized (f)的方式進(jìn)行加鎖以實(shí)現(xiàn)線程安全性虚吟。往鏈表中插入節(jié)點(diǎn)的部分代碼為：

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相同的節(jié)點(diǎn)串慰，就直接覆蓋即可偏塞；2. 如果直到找到了鏈表的末尾都沒有找到的話，就直接將待插入的鍵值對追加到鏈表的末尾即可

6.當(dāng)table[i]為紅黑樹的根節(jié)點(diǎn)邦鲫，在紅黑樹中插入新值

按照之前的數(shù)組+鏈表的設(shè)計方案灸叼，這里存在一個問題，即使負(fù)載因子和Hash算法設(shè)計的再合理庆捺，也免不了會出現(xiàn)拉鏈過長的情況古今，一旦出現(xiàn)拉鏈過長，甚至在極端情況下滔以，查找一個節(jié)點(diǎn)會出現(xiàn)時間復(fù)雜度為O(n)的情況捉腥，則會嚴(yán)重影響ConcurrentHashMap的性能，于是你画，在JDK1.8版本中抵碟，對數(shù)據(jù)結(jié)構(gòu)做了進(jìn)一步的優(yōu)化桃漾，引入了紅黑樹。而當(dāng)鏈表長度太長（默認(rèn)超過8）時拟逮，鏈表就轉(zhuǎn)換為紅黑樹撬统，利用紅黑樹快速增刪改查的特點(diǎn)提高ConcurrentHashMap的性能，其中會用到紅黑樹的插入敦迄、刪除恋追、查找等算法。當(dāng)table[i]為紅黑樹的樹節(jié)點(diǎn)時的操作為：

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判斷當(dāng)前table[i]是否是樹節(jié)點(diǎn)颅崩，這下也正好驗(yàn)證了我們在最上面介紹時說的TreeBin會對TreeNode做進(jìn)一步封裝几于，對紅黑樹進(jìn)行操作的時候針對的是TreeBin而不是TreeNode。這段代碼很簡單沿后，調(diào)用putTreeVal方法完成向紅黑樹插入新節(jié)點(diǎn)沿彭，同樣的邏輯，如果在紅黑樹中存在于待插入鍵值對的Key相同（hash值相等并且equals方法判斷為true）的節(jié)點(diǎn)的話尖滚，就覆蓋舊值喉刘，否則就向紅黑樹追加新節(jié)點(diǎn)。

7.根據(jù)當(dāng)前節(jié)點(diǎn)個數(shù)進(jìn)行調(diào)整

當(dāng)完成數(shù)據(jù)新節(jié)點(diǎn)插入之后漆弄，會進(jìn)一步對當(dāng)前鏈表大小進(jìn)行調(diào)整睦裳，這部分代碼為：

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

很容易理解，如果當(dāng)前鏈表節(jié)點(diǎn)個數(shù)大于等于8（TREEIFY_THRESHOLD）的時候撼唾，就會調(diào)用treeifyBin方法將tabel[i]（第i個散列桶）拉鏈轉(zhuǎn)換成紅黑樹廉邑。

至此，關(guān)于Put方法的邏輯就基本說的差不多了倒谷，現(xiàn)在來做一些總結(jié)：

整體流程：

首先對于每一個放入的值蛛蒙，首先利用spread方法對key的hashcode進(jìn)行一次hash計算，由此來確定這個值在 table中的位置渤愁；
如果當(dāng)前table數(shù)組還未初始化牵祟，先將table數(shù)組進(jìn)行初始化操作；
如果這個位置是null的抖格，那么使用CAS操作直接放入诺苹；
如果這個位置存在結(jié)點(diǎn)，說明發(fā)生了hash碰撞雹拄，首先判斷這個節(jié)點(diǎn)的類型收奔。如果該節(jié)點(diǎn)fh==MOVED(代表forwardingNode,數(shù)組正在進(jìn)行擴(kuò)容)的話，說明正在進(jìn)行擴(kuò)容办桨；
如果是鏈表節(jié)點(diǎn)（fh>0）,則得到的結(jié)點(diǎn)就是hash值相同的節(jié)點(diǎn)組成的鏈表的頭節(jié)點(diǎn)筹淫。需要依次向后遍歷確定這個新加入的值所在位置。如果遇到hash值與key值都與新加入節(jié)點(diǎn)是一致的情況呢撞，則只需要更新value值即可损姜。否則依次向后遍歷，直到鏈表尾插入這個結(jié)點(diǎn)殊霞；
如果這個節(jié)點(diǎn)的類型是TreeBin的話摧阅，直接調(diào)用紅黑樹的插入方法進(jìn)行插入新的節(jié)點(diǎn)；
插入完節(jié)點(diǎn)之后再次檢查鏈表長度绷蹲，如果長度大于8棒卷，就把這個鏈表轉(zhuǎn)換成紅黑樹；
對當(dāng)前容量大小進(jìn)行檢查祝钢，如果超過了臨界值（實(shí)際大小*加載因子）就需要擴(kuò)容比规。

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]桶節(jié)點(diǎn)的key與查找的key相同，則直接返回
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        // 3\. 當(dāng)前節(jié)點(diǎn)hash小于0說明為樹節(jié)點(diǎn)疤估，在紅黑樹中查找即可
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        while ((e = e.next) != null) {
        //4\. 從鏈表中查找灾常，查找到則返回該節(jié)點(diǎn)的value，否則就返回null即可
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

代碼的邏輯請看注釋铃拇，首先先看當(dāng)前的hash桶數(shù)組節(jié)點(diǎn)即table[i]是否為查找的節(jié)點(diǎn)钞瀑，若是則直接返回；若不是慷荔，則繼續(xù)再看當(dāng)前是不是樹節(jié)點(diǎn)雕什？通過看節(jié)點(diǎn)的hash值是否為小于0，如果小于0則為樹節(jié)點(diǎn)显晶。如果是樹節(jié)點(diǎn)在紅黑樹中查找節(jié)點(diǎn)贷岸；如果不是樹節(jié)點(diǎn)，那就只剩下為鏈表的形式的一種可能性了吧碾，就向后遍歷查找節(jié)點(diǎn)凰盔，若查找到則返回節(jié)點(diǎn)的value即可，若沒有找到就返回null倦春。

3.5 transfer方法

當(dāng)ConcurrentHashMap容量不足的時候户敬，需要對table進(jìn)行擴(kuò)容。這個方法的基本思想跟HashMap是很像的睁本，但是由于它是支持并發(fā)擴(kuò)容的尿庐，所以要復(fù)雜的多。原因是它支持多線程進(jìn)行擴(kuò)容操作呢堰，而并沒有加鎖抄瑟。我想這樣做的目的不僅僅是為了滿足concurrent的要求，而是希望利用并發(fā)處理去減少擴(kuò)容帶來的時間影響枉疼。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數(shù)組皮假，容量為之前的兩倍
    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.將原數(shù)組中的元素復(fù)制到新數(shù)組中去
        //4.5 for循環(huán)退出，擴(kuò)容結(jié)束修改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 當(dāng)前數(shù)組中第i個元素為null惹资，用CAS設(shè)置成特殊節(jié)點(diǎn)forwardingNode(可以理解成占位符)
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        //4.2 如果遍歷到ForwardingNode節(jié)點(diǎn)  說明這個點(diǎn)已經(jīng)被處理過了 直接跳過  這里是控制并發(fā)擴(kuò)容的核心
        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 處理當(dāng)前節(jié)點(diǎn)為鏈表的頭結(jié)點(diǎn)的情況贺纲，構(gòu)造兩個鏈表，一個是原鏈表  另一個是原鏈表的反序排列
                        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節(jié)點(diǎn)  表示已經(jīng)處理過該節(jié)點(diǎn)
                       setTabAt(tab, i, fwd);
                       //設(shè)置advance為true 返回到上面的while循環(huán)中 就可以執(zhí)行i--操作
                       advance = true;
                    }
                    //4.4 處理當(dāng)前節(jié)點(diǎn)是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;
                    }
                }
            }
        }
    }
}

代碼邏輯請看注釋,整個擴(kuò)容操作分為兩個部分：

第一部分是構(gòu)建一個nextTable,它的容量是原來的兩倍猴誊，這個操作是單線程完成的。新建table數(shù)組的代碼為:Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1],在原容量大小的基礎(chǔ)上右移一位侮措。

第二個部分就是將原來table中的元素復(fù)制到nextTable中懈叹，主要是遍歷復(fù)制的過程。

根據(jù)運(yùn)算得到當(dāng)前遍歷的數(shù)組的位置i分扎，然后利用tabAt方法獲得i位置的元素再進(jìn)行判斷：

如果這個位置為空澄成，就在原table中的i位置放入forwardNode節(jié)點(diǎn)，這個也是觸發(fā)并發(fā)擴(kuò)容的關(guān)鍵點(diǎn)笆包；
如果這個位置是Node節(jié)點(diǎn)（fh>=0）环揽，如果它是一個鏈表的頭節(jié)點(diǎn)，就構(gòu)造一個反序鏈表庵佣，把他們分別放在nextTable的i和i+n的位置上
如果這個位置是TreeBin節(jié)點(diǎn)（fh<0）歉胶，也做一個反序處理，并且判斷是否需要untreefi巴粪，把處理的結(jié)果分別放在nextTable的i和i+n的位置上
遍歷過所有的節(jié)點(diǎn)以后就完成了復(fù)制工作通今，這時讓nextTable作為新的table，并且更新sizeCtl為新容量的0.75倍肛根，完成擴(kuò)容辫塌。設(shè)置為新容量的0.75倍代碼為 sizeCtl = (n << 1) - (n >>> 1)，仔細(xì)體會下是不是很巧妙派哲，n<<1相當(dāng)于n右移一位表示n的兩倍即2n,n>>>1左右一位相當(dāng)于n除以2即0.5n,然后兩者相減為2n-0.5n=1.5n,是不是剛好等于新容量的0.75倍即2n*0.75=1.5n臼氨。最后用一個示意圖來進(jìn)行總結(jié)（圖片摘自網(wǎng)絡(luò)）：

image

3.6 與size相關(guān)的一些方法

對于ConcurrentHashMap來說，這個table里到底裝了多少東西其實(shí)是個不確定的數(shù)量芭届，因?yàn)?strong>不可能在調(diào)用size()方法的時候像GC的“stop the world”一樣讓其他線程都停下來讓你去統(tǒng)計储矩，因此只能說這個數(shù)量是個估計值。對于這個估計值褂乍，ConcurrentHashMap也是大費(fèi)周章才計算出來的持隧。

為了統(tǒng)計元素個數(shù)，ConcurrentHashMap定義了一些變量和一個內(nèi)部類

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

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

/**
 * 實(shí)際上保存的是hashmap中的元素個數(shù)  利用CAS鎖進(jìn)行更新
 但它并不用返回當(dāng)前hashmap的元素個數(shù) 

 */
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方法的類似從給出的注釋來看逃片，應(yīng)該使用mappingCount代替size方法兩個方法都沒有直接返回basecount 而是統(tǒng)計一次這個值屡拨，而這個值其實(shí)也是一個大概的數(shù)值，因此可能在統(tǒng)計的時候有其他線程正在執(zhí)行插入或刪除操作。

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方法結(jié)尾處調(diào)用了addCount方法呀狼，把當(dāng)前ConcurrentHashMap的元素個數(shù)+1這個方法一共做了兩件事,更新baseCount的值裂允，檢測是否進(jìn)行擴(kuò)容。

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 則需要檢驗(yàn)是否需要進(jìn)行擴(kuò)容操作
    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;
                 //如果已經(jīng)有其他線程在執(zhí)行擴(kuò)容操作
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            //當(dāng)前線程是唯一的或是第一個發(fā)起擴(kuò)容的線程  此時nextTable=null
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}

4. 總結(jié)

JDK6,7中的ConcurrentHashmap主要使用Segment來實(shí)現(xiàn)減小鎖粒度赠潦，分割成若干個Segment叫胖，在put的時候需要鎖住Segment草冈，get時候不加鎖她奥，使用volatile來保證可見性，當(dāng)要統(tǒng)計全局時（比如size）怎棱，首先會嘗試多次計算modcount來確定哩俭，這幾次嘗試中，是否有其他線程進(jìn)行了修改操作拳恋，如果沒有凡资，則直接返回size。如果有谬运，則需要依次鎖住所有的Segment來計算隙赁。

而在1.8的時候摒棄了segment臃腫的設(shè)計，這種設(shè)計在定位到具體的桶時梆暖，要先定位到具體的segment伞访，然后再

在segment中定位到具體的桶。而到了1.8的時候是針對的是Node[] tale數(shù)組中的每一個桶轰驳，進(jìn)一步減小了鎖粒度厚掷。并且防止拉鏈過長導(dǎo)致性能下降，當(dāng)鏈表長度大于8的時候采用紅黑樹的設(shè)計级解。

主要設(shè)計上的變化有以下幾點(diǎn):

不采用segment而采用node冒黑，鎖住node來實(shí)現(xiàn)減小鎖粒度。
設(shè)計了MOVED狀態(tài) 當(dāng)resize的中過程中線程2還在put數(shù)據(jù)勤哗，線程2會幫助resize抡爹。
使用3個CAS操作來確保node的一些操作的原子性，這種方式代替了鎖芒划。
sizeCtl的不同值來代表不同含義冬竟，起到了控制的作用。
采用synchronized而不是ReentrantLock

ConcurrentHashMap源碼分析