Java集合之ConcurrentHashMap

亦凉 2021-08-14 02:05 432阅读 0赞

### ConcurrentHashMap ###

*  1. 简介
 *  2. 继承关系
 *  3. 深入源码
 *   *  3.1 属性
     *  3.2 内部类
     *   *  3.2.1 Node
         *  3.2.2 ForwardingNode
         *  3.2.3 TreeNode。
         *  3.2.4 TreeBin
     *  3.3 构造方法
     *   *  3.3.1 ConcurrentHashMap()
         *  3.3.2 ConcurrentHashMap(int initialCapacity)
         *  3.3.3 ConcurrentHashMap(int initialCapacity, float loadFactor)
         *  3.3.4 ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)
         *  3.3.5 ConcurrentHashMap(Map<? extends K, ? extends V> m)
     *  3.4 初始化
     *   *  3.4.1 initTable()
     *  3.5 哈希计算
     *   *  3.5.1 spread(int h)
     *  3.6 添加元素
     *   *  3.6.1 put(K key, V value)
     *  3.7 获取元素
     *   *  3.7.1 get(Object key)
     *  3.8 删除元素
     *   *  3.8.1 remove(Object key)
     *  3.9 扩容条件
     *   *  3.9.1 addCount(long x, int check)
     *  3.10 扩容
     *   *  3.10.1 tryPresize(int size)
     *  3.11 协助扩容（迁移元素）
     *  3.11 迁移元素
     *   *  3.11.1 transfer(Node

# 1. 简介 #

`ConcurrentHashMap`是`HashMap`的线程安全版本，内部也是使用（数组 + 链表 + 红黑树）的结构来存储元素。

相比于同样线程安全的`HashTable`来说，其并发效率较`HashTable`有极大地提高。

其采用 分段锁 的原理，在 `HashTable`中锁是 `HashTable`这个对象，而在 `ConcurrentHashMap`锁则是数组中的第一个元素。这样的话，我们比如我们要多数组中下标为`0`和`1`的桶进行操作，两个线程就可以同时对`0`和`1`进行操作，而`HashTable` 则必须只能串行操作，可见，并发的效率截然不同。

分段锁示意图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70]

# 2. 继承关系 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 1]

# 3. 深入源码 #

## 3.1 属性 ##

/** * 数组最大长度 */
        private static final int MAXIMUM_CAPACITY = 1 << 30;
    
        /** * 默认初始容量，必须是2的次幂 */
        private static final int DEFAULT_CAPACITY = 16;
    
        /** * 最大可能的数组大小（非2的幂）。 toArray和相关方法需要。 */
        static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
    
        /** * 该表的默认并发级别。没有被使用，定义该字段是为了为与此类的先前版本兼容。 */
        private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    
        /** * 加载因子 */
        private static final float LOAD_FACTOR = 0.75f;
    
        /** * 链表长度大于等于 8 时树化 */
        static final int TREEIFY_THRESHOLD = 8;
    
        /** * 链表长度小于等于 6 时树化 */
        static final int UNTREEIFY_THRESHOLD = 6;
    
        /** * 数组长度大于 64 时，链表才能转为红黑树 */
        static final int MIN_TREEIFY_CAPACITY = 64;
    
        /** * 每个线程负责转移桶的最小值 */
        private static final int MIN_TRANSFER_STRIDE = 16;
    
        /** * 生成sizeCtl所使用的bit位数 */
        private static int RESIZE_STAMP_BITS = 16;
    
        /** * 进行扩容所允许的最大线程数 */
        private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
    
        /** * 记录sizeCtl中的大小所需要进行的偏移位数 */
        private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
    
        /* * 标识 */
        static final int MOVED     = -1; // ForwardingNode的hash值，如果我们将一个桶的元素全部转移到新数组中，旧数组当前桶放一个ForwardNode节点
        static final int TREEBIN   = -2; // TreeNode的hash值
        static final int RESERVED  = -3; // ReservationNode的hash值
        static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
    
        /** 可用处理器数量 */
        static final int NCPU = Runtime.getRuntime().availableProcessors();
    
     	/* ---------------- Fields -------------- */
    
        /** * 用于保存元素的数组 */
        transient volatile Node<K,V>[] table;
    
        /** * 扩容时需要将元素转移到该数组 */
        private transient volatile Node<K,V>[] nextTable;
    
        /** * 用于计算键值对个数，主要在没有多线程竞争时使用，也用作表初始化竞赛期间的后备。 * 通过CAS更新。 * 原理类似于 LongAdder */
        private transient volatile long baseCount;
    
        /** * 表初始化和大小调整控制。如果为负，则表将被初始化或调整大小：-1用于初始化，否则为扩容 -（1 +活动的调整大小线程数）。 * 否则，当table为null时，保留创建时要使用的初始表大小，或者默认为0。初始化后，保留下一个要调整表大小的元素计数值。 */
        private transient volatile int sizeCtl;
    
        /** * 调整大小时要拆分的下一个表索引（加1）。 */
        private transient volatile int transferIndex;
    
        /** * 调整大小和/或创建CounterCell时使用的Spinlock（通过CAS锁定）。 */
        private transient volatile int cellsBusy;
    
        /** * 和 baseCount 共同计算出键值对的个数 */
        private transient volatile CounterCell[] counterCells;
    
    	// 保存所有key
        private transient KeySetView<K,V> keySet;
        // 保存所有的value
        private transient ValuesView<K,V> values;
        // 保存所有的键值对
        private transient EntrySetView<K,V> entrySet;

