9張圖深入剖析ConcurrentHashMap_ZenDei技術網路在線

前言在日常的開發中，我們經常使用key-value鍵值對的HashMap，其使用哈希表實現，用空間換取時間，提升查詢性能但在多線程的併發場景中，HashMap並不是線程安全的如果想使用線程安全的，可以使用ConcurrentHashMap、HashTable、Collections.synch ...

前言

在日常的開發中，我們經常使用key-value鍵值對的HashMap，其使用哈希表實現，用空間換取時間，提升查詢性能

但在多線程的併發場景中，HashMap並不是線程安全的

如果想使用線程安全的，可以使用ConcurrentHashMap、HashTable、Collections.synchronizedMap等

但由於後面二者使用synchronized的粒度太大，因此一般不使用，而使用併發包中的ConcurrentHashMap

在ConcurrentHashMap中，使用volatile保證記憶體可見性，使得讀場景下不需要“加鎖”保證原子性

在寫場景下使用CAS+synchronized，synchronized只鎖哈希表某個索引位置上的首節點，相當於細粒度加鎖，增大併發性能

本篇文章將從ConcurrentHashMap的使用，讀、寫、擴容的實現原理，設計思想等方面進行剖析

查看本文前需要瞭解哈希表、volatile、CAS、synchronized等知識

volatile可以查看這篇文章：5個案例和流程圖讓你從0到1搞懂volatile關鍵字

CAS、synchronized可以查看這篇文章：15000字、6個代碼案例、5個原理圖讓你徹底搞懂Synchronized

使用ConcurrentHashMap

ConcurrentHashMap是併發場景下線程安全的Map，可以在併發場景下查詢存儲K、V鍵值對

不可變對象是絕對線程安全的，無論外界如何使用，都線程安全

ConcurrentHashMap並不是絕對線程安全的，只提供方法的線程安全，如果在外層使用錯誤依舊會導致線程不安全

來看下麵的案例，使用value存儲自增調用次數，開啟10個線程每個執行100次，最終結果應該是1000次，但錯誤的使用導致不足1000


    public void test() {
//        Map<String, Integer> map = new HashMap(16);
        Map<String, Integer> map = new ConcurrentHashMap(16);

        String key = "key";
        CountDownLatch countDownLatch = new CountDownLatch(10);


        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    incr(map, key);
//                    incrCompute(map, key);
                }
                countDownLatch.countDown();
            }).start();
        }

        try {
            //阻塞到線程跑完
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //1000不到
        System.out.println(map.get(key));
    }

	private void incr(Map<String, Integer> map, String key) {
        map.put(key, map.getOrDefault(key, 0) + 1);
    }

在自增方法incr中，先進行讀操作，再計算，最後進行寫操作，這種複合操作沒有保證原子性，導致最終所有結果累加一定不為1000

正確的使用方式是使用JDK8提供的預設方法compute

ConcurrentHashMap實現compute的原理是在put中使用同步手段後再進行計算

	private void incrCompute(Map<String, Integer> map, String key) {
        map.compute(key, (k, v) -> Objects.isNull(v) ? 1 : v + 1);
    }

數據結構

與HashMap類似，使用哈希表+鏈表/紅黑樹實現

哈希表

哈希表的實現由數組構成，當發生哈希衝突（哈希演算法得到同一索引）時使用鏈地址法構建成鏈表

當鏈表上的節點太長，遍歷尋找開銷大，超過閾值時（鏈表節點超過8個、哈希表長度大於64），樹化成紅黑樹減少遍歷尋找開銷，時間複雜度從O（n）優化為（log n）

ConcurrentHashMap由Node數組構成，在擴容時會存在新舊兩個哈希表：table、nextTable

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>, Serializable {	
	//哈希表 node數組
	transient volatile Node<K,V>[] table;
    
    //擴容時為了相容讀寫，會存在兩個哈希表，這個是新哈希表
    private transient volatile Node<K,V>[] nextTable;
    
    // 預設為 0
    // 當初始化時, 為 -1
    // 當擴容時, 為 -(1 + 擴容線程數)
    // 當初始化或擴容完成後，為 下一次的擴容的閾值大小
    private transient volatile int sizeCtl;
    
    //擴容時 用於指定遷移區間的下標
    private transient volatile int transferIndex;
    
    //統計每個哈希槽中的元素數量
    private transient volatile CounterCell[] counterCells;
}

節點

Node用於實現哈希表數組的節點和發生哈希衝突時，構建成鏈表的節點

//實現哈希表的節點，數組和鏈表時使用
static class Node<K,V> implements Map.Entry<K,V> {
    //節點哈希值
	final int hash;
	final K key;
	volatile V val;
    //作為鏈表時的 後續指針 
	volatile Node<K,V> next;    	
}

