概述

本文主要介紹ConcurrentHashMap在JDK8中的原始碼實現和原理。在JDK8中，開發人員幾乎把ConcurrentHashMap的原始碼重寫了一遍，原始碼由之前的2000多行增加到了6300行左右，因此實現也就複雜很多。在學習之前，最好先了解下如下知識：

1、ReentrantLock的實現和原理。
2、Synchronized的實現和原理。
3、硬體對併發支援的CAS操作及JVM中Unsafe對CAS的實現。
4、JDK1.7中關於ConcurrentHashMap的實現原理。
5、volatile的實現原理。
6、二叉樹，平衡二叉樹以及紅黑樹的知識。

資料結構

JDK 1.8中ConcurrentHashMap拋棄了分段鎖技術的實現，直接採用CAS + synchronized保證併發更新的安全性，底層採用陣列+連結串列+紅黑樹的儲存結構。其包含核心靜態內部類 Node。
首先通過一張圖來看下資料結構吧：

資料結構圖

說明：資料結構採用陣列 + 連結串列 + 紅黑樹的方式實現。當連結串列中(bucket)的節點個數超過8個時，會轉換成紅黑樹的資料結構儲存，這樣設計的目的是為了減少同一個連結串列衝突過大情況下的讀取效率。

Java8中主要做了如下優化:
1.將Segment拋棄掉了，直接採用Node（繼承自Map.Entry）作為table元素。
2.修改時，不再採用ReentrantLock加鎖，直接用內建synchronized加鎖，java8的內建鎖比之前版本優化了很多，相較ReentrantLock，效能不併差。
3.size方法優化，增加了CounterCell內部類，用於平行計算每個bucket的元素數量。

內部類和繼承關係

Java8中ConcurrentHashMap增加了很多內部類來支援一些操作和優化效能。下面介紹幾個核心的內部類。

ConcurrentHashMap幾個核心內部類關係圖

（1）Node類：存放元素的key,value,hash值,next下一個連結串列節點的引用。用於bucket為連結串列時。
（2）TreeBin：內部屬性有root，first節點，以及root節點的鎖狀態變數lockState，這是一個讀寫鎖的狀態。用於存放紅黑樹的root節點，並用讀寫鎖lockState控制在寫操作即將要調整樹結構前，先讓讀執行緒完成讀操作。從連結串列結構調整為紅黑樹時，table中索引下標儲存的即為TreeBin。
（3）TreeNode：紅黑樹的節點，存放了父節點，左子節點，右子節點的引用，以及紅黑節點標識。
（4）ForwardingNode：在呼叫transfer()方法期間，插入bucket頭部的節點，主要用來標識在擴容時元素的移動狀態，即是否在擴容時還有併發的插入節點，並保證該節點也能夠移動到擴容後的表中。
（5）ReservationNode：佔位節點，不儲存任何資訊，無實際用處，僅用於computeIfAbsent和compute方法中。

重要屬性介紹

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>, Serializable {
    // table最大容量，為2的冪次方
    private static final int MAXIMUM_CAPACITY = 1 << 30;
    // 預設table初始容量大小
    private static final int DEFAULT_CAPACITY = 16;
    // 預設支援併發更新的執行緒數量
    private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    // table的負載因子
    private static final float LOAD_FACTOR = 0.75f;
    // 連結串列轉換為紅黑樹的節點數閾值，超過這個值，連結串列轉換為紅黑樹
    static final int TREEIFY_THRESHOLD = 8;
    // 在擴容期間，由紅黑樹轉換為連結串列的閾值，小於這個值，resize期間紅黑樹就會轉為連結串列
    static final int UNTREEIFY_THRESHOLD = 6;
    // 轉為紅黑樹時，紅黑樹中節點的最小個數
    static final int MIN_TREEIFY_CAPACITY = 64;
    // 擴容時，併發轉移節點(transfer方法)時，每次轉移的最小節點數
    private static final int MIN_TRANSFER_STRIDE = 16;

    // 以下常量定義了特定節點類hash欄位的值
    static final int MOVED     = -1; // ForwardingNode類物件的hash值
    static final int TREEBIN   = -2; // TreeBin類物件的hash值
    static final int RESERVED  = -3; // ReservationNode類物件的hash值
    static final int HASH_BITS = 0x7fffffff; // 普通Node節點的hash初始值

    // table陣列
    transient volatile Node<K,V>[] table;
    // 擴容時，下一個容量大小的talbe，用於將原table元素移動到這個table中
    private transient volatile Node<K,V>[] nextTable;
    // 基礎計數器
    private transient volatile long baseCount;
    // table初始容量大小以及擴容容量大小的引數，也用於標識table的狀態
    // 其有幾個值來代表也用來代表table的狀態:
    // -1 ：標識table正在初始化
    // - N : 標識table正在進行擴容，並且有N - 1個執行緒一起在進行擴容
    // 正數：初始table的大小，如果值大於初始容量大小，則表示擴容後的table大小。
    private transient volatile int sizeCtl;
    // 擴容時，下一個節點轉移的bucket索引下標
    private transient volatile int transferIndex;
    // 一種自旋鎖，是專為防止多處理器併發而引入的一種鎖，用於建立CounterCells時使用，
    // 主要用於size方法計數時，有併發執行緒插入而計算修改的節點數量，
    // 這個數量會與baseCount計數器彙總後得出size的結果。
    private transient volatile int cellsBusy;
    // 主要用於size方法計數時，有併發執行緒插入而計算修改的節點數量，
    // 這個數量會與baseCount計數器彙總後得出size的結果。
    private transient volatile CounterCell[] counterCells;
    // 其他省略
}