## 3.2 内部类 ##

### 3.2.1 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;
    
            Node(int hash, K key, V val, Node<K,V> next) { 
                this.hash = hash;
                this.key = key;
                this.val = val;
                this.next = next;
            }
       }

### 3.2.2 ForwardingNode ###

`ForwardingNode`：一个特殊的`Node`节点，`hash`值为`-1`，其中存储`nextTable`的引用。

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;
            }
    
    		// 注意 find 方法，其是在旧数组的桶中没有找到元素，则去扩容数组中查找
            Node<K,V> find(int h, Object k) { 
                // loop to avoid arbitrarily deep recursion on forwarding nodes
                outer: for (Node<K,V>[] tab = nextTable;;) { 
                    Node<K,V> e; int n;
                    if (k == null || tab == null || (n = tab.length) == 0 ||
                        (e = tabAt(tab, (n - 1) & h)) == null)
                        return null;
                    for (;;) { 
                        int eh; K ek;
                        if ((eh = e.hash) == h &&
                            ((ek = e.key) == k || (ek != null && k.equals(ek))))
                            return e;
                        if (eh < 0) { 
                        	// 递归查询，可能又发生了一次扩容
                            if (e instanceof ForwardingNode) { 
                                tab = ((ForwardingNode<K,V>)e).nextTable;
                                continue outer;
                            }
                            else
                                return e.find(h, k);
                        }
                        if ((e = e.next) == null)
                            return null;
                    }
                }
            }
        }

### 3.2.3 TreeNode。 ###

红黑树节点

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

### 3.2.4 TreeBin ###

`TreeBin` 用作树的头结点，只存储`root`和`first`节点，不存储节点的`key`、`value`值。

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.3 构造方法 ##

### 3.3.1 ConcurrentHashMap() ###

/** * 默认数组长度为16 */
        public ConcurrentHashMap() { 
        }

### 3.3.2 ConcurrentHashMap(int initialCapacity) ###

public ConcurrentHashMap(int initialCapacity) { 
            if (initialCapacity < 0)
                throw new IllegalArgumentException();
            
            // sizeCtl和容量都为第一个大于等于 initialCapacity * 1.5 + 1 的二次幂值
            // 如果 initialCapacity 为15，那么cap则为32
            // 而在 HashMap 中，则为16 
            int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                       MAXIMUM_CAPACITY :
                       tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
            this.sizeCtl = cap;
        }

### 3.3.3 ConcurrentHashMap(int initialCapacity, float loadFactor) ###

// 指定初始容量和加载因子
        public ConcurrentHashMap(int initialCapacity, float loadFactor) {   		
            this(initialCapacity, loadFactor, 1);
        }

### 3.3.4 ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) ###

// @param concurrencyLevel 并发更新线程的估计数量。
        public ConcurrentHashMap(int initialCapacity,
                                 float loadFactor, int concurrencyLevel) { 
            if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
                throw new IllegalArgumentException();
            if (initialCapacity < concurrencyLevel)   // Use at least as many bins
                initialCapacity = concurrencyLevel;   // as estimated threads
            long size = (long)(1.0 + (long)initialCapacity / loadFactor);
            int cap = (size >= (long)MAXIMUM_CAPACITY) ?
                MAXIMUM_CAPACITY : tableSizeFor((int)size);
            this.sizeCtl = cap;
        }

