ConcurrentHashMap源码深度解析（一）（java8）——不可不知的基本概念（助你拿下源码事半功倍）

- 日理万妓 2022-11-09 11:25 113阅读 0赞

### 文章目录 ###

*   *  一、前言
     *  二、基本定义
     *   *  1、基本常量
         *  2、基本属性
     *  三、构造器优化
     *   *  1、ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)
         *  2、ConcurrentHashMap(int initialCapacity)
         *  3、tableSizeFor
     *  四、不可不知的节点定义
     *   *  1、普通节点Node
         *  2、TreeBin+TreeNode
         *   *  （1）TreeBin 根节点
             *  （2）TreeNode
         *  3、转发节点ForwardingNode
     *  五、哈希高低位扰动spread
     *  六、总结

## 一、前言 ##

前一期对[`ConcurrentHashMap`源码java7版本][ConcurrentHashMap_java7]做了深度解析，数组+链表、分段锁，工业级的哈希表，但是也有一些非常明显的缺点，比如：

*  `Segment`数组一旦初始化不可扩容，这就为后期高并发性能提升埋下不可避免的瓶颈。
 *  以链表的方式解决哈希冲突，链表的检索时间复杂度为O(n)，在哈希冲突剧烈，链表越来越长的情况下性能会急剧下降。
 *  … …

所以在java8版本，作者Doug Lea对`ConcurrentHashMap`做了翻天覆地的改动，在很多方面都做了优化，比如：

*  数据结构采用数组+链表+红黑树，废弃分段锁`Segement`，进一步降低锁的粒度，可将锁直接加在数组占位节点上。同时发生哈希冲突的节点依然采用链表法，但是加了红黑树进行检索优化，即链表与红黑树互相转化，即使到了极端情况，检索的时间复杂度为O(lgn)，相对于O(n)性能提升不少。
 *  构造器传入初始化容量时，会对其根据扩容因子进行预估计算，计算出一个更合理的初始容量，避免不必要的扩容。
 *  哈希函数优化，高低位扰动，进一步降低哈希冲突。
 *  扩容机制优化，java7因为每个`Segment`里独立扩容，天然的线程安全和隔离环境，java8之后废弃了`Segment`，扩容就是扩容整个数组，如何实现安全且高效的扩容，相对于java7有些复杂。作者采用了多线程辅助扩容机制，以及在节点迁移上也做了优化。
 *  … …

java8 `ConcurrentHashMap`数据结构图示：

![ConcurrentHashMap java8][]

## 二、基本定义 ##

源码中的常量和属性非常多，且反复出现，如不预先了解其含义，在读源码时就感觉如鲠在喉。读源码不能逐字逐句读，讲究一个泛读、跳读和精度，泛读和跳读就是为了尽快了解一些定义的作用以及大概的逻辑。所以先来看看`ConcurrentHashMap`有哪些基本常量和属性，顾名思义的就不单独解释了，如`MAXIMUM_CAPACITY`、`DEFAULT_CAPACITY`等。

### 1、基本常量 ###

java8废弃了分段锁`Segment`，但是为了兼容旧版本，依然保留了内部类`Segment`以及一些相关的常量如`DEFAULT_CONCURRENCY_LEVEL`、`LOAD_FACTOR`等，但是用不上，java8中扩容因子固定是3/4（`n - (n >>> 2`)），不可更改。因为数据结构加了红黑树的元素以及在扩容机制上的优化，定义了一些常量：