以上的一些屬性，在初始化，擴容，連結串列轉紅黑樹等方法中用到。屬性眾多，sizeCtl，counterCells都比較重要。
sizeCtl：即作為table初始化狀態的標識，也用作擴容時的執行緒數標識，還用作初始和擴容後table的容量標識，用處很多，不同狀態值代表的含義如下：

1、 -1：標識table正在初始化
2、- N：標識table正在進行擴容，並且有N - 1個執行緒一起在進行擴容
3、正數：初始化table的大小，如果值大於初始容量大小，則表示擴容後的table大小。

counterCells在put size等方法中再介紹。

核心方法原始碼分析

put方法

put方法，呼叫的是putVal方法

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

再看下putVal方法的實現

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());// 計算key的hash值
    int binCount = 0;// 表示table中索引下標代表的連結串列或紅黑樹中的節點數量
    // 採用自旋方式，等待table第一次put初始化完成，或等待鎖或等待擴容成功然後再插入
    for (Node<K,V>[] tab = table;;) {
        // f節點標識table中的索引節點，可能是連結串列的head，也可能是紅黑樹的head
        // n:table的長度，i:插入元素在table的索引下標，fh : head節點的hash值
        Node<K,V> f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0)// 第一次插入元素，先執行初始化
            tab = initTable();
        // 定位到的索引下標節點(head)為null，表示第一次在此索引插入，
        // 不加鎖直接插入在head之後，在casTabAt中採用Unsafe的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
        }
        // head節點為ForwadingNode型別節點，表示table正在擴容，連結串列或紅黑樹也加入到幫助擴容操作中
        else if ((fh = f.hash) == MOVED) 
            tab = helpTransfer(tab, f);
        else {// 索引下標存在元素，且為普通Node節點，給head加鎖後執行插入或更新
            V oldVal = null;
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) {// 為普通連結串列節點，還記得之前定義的幾種常量Hash值嗎？
                        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;
                            // 插入新元素，每次插在單向連結串列的末尾，這點與Java7中不同（插在首部）
                            if ((e = e.next) == null) {
                                pred.next = new Node<K,V>(hash, key, value, null);
                                break;
                            }
                        }
                    }
                    else if (f instanceof TreeBin) {// head為樹節點，按樹的方式插入節點
                        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;
                        }
                    }
                }
            }
            // 連結串列節點樹超過閾值8，將連結串列轉換為紅黑樹結構
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    // 如果是插入新元素，則將連結串列或紅黑樹最新的節點數量加入到CounterCells中
    addCount(1L, binCount);
    return null;
}

初看起來，putVal方法很複雜，但筆者在程式碼上增加了比較詳細的註釋，看起來就方便的多啦，總體流程和步驟如下：

1、採用自旋的方式，保證首次put時，當前執行緒或其他併發put的執行緒等待table初始化完成後再次重試插入。
2、採用自旋的方式，檢查當前插入的元素在table中索引下標是否正在執行擴容，如果正在擴容，則幫助進行擴容，完成後，重試插入到新的table中。
3、插入的table索引下標不為空，則對連結串列或紅黑樹的head節點加synchronized鎖，再插入或更新。訪問入口是Head節點，其他執行緒訪問head，在連結串列或紅黑樹插入或修改時必須等待synchronized釋放。
4、插入後，如果發現連結串列節點數大於等於閾值8，呼叫treeifyBin方法，將連結串列轉換為紅黑樹結構，提高讀寫效能。treeifyBin方法內部也同樣採用synchronized方式保證執行緒安全性。
5、插入元素後，會將索引代表的連結串列或紅黑樹的最新節點數量更新到baseCount或CounterCell中。