### 3.3.5 ConcurrentHashMap(Map<? extends K, ? extends V> m) ###

public ConcurrentHashMap(Map<? extends K, ? extends V> m) { 
            this.sizeCtl = DEFAULT_CAPACITY;
            putAll(m);
        }

可以看到，在任何一个构造方法中，都没有对存储Map元素Node的table变量进行初始化。而是在第一次put操作的时候在进行初始化。

构造方法与`HashMap`对比可以发现，没有了`HashMap`中的`threshold`和`loadFactor`，而是改用了`sizeCtl`来控制，而且只存储了容量在里面，那么它是怎么用的呢？官方给出的解释如下：

（1）-1，表示有线程正在进行初始化操作

（2）-(1 + nThreads)，表示有n个线程正在一起扩容

（3）0，默认值，后续在真正初始化的时候使用默认容量

（4）> 0，初始化或扩容完成后下一次的扩容门槛

## 3.4 初始化 ##

### 3.4.1 initTable() ###

/** * 使用sizeCtl中记录的大小初始化表。 */
        private final Node<K,V>[] initTable() { 
            Node<K,V>[] tab; int sc;
            // 一直循环直到数组被初始化过
            while ((tab = table) == null || tab.length == 0) { 
            	// sizeCtl 小于0，则说明数组正在初始化或扩容，所以当前线程让出CPU让其他线程完成初始化工作
                if ((sc = sizeCtl) < 0)
                    Thread.yield(); // lost initialization race; just spin
                
                // 通过 CAS 的方式将 sizeCtl 的值改为-1，表示当前线程正在进行扩容，
                // 这样其他线程获得CPU的执行权时，知道有其他线程在进行扩容，就可以让出CPU执行权了 
                else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { 
                    try { 
                    	// 在进行一次判断，防止其他线程完成了初始化工作
                        if ((tab = table) == null || tab.length == 0) { 
                        	// 如果sc为0则使用默认值16
                        	// 如果在构造方法中，指定过参数，那么sc为大于等于2的整数次幂的值
                            int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                            @SuppressWarnings("unchecked")
                            // 创建数组
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                            // 赋值给table数组
                            table = tab = nt;
                            // 设置sc为数组长度的0.75倍
                        	// n - (n >>> 2) = n - n/4 = 0.75
                            sc = n - (n >>> 2);
                        }
                    } finally { 
                    	 // 初始化后，sizeCtl变为数组长度的3/4，这时存储的是扩容门槛
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            return tab;
        }

（1）使用`CAS`锁控制只有一个线程初始化桶数组；

（2）`sizeCtl`在初始化后存储的是扩容门槛；

（3）扩容门槛写死的是桶数组大小的`0.75`倍，桶数组大小即`map`的容量，也就是最多存储多少个元素。

## 3.5 哈希计算 ##

### 3.5.1 spread(int h) ###

static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
    
        static final int spread(int h) { 
            return (h ^ (h >>> 16)) & HASH_BITS;
        }

以上便是 `ConcurrentHashMap`计算 `hash`值的方法，相较于 `HashMap`略由不同。

以下是 `HashMap` 的 `hash`值计算方法。

static final int hash(Object key) { 
            int h;
            return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
        }

通过对比我们可以发现，`ConcurrentHashMap`中并不能存储 `null`，并且相比于 `HashMap`，其多了个 `& HASH_BITS`，这样做的意义是啥？不是很清楚。

## 3.6 添加元素 ##

### 3.6.1 put(K key, V value) ###

public V put(K key, V value) { 
            return putVal(key, value, false);
        }
    
        final V putVal(K key, V value, boolean onlyIfAbsent) { 
        	// key 和 value 都不能为 null
            if (key == null || value == null) throw new NullPointerException();
            // 计算 hash 
            int hash = spread(key.hashCode());
            int binCount = 0;
            // 这里一直自旋，直到放入元素成功
            for (Node<K,V>[] tab = table;;) { 
                Node<K,V> f; int n, i, fh;
                // 如果数组没有被初始化，则去初始化
                if (tab == null || (n = tab.length) == 0)
                    tab = initTable();
                // 如果桶中第一个元素没有元素，则通过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
                }
                // 如果当前桶中的元素正在被迁移，则当前线程也帮助迁移元素
                else if ((fh = f.hash) == MOVED)
                    tab = helpTransfer(tab, f);
                else { 
                    V oldVal = null;
                    // 以桶中第一个元素为锁对象
                    synchronized (f) { 
                        if (tabAt(tab, i) == f) { 
                        	// 如果第一个节点是链表节点，则以链表的方式放入元素
                            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;
                                    }
                                }
                            }
                            // 如果当前元素是树节点，则以树的方式放入元素
                            else if (f instanceof TreeBin) { 
                                Node<K,V> p;
                                binCount = 2;
                                if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                               value)) != null) { 
                                    oldVal = p.val;
                                    if (!onlyIfAbsent)
                                        p.val = value;
                                }
                            }
                        }
                    }
                    if (binCount != 0) { 
                    	// 判断是否需要树化
                        if (binCount >= TREEIFY_THRESHOLD)
                            treeifyBin(tab, i);
                        if (oldVal != null)
                            return oldVal;
                        break;
                    }
                }
            }
            // 键值对个数+1
            addCount(1L, binCount);
            return null;
        }