// 擴容時如果某個 bin 遷移完畢, 用 ForwardingNode 作為舊 table bin 的頭結點
static final class ForwardingNode<K,V> extends Node<K,V> {}

// 用在 compute 以及 computeIfAbsent 時, 用來占位, 計算完成後替換為普通 Node
static final class ReservationNode<K,V> extends Node<K,V> {}

// 作為 treebin 的頭節點, 存儲 root 和 first
static final class TreeBin<K,V> extends Node<K,V> {}

// 作為 treebin 的節點, 存儲 parent, left, right
static final class TreeNode<K,V> extends Node<K,V> {}

節點哈希值

//轉發節點
static final int MOVED     = -1;
//紅黑樹在數組中的節點
static final int TREEBIN   = -2;
//占位節點
static final int RESERVED  = -3;

轉發節點：繼承Node，用於擴容時設置在舊哈希表某索引的首節點，遇到轉發節點要去新哈希表中尋找

static final class ForwardingNode<K,V> extends Node<K,V> {
    	//新哈希表
        final Node<K,V>[] nextTable;
    	
        ForwardingNode(Node<K,V>[] tab) {
            //哈希值設置為-1
            super(MOVED, null, null, null);
            this.nextTable = tab;
        }
}

紅黑樹在數組中的節點 TreeBin：繼承Node，first指向紅黑樹的首節點

static final class TreeBin<K,V> extends Node<K,V> {
        TreeNode<K,V> root;
    	//紅黑樹首節點
        volatile TreeNode<K,V> first;
}

紅黑樹節點TreeNode

static final class TreeNode<K,V> extends Node<K,V> {
        TreeNode<K,V> parent;  
        TreeNode<K,V> left;
        TreeNode<K,V> right;
        TreeNode<K,V> prev; 
    	boolean red;
}

占位節點：繼承Node，需要計算時（使用computer方法），先使用占位節點占位，計算完再構建節點取代占位節點

	static final class ReservationNode<K,V> extends Node<K,V> {
        ReservationNode() {
            super(RESERVED, null, null, null);
        }

        Node<K,V> find(int h, Object k) {
            return null;
        }
    }

實現原理

構造

在構造時會檢查入參，然後根據需要存儲的數據容量、負載因數計算哈希表容量，最後將哈希表容量調整成2次冪

構造時並不會初始化，而是等到使用再進行創建（懶載入）

	public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
        //檢查負載因數、初始容量
        if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
        
        //concurrencyLevel：1
        if (initialCapacity < concurrencyLevel)   // Use at least as many bins
            initialCapacity = concurrencyLevel;   // as estimated threads
        //計算大小 = 容量/負載因數 向上取整
        long size = (long)(1.0 + (long)initialCapacity / loadFactor);
        //如果超過最大值就使用最大值 
        //tableSizeFor 將大小調整為2次冪
        int cap = (size >= (long)MAXIMUM_CAPACITY) ?
            MAXIMUM_CAPACITY : tableSizeFor((int)size);
        
        //設置容量
        this.sizeCtl = cap;
    }

讀-get

讀場景使用volatile保證可見性，即使其他線程修改也是可見的，不用使用其他手段保證同步

讀操作需要在哈希表中尋找元素，經過擾動演算法打亂哈希值，再使用哈希值通過哈希演算法得到索引，根據索引上的首節點分為多種情況處理

擾動演算法將哈希值充分打亂（避免造成太多的哈希衝突），符號位&0保證結果正數

int h = spread(key.hashCode())

擾動演算法：哈希值高低16位異或運算

經過擾動演算法後，&HASH_BITS = 0x7fffffff （011111...），符號位為0保證結果為正數

負數的哈希值表示特殊的作用，比如：轉發節點、樹的首節點、占位節點等
```
	static final int spread(int h) {
        return (h ^ (h >>> 16)) & HASH_BITS;
    }
```
使用打亂的哈希值經過哈希演算法得到數組中的索引（下標）

n 為哈希表長度：(n = tab.length)