*  `TREEIFY_THRESHOLD = 8`，链表转为红黑树的阈值，单条链表节点数量>= `TREEIFY_THRESHOLD`时链表可能转为红黑树，记住是可能。
 *  `UNTREEIFY_THRESHOLD = 6`，红黑树退化为链表的阈值，只作用于扩容阶段，在数据从旧数组迁移到新数组时，新组装的红黑树的节点数量<= `UNTREEIFY_THRESHOLD`时，红黑树退化为链表。至于为什么小于`TREEIFY_THRESHOLD`而不是等于，个人猜想是避免链表与红黑树间频繁转化影响性能。
 *  `MIN_TREEIFY_CAPACITY = 64`，数组长度< `MIN_TREEIFY_CAPACITY`时，即使达到链表转红黑树的阈值也不转换，而是扩容。
 *  `MIN_TRANSFER_STRIDE = 16`，扩容时，给线程分配迁移数组元素任务时的最小步长。
 *  `RESIZE_STAMP_BITS = 16`，扩容时，在`sizeCtl`中用于生成戳记的数值，`resizeStamp()`也会用到（不要慌，扩容时会细讲）。
 *  `RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS`，在`sizeCtl`中用于生成记录扩容线程个数的戳记的移位数。（不要慌，扩容时会细讲）
 *  `MOVED = -1`，代表数组正在扩容，且该位置的节点已经被迁移到新数组，也作为`ForwardingNode`节点的hash值。
 *  `TREEBIN = -2`，用于标记红黑树根节点的hash值。
 *  `HASH_BITS = 0x7fffffff`，正常hash值的可用位,在`spread`中用于保证计算的hash值不超过`HASH_BITS`（`spread()`会细讲）。