整体流程跟`HashMap`比较类似，大致是以下几步：

（1）如果桶数组未初始化，则初始化；

（2）如果待插入的元素所在的桶为空，则尝试通过`CAS`的方式把此元素直接插入到桶的第一个位置；

（3）如果正在扩容，则当前线程一起加入到扩容的过程中；

（4）如果待插入的元素所在的桶不为空且不在迁移元素，则锁住这个桶（分段锁）；

（5）如果当前桶中元素以链表方式存储，则在链表中寻找该元素或者插入元素；

（6）如果当前桶中元素以红黑树方式存储，则在红黑树中寻找该元素或者插入元素；

（7）如果元素存在，则返回旧值；

（8）如果元素不存在，整个Map的元素个数加1，并检查是否需要扩容；

添加元素操作中使用的锁主要有（自旋锁 + `CAS` \+ `synchronized` \+ 分段锁）。

为什么使用`synchronized`而不是`ReentrantLock`？

因为`synchronized`已经得到了极大地优化，在特定情况下并不比`ReentrantLock`差。同时，如果使用`ReentrantLock`，则需要为每一个桶单独创建一个锁对象，浪费空间。

## 3.7 获取元素 ##

### 3.7.1 get(Object key) ###

public V get(Object key) { 
            Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
            int h = spread(key.hashCode());
            // 数组不为null且对应的桶中存在元素
            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 小于0，说明是树节点或者正在扩容
                // 使用find寻找元素，find的寻找方式依据 Node 的不同子类有不同的实现方式
                // 如果 Node 是树节点，则以树的方式去获取
                // 如果 Node 是 ForwardingNode ,则去扩容后的数组中查找
                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）`hash`到元素所在的桶；

（2）如果桶中第一个元素就是该找的元素，直接返回；

（3）如果是树或者正在迁移元素，则调用各自`Node`子类的`find()`方法寻找元素；

（4）如果是链表，遍历整个链表寻找元素；

（5）获取元素没有加锁；

## 3.8 删除元素 ##

### 3.8.1 remove(Object key) ###

public V remove(Object key) { 
            return replaceNode(key, null, null);
        }
    
        final V replaceNode(Object key, V value, Object cv) { 
            int hash = spread(key.hashCode());
            for (Node<K,V>[] tab = table;;) { 
                Node<K,V> f; int n, i, fh;
                // 如果数组没有被初始化，或对应桶没元素
                if (tab == null || (n = tab.length) == 0 ||
                    (f = tabAt(tab, i = (n - 1) & hash)) == null)
                    break;
               	// 如果数组正在扩容，当前线程帮助扩容 
                else if ((fh = f.hash) == MOVED)
                    tab = helpTransfer(tab, f);
                else { 
                    V oldVal = null;
                    boolean validated = false;
                    // 以数组第一个元素为锁对象，加锁
                    synchronized (f) { 
                        if (tabAt(tab, i) == f) { 
                        	// 如果是链表，则以链表的方式删除元素
                            if (fh >= 0) { 
                                validated = true；
                                for (Node<K,V> e = f, pred = null;;) { 
                                    K ek;
                                    if (e.hash == hash &&
                                        ((ek = e.key) == key ||
                                         (ek != null && key.equals(ek)))) { 
                                        V ev = e.val;
                                        if (cv == null || cv == ev ||
                                            (ev != null && cv.equals(ev))) { 
                                            oldVal = ev;
                                            if (value != null)
                                                e.val = value;
                                            else if (pred != null)
                                                pred.next = e.next;
                                            else
                                                setTabAt(tab, i, e.next);
                                        }
                                        break;
                                    }
                                    pred = e;
                                    if ((e = e.next) == null)
                                        break;
                                }
                            }
                            // 如果是树，则以树的方式删除元素
                            else if (f instanceof TreeBin) { 
                                validated = true;
                                TreeBin<K,V> t = (TreeBin<K,V>)f;
                                TreeNode<K,V> r, p;
                                if ((r = t.root) != null &&
                                    (p = r.findTreeNode(hash, key, null)) != null) { 
                                    V pv = p.val;
                                    if (cv == null || cv == pv ||
                                        (pv != null && cv.equals(pv))) { 
                                        oldVal = pv;
                                        if (value != null)
                                            p.val = value;
                                        else if (t.removeTreeNode(p))
                                            setTabAt(tab, i, untreeify(t.first));
                                    }
                                }
                            }
                        }
                    }
                    if (validated) { 
                        if (oldVal != null) { 
                            if (value == null)
                            	// 键值对个数减1
                                addCount(-1L, -1);
                            return oldVal;
                        }
                        break;
                    }
                }
            }
            return null;
        }

（1）计算`hash`；

（2）如果所在的桶不存在，表示没有找到目标元素，返回；

（3）如果正在扩容，则协助扩容完成后再进行删除操作；

（4）如果是以链表形式存储的，则遍历整个链表查找元素，找到之后再删除；

（5）如果是以树形式存储的，则遍历树查找元素，找到之后再删除；

（6）如果是以树形式存储的，删除元素之后树较小，则退化成链表；

（7）如果确实删除了元素，则整个`map`元素个数减1，并返回旧值；

（8）如果没有删除元素，则返回`null`；

## 3.9 扩容条件 ##

在`HashMap`中，当键值对的个数大于阈值时，则扩容，而在 `Concurrent`中，我们是通过 `addCount` 方法更改键值对的个数，所以是否需要扩容也是在该方法中进行判断的。

### 3.9.1 addCount(long x, int check) ###

private final void addCount(long x, int check) { 
        	 // 把数组的大小存储根据不同的线程存储到不同的段上（也是分段锁的思想）
        	// 并且有一个baseCount，优先更新baseCount，如果失败了再更新不同线程对应的段
        	// 这样可以保证尽量小的减少冲突
        	
            CounterCell[] as; long b, s;
            if ((as = counterCells) != null ||
            	 // 先尝试把数量加到baseCount上，如果失败再加到分段的CounterCell上
                !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 =
                      // 通过 cas 的方式更改 CounterCell 中的值
                      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) { 
                    	// sc<0说明正在扩容中
                        if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                            sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                            transferIndex <= 0)
                            break;
                         // 扩容未完成，则当前线程加入迁移元素中
                    	 // 并把扩容线程数加1 
                        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();
                }
            }
        }