(e = tabAt(tab, (n - 1) & h)

h為計算後的哈希值，哈希值%（哈希表長度-1）就能求出索引位置

為了性能提升，規定哈希表長度為2的n次冪，哈希表長度二進位一定是1000....，而(n-1)的二進位一定是0111...

因此(n - 1) & h計算索引，進行與運算的結果一定在0~n-1之間 使用位運算提升性能
得到數組上的節點後，需要進行比較

找到哈希表上的首個節點後，進行比較key 查看是否是當前節點

比較規則：先對哈希值進行比較，如果對象哈希值相同，那麼可能是同一個對象，還需要比較key（==與equals），如果哈希值都不相同，那麼肯定不是同一個對象

先比較哈希值的好處就是提升查找性能，如果直接使用equals 可能時間複雜度會上升（比如String的equals）
使用鏈地址法解決哈希衝突，因此找到節點後可能遍歷鏈表或樹；由於哈希表存在擴容，也可能要去新節點上尋找

4.1 首節點比較成功，直接返回

4.2 首節點哈希值為負，說明該節點是特殊情況的:轉發節點、樹的首節點、計算的預訂占位節點
- 如果是轉發節點，正在擴容則去新數組上找
- 如果是TreeBin則去紅黑樹中尋找
- 如果是占位節點直接返回空
4.3 遍歷該鏈表依次比較

get代碼

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    //1.spread：擾動演算法 + 讓key的哈希值不能為負數,因為負數哈希值代表紅黑樹或ForwardingNode
    int h = spread(key.hashCode());
    //2.(n - 1) & h：下標、索引 實際上就是數組長度模哈希值 位運算效率更高
    //e：哈希表中對應索引位置上的節點
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        //3.如果哈希值相等,說明可能找到,再比較key
        if ((eh = e.hash) == h) {
            //4.1 key相等說明找到 返回
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        //4.2 首節點哈希值為負，說明該節點是轉發節點，當前正在擴容則去新數組上找
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        
        //4.3 遍歷該鏈表,能找到就返回值,不能返回null
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

寫-put

添加元素時，使用CAS+synchronized（只鎖住哈希表中某個首節點）的同步方式保證原子性

獲取哈希值：擾動演算法+確保哈希值為正數
哈希表為空，CAS保證一個線程初始化

	private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            //小於0 說明其他線程在初始化 讓出CPU時間片 後續初始化完退出
            if ((sc = sizeCtl) < 0)
                Thread.yield(); 
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                //CAS將SIZECTL設置成-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;
    }

將哈希值通過哈希演算法獲取索引上的節點 f = tabAt(tab, i = (n - 1) & hash)
根據不同情況進行處理
- 4.1 首節點為空時，直接CAS往索引位置添加節點 casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null))
- 4.2 首節點hash為MOVED -1時，表示該節點是轉發節點，說明正在擴容，幫助擴容
- 4.3 首節點加鎖
  - 4.3.1 遍歷鏈表尋找並添加/覆蓋
  - 4.3.2 遍歷樹尋找並添加/覆蓋
addCount統計每個節點上的數據，並檢查擴容

put代碼

//onlyIfAbsent為true時,如果原來有k,v則這次不會覆蓋
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    //1.獲取哈希值：擾動演算法+確保哈希值為正數
    int hash = spread(key.hashCode());
    int binCount = 0;
    //樂觀鎖思想 CSA+失敗重試
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        //2.哈希表為空 CAS保證只有一個線程初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        //3. 哈希演算法求得索引找到索引上的首節點
        //4.1 節點為空時,直接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.2 索引首節點hash 為MOVED 說明該節點是轉發節點，當前正在擴容,去幫助擴容
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            //4.3 首節點 加鎖
            synchronized (f) {
                //首節點沒變
                if (tabAt(tab, i) == f) {
                    //首節點哈希值大於等於0 說明節點是鏈表上的節點  
                    //4.3.1 遍歷鏈表尋找然後添加/覆蓋
                    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;
                                //onlyIfAbsent為false允許覆蓋（使用xxIfAbsent方法時，有值就不覆蓋）
                                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;
                            }
                        }
                    }
                    //如果是紅黑樹首節點,則找到對應節點再覆蓋
                    //4.3.2 遍歷樹尋找然後添加/覆蓋
                    else if (f instanceof TreeBin) {
                        Node<K,V> p;
                        binCount = 2;
                        //如果是添加返回null,返回不是null則出來添加
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                       value)) != null) {
                            oldVal = p.val;
                            //覆蓋
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }
            
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    //鏈表上的節點超過TREEIFY_THRESHOLD 8個（不算首節點） 並且 數組長度超過64才樹化，否則擴容
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    //5.添加計數,用於統計元素（添加節點的情況）
    addCount(1L, binCount);
    return null;
}

擴容

為了避免頻繁發生哈希衝突，當哈希表中的元素數量 / 哈希表長度超過負載因數時，進行擴容（增大哈希表的長度）

一般來說擴容都是增大哈希表長度的2倍，比如從32到64保證長度是2次冪；如果擴容長度達到整型上限則使用整型最大值