有些常量必须结合源码才能明白其用途，现在只需要留一个初步印象。

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
        implements ConcurrentMap<K,V>, Serializable { 
        private static final long serialVersionUID = 7249069246763182397L;
    
        /* ---------------- Constants -------------- */
    
        /** * The largest possible table capacity. This value must be * exactly 1<<30 to stay within Java array allocation and indexing * bounds for power of two table sizes, and is further required * because the top two bits of 32bit hash fields are used for * control purposes. */
        // 1 * 2^30 = 1073741824
        // 最大容量
        private static final int MAXIMUM_CAPACITY = 1 << 30;
    
        /** * The default initial table capacity. Must be a power of 2 * (i.e., at least 1) and at most MAXIMUM_CAPACITY. * 默认容量 */
        private static final int DEFAULT_CAPACITY = 16;
    
        /** * The default concurrency level for this table. Unused but * defined for compatibility with previous versions of this class. * 默认并发级别，为了兼容旧版本，java8没有用上 */
        private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    
        /** * The load factor for this table. Overrides of this value in * constructors affect only the initial table capacity. The * actual floating point value isn't normally used -- it is * simpler to use expressions such as {@code n - (n >>> 2)} for * the associated resizing threshold. * 默认扩容因子，但是用不上。 */
        private static final float LOAD_FACTOR = 0.75f;
    
        /** * The bin count threshold for using a tree rather than list for a * bin. Bins are converted to trees when adding an element to a * bin with at least this many nodes. The value must be greater * than 2, and should be at least 8 to mesh with assumptions in * tree removal about conversion back to plain bins upon * shrinkage. * 链表转为红黑树的阈值，>= TREEIFY_THRESHOLD链表可能转为红黑树 */
        static final int TREEIFY_THRESHOLD = 8;
    
        /** * The bin count threshold for untreeifying a (split) bin during a * resize operation. Should be less than TREEIFY_THRESHOLD, and at * most 6 to mesh with shrinkage detection under removal. * 红黑树转为链表的阈值， <= UNTREEIFY_THRESHOLD 时红黑树转为链表，只作用于扩容阶段 */
        static final int UNTREEIFY_THRESHOLD = 6;
    
        /** * 容量在64以内，即使达到链表转红黑树的阈值也不转换，而是扩容 * The smallest table capacity for which bins may be treeified. * (Otherwise the table is resized if too many nodes in a bin.) * The value should be at least 4 * TREEIFY_THRESHOLD to avoid * conflicts between resizing and treeification thresholds. */
        static final int MIN_TREEIFY_CAPACITY = 64;
    
        /** * Minimum number of rebinnings per transfer step. Ranges are * subdivided to allow multiple resizer threads. This value * serves as a lower bound to avoid resizers encountering * excessive memory contention. The value should be at least * DEFAULT_CAPACITY. * 扩容时，给线程分配迁移数组元素任务时的最小步长 */
        private static final int MIN_TRANSFER_STRIDE = 16;
    
        /** * The number of bits used for generation stamp in sizeCtl. * Must be at least 6 for 32bit arrays. * 扩容时，在sizeCtl中用于生成戳记的数值 */
        private static int RESIZE_STAMP_BITS = 16;
    
        /** * The maximum number of threads that can help resize. * Must fit in 32 - RESIZE_STAMP_BITS bits. * 帮助扩容的最大线程数 */
        private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
    
        /** * The bit shift for recording size stamp in sizeCtl. * 在sizeCtl中用于生成记录扩容线程个数的戳记的移位数 */
        private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
    
        // MOVED=-1 代表数组正在扩容，forwarding nodes的hash=-1
        // hash for forwarding nodes
        static final int MOVED     = -1;
        // 用于标记 红黑树根节点的hash
        static final int TREEBIN   = -2; // hash for roots of trees
        // 在computeIfAbsent and compute中用到可以先不用管
        static final int RESERVED  = -3; // hash for transient reservations
        // 正常hash值的可用位,在spread中用于保证计算的hash值不超过HASH_BITS
        // 2147483647
        // 1111 1111 1111 1111 1111 1111 1111 111
        static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash

### 2、基本属性 ###

有些基本属性也需要预先了解下，后期读源码时才会更顺利一些：

*  `table`，代表当前数组。
 *  `nextTable`，扩容后的新数组，只有在数组扩容时不为null。若`get`操作时，在旧数组`table`找不到节点，对应位置上又有转发节点时，会将`get`操作转发到`nextTable`。
 *  `transferIndex`，记录了扩容任务分配的进度。初始为n，逆序扩容，每次减一个步长的位置，最终减至＜=0，表示整个扩容任务分配完了。
 *  `baseCount`，数组元素基础计数，在没有竞争的时候先cas修改`baseCount`。
 *  `CounterCell[] counterCells`，当cas修改`baseCount`失败后的线程会去修改对应`counterCells`数组中一个计数格子。所以想获取数组内元素的总个数就是`baseCount+counterCells数组内所有计数格记录值之和`。
 *  `cellsBusy`，简易自旋锁，用于`counterCells`数组中保证多线程更新数组元素个数线程安全。

`sizeCtl`的定义较为复杂，但是很重要，不同的值在数组不同状态中起着举足轻重的作用：

*  数组未初始化时，`sizeCtl`被赋值初始容量，以待初始化数组时使用。
 *  数组正在初始化时，`sizeCtl=-1`，相当于一把锁，控制只有一个线程进去初始化数组操作。
 *  数组初始化完成，`sizeCtl`被赋值扩容阈值，以待触发扩容机制时判断。
 *  数组扩容时，`sizeCtl`被赋值一个非常小的负数，控制扩容线程数量的加减以及用来标识数组正在扩容的状态。

/* ---------------- Fields -------------- */
    
    /** * The array of bins. Lazily initialized upon first insertion. * Size is always a power of two. Accessed directly by iterators. * 当前数组 */
    transient volatile Node<K,V>[] table;
    
    /** * 扩容后的新数组，只有在数组扩容时不为null。 * The next table to use; non-null only while resizing. */
    private transient volatile Node<K,V>[] nextTable;
    
    /** * Base counter value, used mainly when there is no contention, * but also as a fallback during table initialization * races. Updated via CAS. * 数组元素基础计数 */
    private transient volatile long baseCount;
    
    /** * Table initialization and resizing control. When negative, the * table is being initialized or resized: -1 for initialization, * else -(1 + the number of active resizing threads). Otherwise, * when table is null, holds the initial table size to use upon * creation, or 0 for default. After initialization, holds the * next element count value upon which to resize the table. * sizeCtl很重要，不同的值在数组不同状态中起着举足轻重的作用。 */
    private transient volatile int sizeCtl;
    
    /** * transferIndex记录了扩容的进度。 * The next table index (plus one) to split while resizing. */
    private transient volatile int transferIndex;
    
    /** * 简易自旋锁，用于控制多线程统计数组元素的 * 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. * 当cas修改baseCount失败后的线程会去修改对应counterCells数组中一个计数格子。 * 所以 想获取数组内元素的总个数就是baseCount+counterCells数组内所有计数格记录值之和。 */
    private transient volatile CounterCell[] counterCells;

## 三、构造器优化 ##

java8中的构造器比java7简单很多，不需要初始化各种数据，也没有初始化数组，只是设置了一个初始容量，但是对如何设置一个合理的初始容量做了优化。

java7对传入的`initialCapacity`除以`segment`数组的长度，然后简单找到一个大于等于且离平均值最近的2的整数次的数作为内部`HashEntry`数组的初始容量。而java8认为传入的`initialCapacity`是使用者预估后面想要添加的元素个数（可能是短期会添加这么多元素），如果预估元素个数已经大于等于或者接近扩容阈值，这样就很容易触发扩容机制。所以java8对初始容量根据扩容阈值做了优化。

### 1、ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) ###

先看这个从外部可以传入`loadFactor`和`concurrencyLevel`的构造器，这个构造器应该是为了兼容旧版本，因为java8已经没有了显式设置扩容因子的定义，扩容因子固定是3/4，也没有了`concurrencyLevel`并发级别的概念。而这里传入的`loadFactor`只在初始化时计算初始容量时有用，不会修改后期的扩容因子（3/4）。

`initialCapacity`认为是使用者后面短期内想要添加的元素个数，该如何找到一个合适的初始容量避免不必要的扩容呢？

假设计算的初始化容量为`size`，那扩容阈值就是`size*loadFactor`，只要扩容阈值大于预估容量`initialCapacity`就不会触发扩容，有如下不等式关系：

`size * loadFactor > initialCapacity`，转换一下就是`size > initialCapacity / loadFactor`。

假设不等式左边只比右边大1，那么只要右边加1就可以使得左右两边相等，即：`size = (initialCapacity / loadFactor) +1`。这样得到size就是一个比较合理的初始容量，但是为了size是一个大于等于且离size最近的2的整数次方的数，还需要经过`tableSizeFor`的处理。

public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) { 
        if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
        // concurrencyLevel的用处不大，为了兼容java7版本
        if (initialCapacity < concurrencyLevel)   // Use at least as many bins
            initialCapacity = concurrencyLevel;   // as estimated threads
        // initialCapacity和loadFactor都是使用者外部传入，所以可对initialCapacity进行优化
        long size = (long)(1.0 + (long)initialCapacity / loadFactor);
        // 获取一个大于等于且离size最近的2的整数次方的数
        int cap = (size >= (long)MAXIMUM_CAPACITY) ?
            MAXIMUM_CAPACITY : tableSizeFor((int)size);
        // 此时sizeCtl存的是初始容量
        this.sizeCtl = cap;
    }