（1）元素个数的存储方式类似于`LongAdder`类，存储在不同的段上，减少不同线程同时更新`size`时的冲突；

（2）计算元素个数时把这些段的值及`baseCount`相加算出总的元素个数；

（3）正常情况下`sizeCtl`存储着扩容门槛，扩容门槛为容量的`0.75`倍；

（4）扩容时sizeCtl高位存储扩容邮戳(`resizeStamp`)，低位存储扩容线程数加1（1+nThreads）；

（5）其它线程添加元素后如果发现存在扩容，也会加入的扩容行列中来；

## 3.10 扩容 ##

### 3.10.1 tryPresize(int size) ###

private final void tryPresize(int size) { 
        	// 如果给定的size大于等于数组容量一半，直接使用最大容量
        	// 否则，返回size * 0.75 + 1 向上的二次幂的值
            int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
                tableSizeFor(size + (size >>> 1) + 1);
            int sc;
            while ((sc = sizeCtl) >= 0) { 
                Node<K,V>[] tab = table; int n;
                if (tab == null || (n = tab.length) == 0) { 
                    n = (sc > c) ? sc : c;
                    // 通过 CAS 的方式更改 sizeCtl 的值
                    if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { 
                        try { 
                            if (table == tab) { 
                                @SuppressWarnings("unchecked")
                                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                                table = nt;
                                sc = n - (n >>> 2);
                            }
                        } finally { 
                            sizeCtl = sc;
                        }
                    }
                }
                /* * 一直扩容到的c小于等于sizeCtl或者数组长度大于最大长度的时候，则退出 */
                else if (c <= sc || n >= MAXIMUM_CAPACITY)
                    break;
                else if (tab == table) { 
                    int rs = resizeStamp(n);
                    // 如果数组正在扩容，则当前线程帮助扩容
                    if (sc < 0) { 
                        Node<K,V>[] nt;
                        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);
                }
            }
        }