當發生擴容時，需要將數組中每個槽里的鏈表或樹遷移到新數組中

如果處理器是多核，那麼這個遷移的操作並不是一個線程單獨完成的，而是會讓其他線程也來幫助遷移

在遷移時讓每個線程從右往左的每次遷移多個槽，遷移完再判斷是否全部遷移完，如果沒遷移完則繼續迴圈遷移

擴容操作主要在transfer方法中，擴容主要在三個場景下：

addCount：添加完節點增加計數檢查擴容
helpTransfer：線程put時發現正在遷移，來幫忙擴容
tryPresize：嘗試調整容量（批量添加putAll，樹化數組長度沒超過64時treeifyBin調用）

分為以下3個步驟

根據CPU核數、哈希表總長度計算每次遷移多少個槽，最小16個
新哈希表為空，說明是初始化
迴圈遷移
- 3.1 分配負責遷移的區間 [bround,i]（可能存在多線程同時遷移）
- 3.2 遷移：分為鏈表遷移、樹遷移
  
  鏈表遷移
  1. 將鏈表上的節點充分散列到新哈希表的index、index+舊哈希表長度的兩個下標中（與HashMap類似）
  2. 將index位置鏈表中的節點（hash & 哈希表長度），結果為0的放到新數組的index位置，結果為1放到新數組index+舊哈希表長度的位置
    
    比如舊哈希表長度為16，在索引3的位置上，16的二進位是10000，hash&16 => hash& 10000 ，也就是說節點哈希值第5位是0就放到新哈希表的3位置上，是1就放到新哈希表的3+16下標
  3. 使用頭插法將計算結果為0構建成ln鏈表，為1構建成hn鏈表，為方便構建鏈表，會先尋找lastRun節點：lastRun節點及後續節點都為同一鏈表上的節點，方便遷移
    
    構建鏈表前先構建lastRun，比如圖中lastRun e->f ，先將lastRun放到ln鏈表上，在遍歷原始鏈表，遍歷到a ：a->e->f，遍歷到b：b->a->e->f
  1. 每遷移完一個索引位置就將轉發節點設置到原哈希表對應位置，當其他線程進行讀get操作時，根據轉發節點來新哈希表中尋找，進行寫put操作時，來幫助擴容（其他區間進行遷移）

擴容代碼

//tab 舊哈希表
//nextTab 新哈希表
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        //1.計算每次遷移多少個槽
        //n:哈希表長度（多少個槽）
        int n = tab.length, stride;
        //stride:每次負責遷移多少個槽
        //NCPU: CPU核數
        //如果是多核，每次遷移槽數 = 總槽數無符號右移3位（n/8）再除CPU核數  
        //每次最小遷移槽數 = MIN_TRANSFER_STRIDE = 16
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
    
        //2.如果新哈希表為空，說明是初始化
        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用於記錄 每次負責遷移的槽右區間下標，從右往左分配，起始為最右
            transferIndex = n;
        }
        //新哈希表長度
        int nextn = nextTab.length;
        //創建轉發節點，轉發節點一般設置在舊哈希表首節點，通過轉發節點可以找到新哈希表
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        //advance:是否繼續迴圈遷移
        boolean advance = true;
        // 
        boolean finishing = false; // to ensure sweep before committing nextTab
        //3.迴圈遷移
        for (int i = 0, bound = 0;;) {
            Node<K,V> f; int fh;
            //3.1 分配負責遷移的區間
            //bound為左區間 i為右區間
            while (advance) {
                int nextIndex, nextBound;
                //處理完一個槽 右區間 自減
                if (--i >= bound || finishing)
                    advance = false;
                //transferIndex<=0說明 要遷移的區間全分配完
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false;
                }
                //CAS設置本次遷移的區間，防止多線程分到相同區間
                else if (U.compareAndSwapInt
                         (this, TRANSFERINDEX, nextIndex,
                          nextBound = (nextIndex > stride ?
                                       nextIndex - stride : 0))) {
                    bound = nextBound;
                    i = nextIndex - 1;
                    advance = false;
                }
            }
            
            //3.2 遷移
            
            //3.2.1 如果右區間i不再範圍，說明遷移完
            if (i < 0 || i >= n || i + n >= nextn) {
                int sc;
                //如果完成遷移，設置哈希表、數量
                if (finishing) {
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                }
                //CAS 將sizeCtl數量-1 表示 一個線程遷移完成 
                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                    //如果不是最後一條線程直接返回
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    //是最後一條線程設置finishing為true  後面再迴圈 去設置哈希表、數量等操作
                    finishing = advance = true;
                    i = n; // recheck before commit
                }
            }
            //3.2.2 如果舊哈希表i位置節點為空就CAS設置成轉發節點
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            //3.2.3 如果舊哈希表該位置首節點是轉發節點，說明其他線程已處理，重新迴圈
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
                //3.2.4 對首節點加鎖 遷移
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        Node<K,V> ln, hn;
                        //3.2.4.1 鏈表遷移
                        //首節點哈希值大於等於0 說明 是鏈表節點
                        if (fh >= 0) {
                            int runBit = fh & n;
                            Node<K,V> lastRun = f;
                            //尋找lastRun節點 
                            for (Node<K,V> p = f.next; p != null; p = p.next) {
                                int b = p.hash & n;
                                if (b != runBit) {
                                    runBit = b;
                                    lastRun = p;
                                }
                            }
                            //如果最後一次計算值是0
                            //lastRun節點以及後續節點計算值都是0構建成ln鏈表 否則 都是1構建成hn鏈表
                            if (runBit == 0) {
                                ln = lastRun;
                                hn = null;
                            }
                            else {
                                hn = lastRun;
                                ln = null;
                            }
                            
                            //遍歷構建ln、hn鏈表 （頭插）
                            for (Node<K,V> p = f; p != lastRun; p = p.next) {
                                int ph = p.hash; K pk = p.key; V pv = p.val;
                                //頭插：Node構造第四個參數是後繼節點
                                if ((ph & n) == 0)
                                    ln = new Node<K,V>(ph, pk, pv, ln);
                                else
                                    hn = new Node<K,V>(ph, pk, pv, hn);
                            }
                            //設置ln鏈表到i位置
                            setTabAt(nextTab, i, ln);
                            //設置hn鏈表到i+n位置
                            setTabAt(nextTab, i + n, hn);
                            //設置轉發節點
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }
                        //3.2.4.2 樹遷移
                        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;
                        }
                    }
                }
            }
        }
    }