### 2、ConcurrentHashMap(int initialCapacity) ###

只传一个参数`initialCapacity`，这个构造器应该是平时较为常用的，还有一个无参构造器，默认初始容量是16。

优化的细节和上一个构造器差不多，扩容因子是固定的3/4，假设初始容量是`size`，则一元一次方程：

`size = initialCapacity * 4/3 + 1` ，即`size =(initialCapacity + initialCapacity * 1/3) +1`。

但是代码中为什么没有这么做呢？而是`size = (initialCapacity+ initialCapacity*1/2) + 1`，这样阈值就是2/3，不是3/4了。

个人猜想：为了追求计算性能，1/3无法使用位运算，1/2可以用位运算>>>1，且1/2比1/3稍微大一些，计算出的`size`不会与理想值偏差太大而浪费空闲资源。

public ConcurrentHashMap(int initialCapacity) {
        if (initialCapacity < 0)
            throw new IllegalArgumentException();
        int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                   MAXIMUM_CAPACITY :
                   tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
        // 1.5倍的初始容量+1，再往上取最接近的2的整数次方
        // 初始化时只是设置了初始值，并没有初始化数组，懒加载，put时在初始化数组
        this.sizeCtl = cap;
    }

### 3、tableSizeFor ###

有必要讲一下`tableSizeFor`，这个函数的作用很简单，就是获取大于等于且离传入值最近的2的整数次的数。其运算过程是使一个数二进制最左边的1右边位全部转化为1，然后+1就可得一个2的整数次方的数。