## 3.11 协助扩容（迁移元素） ##

线程添加元素时发现正在扩容且当前元素所在的桶元素已经迁移完成了，则协助迁移其它桶的元素。

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { 
        Node<K,V>[] nextTab; int sc;
        // 如果桶数组不为空，并且当前桶第一个元素为ForwardingNode类型，并且nextTab不为空
        // 说明当前桶已经迁移完毕了，才去帮忙迁移其它桶的元素
        // 扩容时会把旧桶的第一个元素置为ForwardingNode，并让其nextTab指向新桶数组
        if (tab != null && (f instanceof ForwardingNode) &&
                (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { 
            int rs = resizeStamp(tab.length);
            // sizeCtl<0，说明正在扩容
            while (nextTab == nextTable && table == tab &&
                    (sc = sizeCtl) < 0) { 
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || transferIndex <= 0)
                    break;
                // 扩容线程数加1
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { 
                    // 当前线程帮忙迁移元素
                    transfer(tab, nextTab);
                    break;
                }
            }
            return nextTab;
        }
        return table;
    }

当前桶元素迁移完成了才去协助迁移其它桶元素；

## 3.11 迁移元素 ##

### 3.11.1 transfer(Node<K,V>\[\] tab, Node<K,V>\[\] nextTab) ###

扩容时容量变为两倍，并把部分元素迁移到其它桶中。

/** * 将每个节点移动到新表。 */
        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
           
             // 如果复制的目标nextTab为null的话，则初始化一个table两倍长的nextTab 
            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;
             // 新建一个ForwardingNode类型的节点，并把新桶数组存储在里面
        	// 这个是用来控制并发的，当一个节点为空或已经被转移之后，就设置为fwd节点
        	// 这是一个空的标志节点
            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;
                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;
                    }
                }
                if (i < 0 || i >= n || i + n >= nextn) { 
                	// 如果一次遍历完成了
               		// 也就是整个map所有桶中的元素都迁移完成了
                    int sc;
                    if (finishing) { 
                    	// 如果全部迁移完成了，则替换旧桶数组
                    	// 并设置下一次扩容门槛为新桶数组容量的0.75倍
                        nextTable = null;
                        table = nextTab;
                        sizeCtl = (n << 1) - (n >>> 1);
                        return;
                    }
                    if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { 
                    	// 当前线程扩容完成，把扩容线程数-1
                        if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                            return;
                        finishing = advance = true;
                        i = n; // recheck before commit
                    }
                }
                // 数组中把null的元素设置为ForwardingNode节点(hash值为MOVED[-1])
                else if ((f = tabAt(tab, i)) == null)
                    advance = casTabAt(tab, i, null, fwd);
                else if ((fh = f.hash) == MOVED)
                	// 如果桶中第一个元素的hash值为MOVED
                    // 说明它是ForwardingNode节点
                    // 也就是该桶已迁移
                    // 但并不代表整个数组都迁移完成
                    advance = true; // already processed
                else { 
                	// 锁定该桶并迁移元素
                    synchronized (f) { 
                        if (tabAt(tab, i) == f) { 
                            Node<K,V> ln, hn;
                            // 把一个链表分化成两个链表
                       		// 规则是桶中各元素的hash与桶大小n进行与操作
                        	// 等于0的放到低位链表(low)中，不等于0的放到高位链表(high)中
                        	// 其中低位链表迁移到新桶中的位置相对旧桶不变
                        	// 高位链表迁移到新桶中位置正好是其在旧桶的位置加n
                            if (fh >= 0) { 
                                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);
                                }
                                // 遍历链表，把hash&n为0的放在低位链表中
                                // 不为0的放在高位链表中
                                // 低位链表的位置不变
                                setTabAt(nextTab, i, ln);
                                setTabAt(nextTab, i + n, hn);
                                setTabAt(tab, i, fwd);
                                advance = true;
                            }
                            // 如果第一个元素是树节点
                            // 也是一样，分化成两颗树
                            // 也是根据hash&n为0放在低位树中
                            // 不为0放在高位树中
                            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;
                               // 遍历整颗树，根据hash&n是否为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;
                                    }
                                }
                                // 如果分化的树中元素个数小于等于6，则退化成链表
                                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);
                                // 高位树的位置是原位置加n
                                setTabAt(nextTab, i + n, hn);
                                 // 标记该桶已迁移
                                setTabAt(tab, i, fwd);
                                advance = true;
                            }
                        }
                    }
                }
            }
        }