實現原理並沒有對紅黑樹進行太多描述，一方面是紅黑樹的概念太多，另一方面是我忘的差不多了（已經老了，不像大學那樣可以手寫紅黑樹了）

還有一方面是：我認為只需要知道使用紅黑樹的好處就足夠，而且工作中也不常用，就算死扣紅黑樹要怎麼變色、左旋、右旋去滿足紅黑樹的條件也沒什麼意義，感興趣的同學去學習就好了

迭代器

ConcurrentHashMap中的迭代器是弱一致性，在獲取時使用記錄的哈希表重新構建新對象

Entry迭代器：

public Iterator<Map.Entry<K,V>> iterator() {
    ConcurrentHashMap<K,V> m = map;
    Node<K,V>[] t;
    int f = (t = m.table) == null ? 0 : t.length;
    return new EntryIterator<K,V>(t, f, 0, f, m);
}

key迭代器

public Enumeration<K> keys() {
    Node<K,V>[] t;
    int f = (t = table) == null ? 0 : t.length;
    return new KeyIterator<K,V>(t, f, 0, f, this);
}

value迭代器

public Enumeration<V> elements() {
    Node<K,V>[] t;
    int f = (t = table) == null ? 0 : t.length;
    return new ValueIterator<K,V>(t, f, 0, f, this);
}

總結

ConcurrentHashMap使用哈希表的數據結構，當發生哈希衝突時，使用鏈地址法解決，將哈希到同一索引的節點構建成鏈表，當數據量達到一定閾值，會將鏈表轉化為紅黑樹

ConcurrentHashMap使用volatile修飾存儲數據，使得在讀場景下對其他線程的修改可見，不需要使用同步機制，使用CAS與synchronzied保證寫場景下的原子性

在get查詢數據時，先將key的哈希值通過擾動演算法（高低16位異或）並保證結果為正數（與上符號位0），再與上哈希表長度-1求出索引值，找到索引後再根據不同情況查找（比較先判斷哈希值，相等再判斷key）

在put添加/覆蓋數據時，也是先通過擾動演算法和哈希求出索引位置，在根據不同情況查找，找到則覆蓋，找不到則替換

在需要擴容時，會為線程安排需要遷移的槽區間，當其他線程進行put時也會來幫忙遷移，每次線程遷移完一個槽，會設置轉發節點到原哈希表中，這樣有線程查詢就可以通過轉發節點來新哈希表中查找，當遷移完所有槽時留一個線程來設置哈希表、數量等

迭代器使用的是弱一致性，在獲取迭代器時通過哈希表去構建新的對象

ConcurrentHashMap 只保證相對線程安全，不能保證絕對線程安全，如果需要進行一系列操作時，要正確的去使用

本文由博客一文多發平臺 OpenWrite 發佈！