putVal方法用到了很多字方法，如下，我們一一來分析：
（1）spread：計算元素的hash值
（2）initTable：初始化table，在首次執行put，computeIfAbsent，computIfPresent,compute,merge方法時呼叫。
（3）tabAt：用於定位key在table中的索引節點(head節點)。
（4）casTabAt：採用Unsafe的compareAndSwapObject方法，用CAS的方式更新或替換節點。
（5）helpTransfer：幫忙擴容。
（6）treeifyBin：連結串列轉紅黑樹，實現原始碼就不分析了，感興趣的同學可以自行研究下。
（7）addCount：連結串列或紅黑樹節點最新數量新增到CounterCell中。

spread方法

計算key的hash值，將key的hashCode的高16位也加入到計算中，避免平凡衝突。如果僅用key的hashCode作為hash值，那麼2,4之類的整形key值，只有低4位，那麼很容易發生衝突。

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

initTable方法

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {// while自旋
        // sizeCtl小於0，表示table正在被其他執行緒執行初始化，
        // 放棄初始化競爭，自旋等待初始化完成
        // 還記得前面介紹的sizeCtl的含義嗎？
        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) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

初始化比較簡單，步驟如下：

1、自旋檢查table是否完成初始化。
2、若發現sizeCtl值為負數，則放棄初始化的競爭，讓其他正在初始化的執行緒完成初始化。
3、如果沒有其他執行緒初始化，則用Unsafe.compareAndSwapInt更新sizeCtl的值為-1，表示table開始被當前執行緒執行初始化，其他執行緒禁止執行。
4、初始化：table設定為預設容量大小（元素並未初始化，只是劃定了大小），sizeCtl設為下次擴容table的size大小。
5、初始化完成。

整個初始化，用到了sizeCtl和Unsafe.compareAndSwapInt來保證初始化的執行緒安全性。有沒有覺得Doug Lea大神對併發程式設計的出神入化。

tabAt和casTabAt方法

這兩個方法比較簡單，都是利用Unsafe的CAS方法保證讀取和替換的原子性，保證執行緒安全。

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

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

疑問解答：為什麼table本身明明用了volatile修飾，不直接用table[i]的方式取節點，而非要用Unsafe.getObjectVolatile方法的CAS操作取節點。
答：雖然table本身是volatile型別，但僅僅是指table陣列引用本身，而陣列中每個元素並不是volatile型別，Unsafe.getObjectVolatile保證了每次從table中讀取某個位置連結串列引用的時候都是從主記憶體中讀取的，如果不用該方法，有可能讀的是快取中已有的該位置的舊資料。

helpTransfer方法

這是一個輔助擴容的方法，能夠支援擴容時直接加入到擴容中，其中真正擴容的核心方法是transfer，擴容前，會更新SIZECTL的值，表示併發擴容的執行緒數，transfer擴容方法太過複雜，本文不做介紹，將在下篇文章中介紹。

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

static final int resizeStamp(int n) {
    return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}

addCount方法

/**
 * Adds to count, and if table is too small and not already
 * resizing, initiates transfer. If already resizing, helps
 * perform transfer if work is available.  Rechecks occupancy
 * after a transfer to see if another resize is already needed
 * because resizings are lagging additions.
 *
 * @param x the count to add
 * @param check if <0, don't check resize, if <= 1 only check if uncontended
 */
private final void addCount(long x, int check) {
    // check,即連結串列或紅黑樹的節點數，<0不檢查是否正在擴容， 
    // <=1僅檢查是否存在競爭，沒有競爭則直接返回
    CounterCell[] as; long b, s;
    // 如果首次執行addCount，並且嘗試用CAS對baseCount計數失敗，表示有競爭，則執行如下操作。
    // 或者非首次addCount，也執行如下的操作
    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();
    }
    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);
            }
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}
// sumCount方法
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;
}

addCount方法做了如下操作：

1、判斷是否首次執行addCount，並判斷是否存在競爭關係，如果CAS成功，數量就成功彙總到baseCount中，如果CAS操作失敗，則表示有競爭，有其他執行緒併發插入，則修改的數量會被記錄到CounterCell中。
2、BaseCount和CounterCell相加就表示正常無併發下的節點數量和併發插入下的節點數量，table索引下標所代表的連結串列或紅黑樹節點的數量就能達到精確計算的效果。
3、在addCount時，還會去檢查sizeCtl是否為-N，以確定table是否正在擴容，如果正在擴容，則加入到擴容的操作中。