/** * 找到大于等于c的最近2的整数次的数 * @param c * @return */
    private static final int tableSizeFor(int c) { 
        // 为什么要减1呢?
        int n = c - 1;
        // 向右移1位，与旧值|，逻辑上可得到2个1
        n |= n >>> 1;
        // 向右移2位，与旧值|，逻辑上可得到4个1
        n |= n >>> 2;
        // 向右移4位，与旧值|，逻辑上可得到8个1
        n |= n >>> 4;
        // 向右移8位，与旧值|，逻辑上可得到16个1
        n |= n >>> 8;
        // 向右移16位，与旧值|，逻辑上可得到32个1
        n |= n >>> 16;
        // n+1 得到大于等于c的最近2的整数次的数
        return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
    }

计算过程如图：  
![tableSizeFor][]  
**为何传入的c还要减1呢？**

假设c=16 二进制：10000，c-1=15，二进制：1111经过一顿右移和旧值的|运算，得到的还是1111，+1还是16。

而c不-1，直接拿10000做运算，得到的是11111，+1是32，这就不太对了，明明传进去的就是一个2的整数次方的数，得到的确实2倍c。所以为了兼容这种情况 传进来的c都减1。

## 四、不可不知的节点定义 ##

在阅读put、get等源码逻辑前，还必须了解以下几种节点的定义：

1.  构成链表的普通节点`Node`。
2.  构成红黑树的`TreeBin+TreeNode`。
3.  转发节点`ForwardingNode`。

### 1、普通节点Node ###

和java7中`HashEntry`定义差不多，是构成链表的基本元素。hash >= 0，后面代码中会根据hash>=0 判断为普通节点，即链表。

static class Node<K,V> implements Map.Entry<K,V> { 
        final int hash;
        final K key;
        volatile V val;
        volatile Node<K,V> next;
        ... ...
        /** * Virtualized support for map.get(); overridden in subclasses. * 遍历链表 寻找哈希和key都相同的节点 */
        Node<K,V> find(int h, Object k) { 
            Node<K,V> e = this;
            if (k != null) { 
                do { 
                    K ek;
                    if (e.hash == h &&
                        ((ek = e.key) == k || (ek != null && k.equals(ek))))
                        return e;
                } while ((e = e.next) != null);
            }
            return null;
        }
    }

### 2、TreeBin+TreeNode ###

构成红黑树的节点有两种，`TreeBin`是根节点，也是一个空节点，不存任何`key-value`，`TreeNode`是真正存`key-value`的节点。

#### （1）TreeBin 根节点 ####

`TreeBin`虽然不存任何元素，但是肩负管理红黑树的职责：向红黑树添加、删除节点，查找节点等。如何构成红黑树以及如何维护红黑树的特性，不是本篇的重点，后期会单独拎出来研究。

一个节点新增时首先会以头插法的方式串成一个链表，然后另外再维护一棵红黑树的形态。而依然维护链表的结构主要是为了当红黑树在做平衡算法时，依然可以用遍历链表的方式遍历节点。