（1）新桶数组大小是旧桶数组的两倍；

（2）迁移元素先从靠后的桶开始；

（3）迁移完成的桶在里面放置一`ForwardingNode`类型的元素，标记该桶迁移完成；

（4）迁移时根据`hash&n`是否等于`0`把桶中元素分化成两个链表或树；

（5）低位链表（树）存储在原来的位置；

（6）高们链表（树）存储在原来的位置加n的位置；

（7）迁移元素时会锁住当前桶，也是分段锁的思想；

## 3.12 获取元素个数 ##

### 3.12.1 size() ###

public int size() { 
        	// 调用sumCount()计算元素个数
            long n = sumCount();
            return ((n < 0L) ? 0 :
                    (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                    (int)n);
        }
    
    	final long sumCount() { 
    	    // 计算CounterCell所有段及baseCount的数量之和
    	    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;
    	}

（1）元素的个数依据不同的线程存在在不同的段里；（见addCounter()分析）

（2）计算`CounterCell`所有段及`baseCount`的数量之和；

（3）获取元素个数没有加锁；

# 3.13 总结 #

（1）`ConcurrentHashMap`是`HashMap`的线程安全版本；

（2）`ConcurrentHashMap`采用（数组 + 链表 + 红黑树）的结构存储元素；

（3）`ConcurrentHashMap`相比于同样线程安全的`HashTable`，效率要高很多；

（4）`ConcurrentHashMap`采用的锁有`synchronized`，`CAS`，自旋锁，分段锁，`volatile`等；

（5）`ConcurrentHashMap`中没有`threshold`和`loadFactor`这两个字段，而是采用`sizeCtl`来控制；

（6）`sizeCtl = -1`，表示正在进行初始化；

（7）`sizeCtl = 0`，默认值，表示后续在真正初始化的时候使用默认容量；

（8）`sizeCtl > 0`，在初始化之前存储的是传入的容量，在初始化或扩容后存储的是下一次的扩容门槛；

（9）`sizeCtl = (resizeStamp << 16) + (1 + nThreads)`，表示正在进行扩容，高位存储扩容邮戳，低位存储扩容线程数加1；

（10）更新操作时如果正在进行扩容，当前线程协助扩容；

（11）更新操作会采用`synchronized`锁住当前桶的第一个元素，这是分段锁的思想；

（12）整个扩容过程都是通过`CAS`控制`sizeCtl`这个字段来进行的，这很关键；

（13）迁移完元素的桶会放置一个`ForwardingNode`节点，以标识该桶迁移完毕；

（14）元素个数的存储也是采用的分段思想，类似于`LongAdder`的实现；

（15）元素个数的更新会把不同的线程`hash`到不同的段上，减少资源争用；

（16）元素个数的更新如果还是出现多个线程同时更新一个段，则会扩容段（`CounterCell`）；

（17）获取元素个数是把所有的段（包括`baseCount`和`CounterCell`）相加起来得到的；

（18）查询操作是不会加锁的，所以`ConcurrentHashMap`不是强一致性的；

（19 `ConcurrentHashMap`中不能存储`key`或`value`为`null`的元素；

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70]: /images/20210813/a7dbb5cf082449618f9ce1a2b84204bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 1]: /images/20210813/4d1fbcd805f84012a3bc75552f8fcdd4.png