addCount方法所統計的數值baseCount和counterCells將會被用到size方法中，用於精確計算併發讀寫情況下table中元素的數量。這種設計多麼巧妙，不愧為Doug Lea大神的設計，因此這種設計思路也可以在實際的工作應用中多思考思考，作為很多併發統計問題的解決做設計參考。

size方法

public int size() {
    long n = sumCount();
    return ((n < 0L) ? 0 :
            (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
            (int)n);
}
// sumCount方法
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;
}

size方法最終執行的是sumCount方法，在sumCount方法中，其實就是將baseCount的數值與CounterCell表中併發情況下插入的節點數量進行彙總累加得到。這個結果也把併發的情況也考慮進去了。看這個方法之前最好先看addCount方法。

get方法

get方法步驟：
1、計算key的hash值，並定位table索引
2、若table索引下元素(head節點)為普通連結串列，則按連結串列的形式迭代遍歷。
3、若table索引下元素為紅黑樹TreeBin節點，則按紅黑樹的方式查詢(find方法)。

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;
        }
        // hash值小於-1,即為紅黑樹，還記得之前定義的TreeBin節點的hash值嗎
        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;
}

紅黑樹的查詢方法原始碼如下：
步驟如下：
1、檢查lockState是否為寫鎖，如果是，則表示有併發寫入執行緒在寫入，則按正常的連結串列方式遍歷並查詢。
2、如果沒有寫鎖，僅加讀鎖，然後按紅黑樹的方式查詢(TreeBin.findTreeNode方法)。

final Node<K,V> find(int h, Object k) {
    if (k != null) {
        for (Node<K,V> e = first; e != null; ) {
            int s; K ek;
            if (((s = lockState) & (WAITER|WRITER)) != 0) {
                if (e.hash == h &&
                    ((ek = e.key) == k || (ek != null && k.equals(ek))))
                    return e;
                e = e.next;
            }
            else if (U.compareAndSwapInt(this, LOCKSTATE, s, s + READER)) {
                TreeNode<K,V> r, p;
                try {
                    p = ((r = root) == null ? null : r.findTreeNode(h, k, null));
                } finally {
                    Thread w;
                    if (U.getAndAddInt(this, LOCKSTATE, -READER) ==
                        (READER|WAITER) && (w = waiter) != null)
                        LockSupport.unpark(w);
                }
                return p;
            }
        }
    }
    return null;
}

疑問解答：前文不是說了，連結串列元素超過8個時，會被轉成紅黑樹的結構嗎？為什麼在樹節點遍歷方法中，第一點仍然採用連結串列的方式遍歷？
回答：還記得TreeBin和TreeNode節點和Node節點的繼承關係嗎？Node本身可以鏈成一個連結串列，而TreeBin和TreeNode也繼承自Node節點，也自然繼承了next屬性，同樣擁有連結串列的性質，其實真正在儲存時，紅黑樹仍然是以連結串列形式儲存的，只是邏輯上TreeBin和TreeNode多了支援紅黑樹的root，first, parent，left，right，red屬性，在附加的屬性上進行邏輯上的引用和關聯，也就構造成了一顆樹。這一點有點像LinkedHashMap，裡面的節點又是在Table中，各個table中的元素又通過before和after引用進行雙向連結，達到各個節點之間在邏輯上互鏈起來的效果。

紅黑樹的查詢遍歷如下，其實就是二叉樹查詢，紅黑樹是按hash值的大小來構造左子節點和右子節點的，比父節點hash值小放在左邊，大則放在右邊的：

final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
    if (k != null) {
        TreeNode<K,V> p = this;
        do  {
            int ph, dir; K pk; TreeNode<K,V> q;
            TreeNode<K,V> pl = p.left, pr = p.right;
            if ((ph = p.hash) > h)
                p = pl;
            else if (ph < h)
                p = pr;
            else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
                return p;
            else if (pl == null)
                p = pr;
            else if (pr == null)
                p = pl;
            else if ((kc != null ||
                      (kc = comparableClassFor(k)) != null) &&
                     (dir = compareComparables(kc, k, pk)) != 0)
                p = (dir < 0) ? pl : pr;
            else if ((q = pr.findTreeNode(h, k, kc)) != null)
                return q;
            else
                p = pl;
        } while (p != null);
    }
    return null;
}

紅黑樹的原理和構造過程

請參見筆者寫的下一篇文章ConcurrentHashMap與紅黑樹實現分析Java8，單獨作為一篇文章來描述這個過程。

寫到這裡，已經花費我很多的時間來分析了，但依然沒有寫全，未完待續，後面將新增如何擴容的分析。

作者：Misout
連結：https://www.jianshu.com/p/85d158455861
來源：簡書
簡書著作權歸作者所有，任何形式的轉載都請聯絡作者獲得授權並註明出處。