`TreeBin`内部简单维护了一把自旋读写锁，目的是在当新增和删除节点时，需要维护红黑树的结构特性，这个平衡算法的过程需要加锁。而新增节点以头插法的方式串成链表，所以修改不会影响遍历过程且`next`指针被`volatile`修饰，修改指针后会立即通知到所有线程获取最新值。

读写锁比较有意思，`lockState`记录锁的状态，有三种标志位：

*  `WRITER=1`，二进制低位第一位用来标识写线程持有锁的状态 （不可重入写锁）。
 *  `WAITER=2`，二进制低位第二位用来标识阻塞状态。
 *  `WAITER=4`，二进制低位第三位之后都是用来标识读线程持有锁的状态。读读不互斥，`lockState+READER`代表一个读线程获取锁，`lockState-READER`代表一个读线程释放锁。读锁释放时如果有线程正在等待阻塞（写线程），则唤醒。

还有一个点需要强调，`TreeBin`作为一棵红黑树的根节点也就是头节点，同是数组中占位的节点，其hash值为`TREEBIN=-2`，后面代码会根据`hash< 0 && f instanceof TreeBin` 判断是红黑树。

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
    
        /** * Creates bin with initial set of nodes headed by b. * * hash=TREEBIN < 0 && f instanceof TreeBin 可判断是红黑树 */
        TreeBin(TreeNode<K,V> b) { 
            super(TREEBIN, null, null, null);
            this.first = b;
            ... ...
        }
    
        /** * Acquires write lock for tree restructuring. */
        private final void lockRoot() { 
            if (!U.compareAndSwapInt(this, LOCKSTATE, 0, WRITER))
                // 获取写锁失败，则竞争加锁，可能会阻塞
                contendedLock(); // offload to separate method
        }
    
        /** * Releases write lock for tree restructuring. */
        private final void unlockRoot() { 
            lockState = 0;
        }
    
        /** * Possibly blocks awaiting root lock. */
        private final void contendedLock() { 
            boolean waiting = false;
            for (int s;;) { 
                // ~WAITER = - (WAITER + 1) 11111111111111111111111111111101
                if (((s = lockState) & ~WAITER) == 0) { 
                    // 没有线程还有锁，lockState=0 or 2，则获取写锁
                    if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) { 
                        if (waiting)
                            waiter = null;
                        return;
                    }
                }
                // != 0, s有可能=1 or 4，8,12... 即有线程持有写锁or读锁，则当前线程需要阻塞
                else if ((s & WAITER) == 0) { 
                    if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) { 
                        waiting = true;
                        waiter = Thread.currentThread();
                    }
                }
                else if (waiting)
                    LockSupport.park(this);
            }
        }
    
        /** * Returns matching node or null if none. Tries to search * using tree comparisons from root, but continues linear * search when lock not available. */
        final Node<K,V> find(int h, Object k) { 
            if (k != null) { 
                for (Node<K,V> e = first; e != null; ) { 
                    int s; K ek;
                    // 二进制的低位1位是标识写锁，2位标识阻塞
                    if (((s = lockState) & (WAITER|WRITER)) != 0) { 
                        // 可能有线程持有写锁or阻塞，说明正在做平衡算法，不能使用红黑树来遍历节点
                        // 但是可以像普通链表一样遍历节点
                        if (e.hash == h &&
                            ((ek = e.key) == k || (ek != null && k.equals(ek))))
                            return e;
                        e = e.next;
                    }
                    // == 0 说明没有线程持有写锁和阻塞，则可获取读锁
                    // 读线程间是不互斥的，所以读线程累加READER
                    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;
                            // 释放读锁，-READER，释放锁的同时，若有线程在阻塞则唤醒他（一般就是写线程在等待）
                            // 这里会有不必要的唤醒，因为若不是最后一个读线程释放锁唤醒阻塞的写线程的话，
                            // 此时还有读线程持有锁，则写线程继续阻塞。
                            if (U.getAndAddInt(this, LOCKSTATE, -READER) ==
                                (READER|WAITER) && (w = waiter) != null)
                                LockSupport.unpark(w);
                        }
                        return p;
                    }
                }
            }
            return null;
        }
    
        /** * 返回值是已经存在的节点，所以可以根据返回值判断是替换还是新增 * Finds or adds a node. * @return null if added */
        final TreeNode<K,V> putTreeVal(int h, K k, V v) { 
            Class<?> kc = null;
            ... ...
        }
    
        /** * 返回值 true为当前树比较小，需要退化为链表 * @return true if now too small, so should be untreeified */
        final boolean removeTreeNode(TreeNode<K,V> p) { 
            ... ...
        }
    
        /* ------------------------------------------------------------ */
        // Red-black tree methods, all adapted from CLR
    	/** * 平衡算法过程中的左旋 */
        static <K,V> TreeNode<K,V> rotateLeft(TreeNode<K,V> root,
                                              TreeNode<K,V> p) { 
            ... ...
        }
    	/** * 平衡算法过程中的右旋 */
        static <K,V> TreeNode<K,V> rotateRight(TreeNode<K,V> root,
                                               TreeNode<K,V> p) { 
            ... ...
        }
    
        /** * 插入平衡算法，每插入一个节点就要维护红黑树的结构特性，而这个过程是需要上锁的 * @param root * @param x * @param <K> * @param <V> * @return */
        static <K,V> TreeNode<K,V> balanceInsertion(TreeNode<K,V> root,
                                                    TreeNode<K,V> x) { 
            ... ...
        }
    
        static <K,V> TreeNode<K,V> balanceDeletion(TreeNode<K,V> root,
                                                   TreeNode<K,V> x) { 
          ... ...
        }
    }

#### （2）TreeNode ####

实质存储元素的树节点。

/** * 红黑树中保存key-value的节点 * 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;
    
        TreeNode(int hash, K key, V val, Node<K,V> next,
                 TreeNode<K,V> parent) { 
            super(hash, key, val, next);
            this.parent = parent;
        }
    
        Node<K,V> find(int h, Object k) { 
            return findTreeNode(h, k, null);
        }
    
        /** * Returns the TreeNode (or null if not found) for the given key * starting at given root. */
        final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) { 
            // 遍历红黑树，找节点
            ... ...
        }
    }

### 3、转发节点ForwardingNode ###

`ForwardingNode`是一个空节点，也是一个临时占位节点。其主要有两个作用：

*  占位标识，用于标识数组该位置的元素已经迁移完毕，但还处于扩容状态。
 *  转发检索，查找操作在旧数组找不到元素节点，如若遇到`ForwardingNode`就会被转发到新数组中继续寻找。

`key`、`value`、`next`属性均为 null ，`nextTable`指向扩容后的新数组，hash值为`MOVED=-1`，后面代码中会根据`hash=MOVED`判断该占位节点为`ForwardingNode`，即数组正在扩容，该位置的元素已经被迁移。

get操作遇到`ForwardingNode`是转发，put操作遇到`ForwardingNode`是帮助扩容。

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;
        }
        // 转发到nextTable中继续检索
        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)
                    // 新数组 映射的槽是空的则返回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;
                            // 又遇到另一个转发节点，跳过一次外围循环，从新的tab检索，
                            // 不会在扩容阶段又在新数组上扩容把？有待后续验证
                            continue outer;
                        }
                        else
                            // 这里就是红黑树了，去树上找
                            return e.find(h, k);
                    }
                    if ((e = e.next) == null)
                        // 到最后了还没找到则返回null
                        return null;
                }
            }
        }
    }

## 五、哈希高低位扰动spread ##

java7中哈希映射`Segment`数组时使用的是哈希值的高位，映射`Segment`内部`HashEntry`数组时用的是哈希值的低位，哈希值的高低位都用上了，这样在一定程度上可以降低哈希冲突。

而java8中只有一种数组了，不管使用哈希值的高位和低位都会使其一部分丧失作用。所以作者就将哈希值的高16位和低16位做混合，目的是使得低16位也具有高16的特性，使得哈希值更不易发生冲突。

例如两个key的哈希值低16位比较相似，而高16位非常不相同，这样使用低位参与数组映射就容易产生哈希冲突，而将高位特征散播到低位就能很好的降低冲突。

那如何做高低位扰动呢？

一般计算一个key的哈希值如下，会调用一个`spread`函数。

int h = spread(key.hashCode());

static final int spread(int h) { 
        // 高低位扰动
        return (h ^ (h >>> 16)) & HASH_BITS;
    }

首先h右移16位，然后和旧h做^异或运算。而为什么不可以做&与运算和|或运算呢？这就讲到一个概率的问题：

*  & : 按位与，`1 & 1 = 1，1 & 0 = 0，0 & 1 = 0，0 & 0 = 0` ，1的概率是1/4，0的概率是3/4；
 *  | : 按位或，`1 | 1 = 1，1 | 0 = 1，0 | 1 = 1，0 | 0 = 0` ，1的概率是3/4，0的概率是1/4；
 *  ^ : 按位异或，`1 ^ 1 = 0，1 ^ 0 = 1，0 ^ 1 = 1，0 ^ 0 = 0`， 1的概率是1/2，0的概率是1/2。

^异或运算1和0出现的概率相等，所以用于扰动使得哈希值更均匀。

最后还和`HASH_BITS`做&运算，这又是为什么目的呢？

`HASH_BITS`是32位整数中最大的正整数，其二进制为：`1111 1111 1111 1111 1111 1111 1111 111`，再加1就变成了负数（真是一念成魔）。高低位扰动之后的哈希值与`HASH_BITS`做&运算可保证最终的哈希值不会溢出变成负数。

## 六、总结 ##

读源码不容易，精读源码更不容易，但是逐字逐句不可取，寸步难行，容易半途而废。不是所有的函数都需要弄明白，有些平时都没用过，或者不经常用的函数暂时就没有必要读。而能把`ConcurrentHashMap`基本的几个优化点以及扩容等非常重要的点搞明白搞透了，就可以了，已经很不容易了。有几个点需要再三强调：

1.  `ConcurrentHashMap` java8数据结构：数组+链表+红黑树。
2.  `ConcurrentHashMap` java8中废弃了`Segment`，连带并发级别，扩容因子等定义也只是留着为了兼容旧版本，扩容因子被固定为0.75，不可修改。
3.  `sizeCtl`很重要，再次强调：数组未初始化时，`sizeCtl>0`表示初始容量；初始化时，`sizeCtl=-1`可作为一把锁；初始化完成，`sizeCtl=n - (n >>> 2)`表示扩容阈值；扩容时，`sizeCtl<0`可用于记录扩容线程增减。
4.  构造器优化，传入的初始容量根据扩容机制（`(initialCapacity / loadFactor) +1`），预估出更科学的初始容量。
5.  节点判断：`hash>=0`是普通链表节点；`hash=MOVED=-1`是转发节点`ForwardingNode`，可判断是帮助扩容还是转发检索；`hash< 0 && f instanceof TreeBin`是红黑树根节点。
6.  key的哈希值高低位扰动，使低位也具备高位的特征，降低哈希冲突。

PS: **如若文章中有错误理解，欢迎批评指正，同时非常期待你的评论、点赞和收藏。我是徐同学，愿与你共同进步！**

[ConcurrentHashMap_java7]: https://stefan.blog.csdn.net/article/details/114478666
[ConcurrentHashMap java8]: /images/20221023/9b98fd4457c845248d9d964db6219a9a.png
[tableSizeFor]: /images/20221023/4fce8595491d4a0fa6e6265d7ba3c570.png