万字图文——ConcurrentHashMap源码深度解析

爪哇缪斯

发布于 2023-05-10 09:38:33

74500

代码可运行

文章被收录于专栏：爪哇缪斯爪哇缪斯

运行总次数：0

代码可运行

本次ConcurrentHashMap的源码解析是针对JDK8，内容比较多，具体大纲请见下方截图：

废话不多说了，我们进入正题把。

一、ConcurrentHashMap使用场景

我们平时最常用的HashMap其实不是线程安全的，而当我们有多线程使用场景的时候，即想线程安全，又想拥有Map的能力，我们可以选择HashTable，因为它是针对我们常用的方法上面加上了synchronize锁，但是在高并发的场景下，效率低是它的弊端。如果我们还非常在意效率，那么我们更好的选择是使用ConcurrentHashMap。
我们来看下一个例子，启动100个线程，每个线程循环100次，像容器中应该放入10000个元素。我们看到运行结果可以发现，HashMap并不是10000，这就说明，它在多线程并发的情况下，出现了线程不安全的问题。而ConcurrentHashMap返回的结果是没有问题的。

二、put方法的整体流程

我们先看put方法

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

在put方法中，调用了putVal方法。由于该方法代码较多，我们只保留框架性质的代码，这样会更方便我们理解。如下所示：

final V putVal(K key, V value, boolean onlyIfAbsent) {
        ... ...
        for (Node<K,V>[] tab = table;;) {
            ... ...
            if (tab == null || (n = tab.length) == 0) {
                tab = initTable();
            }
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) {
                    break; 
                }
            }
            else if ((fh = f.hash) == MOVED) {
                tab = helpTransfer(tab, f);
            }
            else {
                V oldVal = null;
                synchronized (f) {
                    ... 针对f链表或红黑树进行添加Node节点操作，执行完毕后break ...    
                }
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD) {
                        treeifyBin(tab, i);
                    }
                    if (oldVal != null) {
                        return oldVal;
                    }
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }

在上面的代码中，可以分为两部分内容：

第一部分：首先开启了无限循环，在里面进行了4中情况的判断。

case1：【如果table数组需要被创建】

如果table数组为null或者长度为0，则创建table数组。

case2：【如果寻址后的位置没有被占用】

创建Node节点，插入到这个位置

case3：【如果寻址后的位置是正在迁移状态】

加入到迁移大军中，帮助一起进行扩容迁移操作。

case4：【其他情况】

将节点插入到链表中或者红黑树中。

如果链表长度

第二部分：执行addCount，将ConcurrentHashMap中存储的k,v总数+1。

那么，下面我们就来逐一的对上述的四个case和addCount骤一进行分析。

三、case1：初始化table数组

源码如下所示：

3.1> sizeCtl

如果sizeCtl为-1，则表示table数组正在被别的线程初始化。默认sizeCtl=0，当table数组初始化或者扩容完毕的时候，sizeCtl会表示扩容阈值。

默认table数组的长度为16

3.2> 流程解释

我们通过源码可以看到，如果table没有被初始化完毕的话，那么会一直在while循环中，直到table数组初始化完毕：

while ((tab = table) == null || tab.length == 0)

上面已经介绍了sizeCtl的含义了，那么假设现在有4条线程同时的要去创建table数组，那么当有一条线程已经优先开始初始化table数组操作的时候，sizeCtl就会被赋值为-1，那么其他线程就会执行Thread.yield()让出cpu，并继续while循环，然后再执行Thread.yield()，在那spin旋转，直到那个最早的线程创建好创建table数组之后，所有线程都会跳出while继续往下执行。

private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0) {
                Thread.yield(); // 其他竞争失败的线程，会在while循环中spin，直到table创建完毕才能跳出while循环
            }
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                ... 只有一个线程可以成功执行CAS操作，将sizeCtl赋值为-1 ... 
                ... 执行创建table数组操作 ... 
            }
        }
        return tab; // 返回创建好的table数组
    }

四、case2：寻址后的位置没有被占用

源码如下所示：

其中，关键点是，我们要理解tabAt和casTabAt这两个方法的实现逻辑，那么在了解这两个方法之前，我们需要先了解两个变量的含义，即：ABASE和ASHIFT。那么下面我们来着重看一下这两个变量到底是干啥的。

4.1> ABASE&ASHIFT

源码部分：

Class<?> ak = Node[].class;

ABASE = U.arrayBaseOffset(ak);

基础偏移量：public native int arrayBaseOffset(Class<?> arrayClass)
- 返回值为16，因为数组对象是由对象头（8字节）+指针（4字节）+数组长度（4字节）组成的，所以从16开始。
- 返回数组类型的第一个元素的偏移地址（基础偏移地址）。
- 如果arrayIndexScale方法返回的比例因子不为0，你可以通过结合基础偏移地址和比例因子访问数组的所有元素。
int scale = U.arrayIndexScale(ak);
比例因子：public native int arrayIndexScale(Class<?> arrayClass)
- 返回数组类型的比例因子(其实就是数据中元素偏移地址的增量，因为数组中的元素的地址是连续的)。
ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
Integer.numberOfLeadingZeros(int i)
- 给定一个int类型数据，返回这个数据的二进制串中从最左边算起连续的“0”的总数量。
- 而ASHIFT其实就是将scale数值转换为按位左移对应的数值。
- 即：通过scale=4，那么计算ASHFIT=2，而N<<2其实就相当于N*2*2=N*4=N*scale
- 举例：

如果scale=8（十进制）=1000（二进制）

那么计算出Integer.numberOfLeadingZeros(scale)=28

ASHIFT=31-28=3

N<<ASHIFT = N<<3 = N*2*2*2 = N*8 = N*scale

4.2> tabAt(Node<K,V>[] tab, int i)

作用：获得tab数组下标为i位置上的Node元素
数组的寻址公式为：a[i]_address = base_address + i*data_type_size，通过该方式可以获得对应下标为i的值，即：获得tab[i]的值。
源码部分：

public native Object getObjectVolatile(Object o, long offset);
- 此方法和getObject功能类似，不过附加了volatile语义，也就是强制从主存中获取属性值。
- 类似的方法有getIntVolatile、getDoubleVolatile等等。
- 这个方法要被使用的属性由volatile修饰，否则功能和getObject方法相同。
- offset= ((long)i << ASHIFT) + ABASE，表示从ABASE开始，计算第i个元素的偏移量。
- 所以：tabAt(tab, i)就等同于tab[i]

4.3> casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v)

作用：如果tab数组下标为i的Node元素是c，则将c修改为v。
源码部分：

public final native boolean compareAndSwapObject(Object o, long offset, Object expected, Object x);

针对Object对象进行CAS操作。即是对应Java变量引用o，原子性地更新o中偏移地址为offset的属性的值为x，当且仅的偏移地址为offset的属性的当前值为expected才会更新成功返回true，否则返回false。

o：目标Java变量引用。
offset：目标Java变量中的目标属性的偏移地址。
expected：目标Java变量中的目标属性的期望的当前值。
x：目标Java变量中的目标属性的目标更新值。

类似的方法有compareAndSwapInt和compareAndSwapLong，在Jdk8中基于CAS扩展出来的方法有getAndAddInt、getAndAddLong、getAndSetInt、getAndSetLong、getAndSetObject，它们的作用都是：通过CAS设置新的值，返回旧的值。

4.4> 流程解释

这部分属于判断中的case2，通过hash值计算出来应该插入的下标i，如果这个位置是空的，即：还没有保存Node元素，那么就根据我们要put的key和value来创建一个新的Node，并插入到下标为i的位置上，即：

if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) {
    break;
}

这段采用CAS来保证只有一个线程可以赋值成功。如果我们还是有A，B，C，D这4个线程都执行到了这个判断语句中，假设线程A第一个执行的这个CAS操作，那么只有它会执行成功，其余的3个线程（B，C，D）则会执行失败，casTabAt的结果为false。那么线程A会执行break语句跳出for循环，而其他三个线程会再次执行for循环，并执行到case4的代码段中。

五、case4：其他情况

本来我们此处应该分析case3的，但是由于里面涉及的内容比较多，而关键的点都在case4中，所以我们先暂时跳过case3，先把case4的情况分析明白，这样我们转过头来在分析case3就会清晰简单很多。
那么case4部分，就是就是我们调用put方法的时候，最核心的处理逻辑了。因为当table表被初始化了，并且出现哈希冲突了，并且Node f这个位置并没有发生移动的情况下，都会走到这个代码段中。而这种情况又是最多发生的情况。所以，这部分我们要着重的仔细分析一番。
下面我们来看一下这块的源码如下所示：

5.1> 流程解释

上面代码一共可以分为3部分内容，已经都用红框标注好了，分别是：
- part1：向链表中插入Node的操作
- part2：向红黑树中插入Node的操作（红黑树本次不涉及）
- part3：扩容或者转换元素存储类型的操作

当然，前提是针对f进行synchronize加锁。通过这段代码，我们可以看得出，ConcurrentHashMap是针对具体某个下标的Node进行并发竞争加锁的。极大的避免了由于加锁导致的效率低下的问题。

下面，我们就针对这3个part进行解析

5.2> part1：向链表中插入Node的操作

相关源码如下所示：

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

fh表示Node f的hash值，如果大于等于0，则表示正常的Node节点。
binCount=1对应链表中的第2个Node节点。
从链表的头节点遍历到末尾节点：
- 如果f节点的hash值与put的key的hash值相同，并且两个key值也是相同，那么如果onlyIfAbsent=false则将新的value值替换旧的value值，否则不替换value值。执行完毕后，break跳出循环。
- 遍历到末尾节点，依然没有找到key值且hash值相同的Node，则将新Node加入到链表末尾。执行完毕后，break跳出循环。

5.3> part3：扩容或者转换元素存储类型的操作

如果Node链表长度大于等于9，则执行treeifyBin方法进行扩容或者转换元素存储操作。

5.3.1> treeifyBin(Node<K,V>[] tab, int index)

源码部分：

如果table数组长度小于64，则执行扩容操作。
如果数组长度大于等于64，则进行红黑树扩充。

5.3.2> tryPresize(int size)

首先，通过tableSizeFor根据size计算出2的n次方所有值中，所有大于size值中最小的那个值。具体的逻辑，参考【5.3.7> tableSizeFor(int c)】
下面是tryPresize的具体判断逻辑框架代码

private final void tryPresize(int size) {
    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) {
            ... ...
        }
        else if (c <= sc || n >= MAXIMUM_CAPACITY) {
            break;
        }
        else if (tab == table) {
            ... ...
        }
    }
}

首先，正常情况下，sizeCtl表示table数组的阈值，所以肯定是大于0的。while循环里一共有3个判断逻辑：
- case1：table数组没有初始化完毕。

这块就是创建table数组，没什么别的复杂逻辑。

case2：数组超过最大值，或者扩容发生越界。（MAXIMUM_CAPACITY=1<<30）

针对如上特殊情况，即直接break跳出循环。

case3：table还是那个table，这个过程中没有被其他线程重建过。

5.3.3> case1：table数组没有初始化完毕

源码如下：

if (tab == null || (n = tab.length) == 0) {
    n = (sc > c) ? sc : c;
    if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { /** 将sizeCtl设置为-1 */
        try {
            if (table == tab) {
                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                table = nt;
                sc = n - (n >>> 2); /** 3/4*n */
            }
        } finally {
            sizeCtl = sc; /** 将sizeCtl设置为3/4的table数组长度 */
        }
    }
}

执行过程如下所示：
- 首先，将sizeCtl赋值为-1，则表示针对table正在操作中。
- 其次，创建table，且sc赋值为table数组的3/4长度。
- 最后，将sizeCtl赋值为扩容阈值，表示针对table的操作已经执行完毕。

5.3.4> case2：数组超过最大值，或者扩容发生越界

源码如下：

else if (c <= sc || n >= MAXIMUM_CAPACITY) { /** 表越界 */
    break;
}

5.3.5> case3：table还是那个table，这个过程中没有被其他线程重建过

源码如下：

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)) { // sizeCtl=-2145714174
        transfer(tab, null);
    }
}

resizeStamp方法的具体作用是返回table数组长度相关信息。源码如下所见：

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

【解释】

其中，Integer.numberOfLeadingZeros(n)具体逻辑，请参考【5.3.8> numberOfLeadingZeros(int i)】

我么可以举例上面方面发计算过程

1> 假设n=16，那么二进制就是00010000，那么从左侧最高位开始计算，连续一共有27个0，那么Integer.numberOfLeadingZeros(16)就返回27。

2> RESIZE_STAMP_BITS=16，那么1<<(RESIZE_STAMP_BITS - 1)=1<<(16-1)=1<<15

3> 我们在计算27 | 15，转换为二进制就是：00011011 | 00001000000000000000 = 00001000000000011011

综上所述，resizeStamp返回的结构由三部分组成，就是：

【高16位】16个0

【第16位】1

【第0~15位】"以二进制对table数组长度进行转换，然后计算从最左边算起连续的“0”的总数量"的二进制表现。

我们介绍完resizeStamp方法后，往下看if判断，sc表示sizeCtl，如果sc < 0，则说明table数组正在被其他线程操作着（比如：扩容）。

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

因为在table被其他线程操作的时候，sc其实高16位表示的就是resizeStamp返回的值，即，我们可以认为是table数组长度的信息。那么如果不等于rs，说明数组的长度已经变化了，如果sc==rs+1，说明数组长度是二进制里前进了1个bit，即：十进制中原数组长度的2倍。后面其他的判断，就不一一说明了，比较容易理解。我们来关注else if中的内容，因为这部分的transfer方法，才是我们关注的重点。如下所示：

else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) { 
    transfer(tab, null);
}

我们先理解一下(rs << RESIZE_STAMP_SHIFT) + 2)是什么意思？

高16位：记录了旧数组的容量大小。

低16位：保存了参与扩容的线程数量，假设低16位是n，则n-1就是扩容的线程数。

问题：为什么要+2而不是从1开始计数呢？

答：因为数组初始化时，sizeCtl设置为-1，所以1的那个位置被占了，所以从2开始计算。

下面我们来解析一下transfer方法。

5.3.6> transfer(Node<K,V>[] tab, Node<K,V>[] nextTab)

transfer这个方法主要是用来执行扩容和之后的数据迁移操作的。这个方法逻辑比较复杂，我们就不把这个方法整个的源码都粘过去了。而是分步骤来给大家标注讲解。
我们先来整体的介绍一下这个方法的主要流程：
- Step1：计算迁移时的步长（stride）。
- Step2：如果入参nextTab传入null，那么创建初始化nextTab。
- Step3：开启无限循环。
  - 首先：计算需要转移的节点范围。
  - 然后：将待转移的节点范围中的每个节点的数据进行转移。

a> Step1：计算迁移时的步长

源码如下所示：

if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) {
    stride = MIN_TRANSFER_STRIDE; 
}

扩容时，计算每次转移的固定节点数(步长)
- 如果NCPU大于1，则stride=n/8/NCPU，否则stride=n；但是，如果计算出来的stride小于16，那么stride就被赋值为16
- 其中：最小转移的节点数为：MIN_TRANSFER_STRIDE=16
我们假设现在有A、B两个线程共同执行transfer，入参tab的长度为32，nextTable=null。
- MIN_TRANSFER_STRIDE=16
- n=tab.length=32
- NCPU=8（我自己的电脑）
- 那么n>>>3=32/8=4，由于4<16，所以stride=16。

b> Step2：初始化nextTab

源码如下所示：

if (nextTab == null) { // eg1: nextTab=null
    try {
        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; /** nt=2*n */
        nextTab = nt; /** 由于nextTab为null，所以此处初始化一个nextTab */
    } catch (Throwable ex) {
        sizeCtl = Integer.MAX_VALUE;
        return;
    }
    nextTable = nextTab;
    transferIndex = n; /** 当出现并发扩容时，这个全局变量，是用来给各个线程分配节点的。*/
}

如果transfer的第二个入参nextTab为null，那么就会执行上面这段代码。
- 首先，创建n的2倍长度的新数组。n如果等于32，那么nt数组的长度就是64。
- 其次，将新的table数组nt赋值给nextTable和nextTab变量。即：nextTable=nextTab=nt
- 最后，transferIndex=n=32。

c> Step3：扩容及数据迁移

首先，创建ForwardingNode，源码如下所示：

ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);

数据结构如下所示：

开启无限循环

boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
    Node<K,V> f;
    int fh;
    while (advance) {
        int nextIndex; /** 下一个索引 */
        int nextBound; /** 下一个边界 */
        if (--i >= bound || finishing) { /** i执行-1操作，即：往前遍历一步 */
            advance = false; /** 跳出while循环 */
        }
        else if ((nextIndex = transferIndex) <= 0) {  /** transferIndex<=0，表示数据迁移的活儿都分配完毕了，不需要再划分范围执行迁移了 */
            i = -1;
            advance = false;
        }
        else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,
                nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { /** nextBound表示下一个迁移的边界 */
            bound = nextBound; /** 设置边界 */
            i = nextIndex - 1; /** i表示需要遍历数组的下标，用于下面根据i来迁移对应的链表 */
            advance = false; /** 因为advance=false，所以跳出while循环 */
        }
    }
    ... 下面代码后续我们在分析 ...
}

【解释】

while循环中其实主要做了两件事：
- 第一件：i其实表示要数据迁移的数组下标。那么--i，其实相当于往前遍历一步，随着每次执行for循环，都会一步一步往前走。
- 第二件：如果多线程执行，会在CAS赋值transferIndex的时候发生碰撞。transferIndex表示下一个待迁移的边界。

所以，每个线程迁移数据的范围是从bound到i，迁移长度为stride。

transferIndex就是当出现多线程并发扩容时，这个全局变量，这个变量是线程共享的，是用来给各个线程分配节点的。其他变量都是线程内部自有的，线程私有参数。
下面代码的主框架是4个case的判断，如下所示：

if (i < 0 || i >= n || i + n >= nextn) {
    ... case1：扩容迁移结束逻辑 ...
}
else if ((f = tabAt(tab, i)) == null) {
    ... case2：如果下标i处没有节点，则不需要进行扩容迁移操作 ...
}
else if ((fh = f.hash) == MOVED) {
    ... case3：下标为i的这个位置已经被处理过了，设置advance = true，重新执行while(advance) ...
}
else {
    ... case4：执行扩容迁移操作 ...
}

【解释】

下面的c-1、c-2、c-3、c-4四步骤进行详细解析。

c-1> case1：扩容迁移结束逻辑

源码如下所示：

if (i < 0 || i >= n || i + n >= nextn) { /** 扩容的活儿都分配完毕了 */
    int sc;
    if (finishing) {
        nextTable = null;
        table = nextTab;
        sizeCtl = (n << 1) - (n >>> 1); /** 设置新数组的阈值——sizeCtl=1.5*n=0.75*2n */
        return;
    }
    if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
        if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) {
            return;
        }
        finishing = advance = true;
        i = n; // recheck before commit
    }
}

【解释】

当扩容的活儿都分配完毕了，那么当前线程就不用执行扩容迁移行为了。那么此时i=-1
由于finishing为false，所以，第一次执行这段代码的时候，会直接跳过if(finishing)的代码块。
第二个if判断的意思就是，由于当前线程不需要再执行扩容迁移任务了，所以将总参与线程数减掉一个

sc和sizeCtl前面已经介绍过，它包含两部分含义：高16位表示数组长度信息，低16位表示参与扩容数据迁移的总线程数。

finishing=true之后，再次执行此代码块，就会执行if(finishing)部分。
- 这段代码会将table数组赋值为扩容后的新数组nextTab（长度为旧数组的2倍）
- 根据新的数组nextTab，设置新的阈值。

c-2> case2：如果下标i处没有节点，则不需要进行扩容迁移操作

源码如下所示：

else if ((f = tabAt(tab, i)) == null) { /** 如果下标i处没有节点，则不需要进行扩容迁移操作 */
    advance = casTabAt(tab, i, null, fwd);
}

【解释】

这段由于不需要进行数据迁移了，所以将旧table的i位置赋值fwd节点，表明该位置已经处理过了。

c-3> case3：下标为i的这个位置已经被处理过了

源码如下所示：

else if ((fh = f.hash) == MOVED) { /** 下标为i的这个位置已经被处理过了，设置advance = true，重新执行while(advance) */
    advance = true; // already processed
}

【解释】

如果已经被处理，则跳过这个位置，继续往前遍历。

c-4> case4：执行扩容迁移操作

源码如下所示：

synchronized (f) {
    if (tabAt(tab, i) == f) {
        Node<K,V> ln; /** 低位的node元素 */
        Node<K,V> hn; /** 高位的node元素 */
        if (fh >= 0) {
            int runBit = fh & n; /** runBit=0时，表明元素f在当前的位置不用移动；否则需要移动到新扩展的区域 */
            Node<K,V> lastRun = f;
            /** 遍历到链表的最后一个元素，跳出for循环 */
            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) { /** 将存储在低位区的Node拼装成新的链表 */
                    ln = new Node<K,V>(ph, pk, pv, ln);
                }
                else { /** 将存储在高位区的Node拼装成新的链表 */
                    hn = new Node<K,V>(ph, pk, pv, hn);
                }
            }
            setTabAt(nextTab, i, ln); /** 设置低位 */
            setTabAt(nextTab, i + n, hn); /** 设置高位 */
            setTabAt(tab, i, fwd);
            advance = true;
        }
        else if (f instanceof TreeBin) {
            ... 红黑树处理逻辑（不涉及讲解部分） ...
        }
    }
}

【解释】

首先，锁住下标为i的元素Node<K,V> f
计算runBit值——int runBit = fh & n; n为旧table数组的长度（假如n=16，则二进制为10000，那么其实后面四个0才是原数组对应的下标位）综上所述，如果runBit=0，则说明元素不需要迁移，因为还是在低位区ln；否则需要迁移，因为是在高位区hn（也就是我们2倍扩展新生成的位置）；
针对原table数组中旧数据拆出高位区和低位区的处理方式如下图所示：

组装好ln和hn后，将其插入到相应的高区位下标i+n和低区位i上。
将旧的table数组对应i的位置插入fwd节点，表明该位置已经处理过了。（因为hash=MOVED=-1）

5.3.7> tableSizeFor(int c)

源码部分：

函数的作用——返回指定容量的2的n次方。即:输入值为c，返回值为result：满足result=(2^k>=c)，k取最小值。
输入输出结果如下：
- 输入3、4、5时，返回4、4、8
- 输入11、12、13时，都返回16
- 输入200、210、220时，都返回256。
由于table数组长度都是2的n次方，且初始值为16，所以，可以通过高位连续多少个0来判断数组的长度是否相同。
如果是偶数，则原样输出

如果是奇数，则(n-1)*2

5.3.8> numberOfLeadingZeros(int i)

这个方法的作用是

传入一个int类型数据，返回这个数据的二进制串中从最左边算起连续的“0”的总数量。因为int类型的数据长度为32所以高位不足的地方会以“0”填充。

我们先看一下这个方法的源码：

public static int numberOfLeadingZeros(int i) {
        if (i == 0)
            return 32;
        int n = 1;
        /**
         * 00 | 00 | 0000 | 0000 0000 | 0000 0000 0000 0000
         *   [30] [28]   [24]        [16]
         */
        /** 【32~17】（高16位）是否都为0，如果都是0，那么n=17，i左移16位，实现对高16位的清空操作 */
        if (i >>> 16 == 0) { // eg1: 0100>>>16=0000
            n += 16; // eg1: n=1+16=17
            i <<= 16; // eg1: i=0100<<16= 0100 0000 0000 0000 0000
        }
        /** 【16~9】是否都为0 */
        if (i >>> 24 == 0) {
            n += 8; // eg1: n=17+8=25
            i <<= 8; // eg1: i=0100 0000 0000 0000 0000 << 8 = 0100 0000 0000 0000 0000 0000 0000
        }
        /** 【8~5】是否都为0 */
        if (i >>> 28 == 0) {
            n += 4; // eg1: n=25+4=29
            i <<= 4; // eg1: i=0100 0000 0000 0000 0000 0000 0000 << 4 = 0100 0000 0000 0000 0000 0000 0000 0000
        }
        /** 【4~3】是否都为0 */
        if (i >>> 30 == 0) {
            n += 2;
            i <<= 2;
        }
        n -= i >>> 31; // eg1: n=29-(0100 0000 0000 0000 0000 0000 0000 0000 >>> 31)=29-0=29
        return n;
    }

方法源码看起来不太直观，我们通过图来了解一下

六、case3：如果寻址后的位置是正在迁移状态

源码如下所示：

我们来看MOVED变量的值时什么？

也就是说，当我们通过hash寻址到了我们应该插入的下标为i的位置上，已经存在了Node f，并且这个f的hash值等于-1，说明当前这个下标为i的位置，正在执行移动操作。那么，我们会通过执行helpTransfer方法来协助其他线程进行扩容操作。详细操作我们来看helpTransfer方法的具体实现。

6.1> helpTransfer(Node<K,V>[] tab, Node<K,V> f)

该方法的作用就是参与扩容数据迁移的操作。
源码如下所示：

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

【解释】

f.nextTable存储的是扩容后新的table数组。
int rs = resizeStamp(tab.length);返回的是旧数组的长度信息。
(sc = sizeCtl) < 0说明当前还是在对旧表操作中的状态，即：扩容数据转移还在操作中，没有操作完毕。
U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)含义是，由于当前线程要帮忙去执行扩容和数据迁移操作，所以将总参与线程数+1。
执行transfer操作。该方法具体操作请参考【5.3.6> transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) 】

七、addCount将kv的总数+1

该方法主要作用就是维护ConcurrentHashMap中总的kv数量值，当存储的总kv值超过了阈值，那么会执行扩容操作。
该方法里面内容也不少，我们还是先简化代码，看一下框架代码：

private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            ...  ...
        }
        if (check >= 0) {
           ...  ...
        }
    }

【解释】

case1：计算当前存入key-value的总数
case2：存储的总kv数量达到了阈值，执行扩容

下面我们针对这两种case进行详细介绍。

7.1> case1：计算当前存入key-value的总数

相关源码如下所示：

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))) {  /** 使用CellValue来计数 */
        fullAddCount(x, uncontended); /** 使用CounterCell来计数 */
        return;
    }
    if (check <= 1) {
        return;
    }
    /** 统计所有count值 */
    s = sumCount();
}

【解释】

第一次执行这段方法的时候，counterCells默认等于null。
假设现在有线程A和线程B两个线程同时执行addCount的这段代码，那么(as = counterCells) != null判断结果为false。
s=b+x，其中b是baseCount，x是addCount的第一个入参，表示总容量需要增加n个kv。
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x))操作只有一条线程可以执行成功，假设，线程A执行将baseCount修改为s成功，则不用进入if判断的方法体中。而线程B则需要进入if的方法体中。那么，这里大家需要记住一点，总量是有两部分组成的，baseCount就是计数其中之一。
线程B由于执行CAS操作失败，那么对于这种所有竞争失败的线程，都会执行fullAddCount方法。这个方法里面就是计算总量的第二部分——CounterCell.value。

7.1.1> fullAddCount(long x, boolean wasUncontended)

fullAddCount的源码比较多，我们还是先去除“冗余”代码，只看框架相关的代码，如下所示：

private final void fullAddCount(long x, boolean wasUncontended) {
int h;
if ((h = ThreadLocalRandom.getProbe()) == 0) {
    ... 初始化操作随机数生成器 ...
}
/**【开启无限循环】*/       
for (;;) {
    /**
     * case1: counterCells不为空 且 数组里面有元素
     */
    if ((as = counterCells) != null && (n = as.length) > 0) {
        f ((a = as[(n - 1) & h]) == null) {
          ...如果随机数h待插入的下标位置没有元素,则插入CounterCell ...
        }
        else if (!wasUncontended) { /** 如果wasUncontended=false（表示当前线程CAS竞争失败），则将wasUncontended重置为true */
            wasUncontended = true;      
        }
        else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)) { /** 尝试将a的value值加x */
            break;
        }
        else if (counterCells != as || n >= NCPU) {
            collide = false; // At max size or stale
        }
        else if (!collide) {
            collide = true;
        }
        else if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
            ... 多线程之间设置CounterCell的value值时发生碰撞，那么扩展CounterCell的长度，以减少碰撞次数 ...
        }
        ... ...
    }
    /**
     * case2: cellsBusy为0且counterCells为空
     */
    else if (cellsBusy == 0 && counterCells == as && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
        ... 创建一个长度为2的CounterCell数组，将x赋值进数组，跳出循环 ...
    }
    /**
     * case3: 尝试修改baseCount的值
     */
    else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x)) {
        ... 如果修改成功，则跳出循环 ...
        break; 
    }
}

【解释】

fullAddCount代码较多，总体可以分为两部分：
- 第一部分，如果获得随机数为0，则初始化当前线程的探针哈希值。
- 第二部分，开启无限循环，利用CounterCell进行计数。

下面我们来关注这两部分。

第一部分比较简单。只需要记住一下几个方法的作用：

ThreadLocalRandom.getProbe()：用来获得随机数

ThreadLocalRandom.localInit()：初始化当前线程的探针哈希值

ThreadLocalRandom.advanceProbe(h)：更改当前线程的探针哈希值

第二部分，利用CounterCell进行计数一共分为3种情况：
- case1：counterCells不为空且数组里面有元素
- case2：cellsBusy为0且counterCells为空
- case3：尝试修改baseCount的值

这里我通过一张图，先介绍一下代码里的逻辑：

a> case1：counterCells不为空且数组里面有元素

我们先看一下这部分源码：

CounterCell[] as;
CounterCell a;
int n;
long v;
if ((as = counterCells) != null && (n = as.length) > 0) {
    if ((a = as[(n - 1) & h]) == null) { /** 如果随机数h待插入的下标位置没有元素 */
        if (cellsBusy == 0) {            // Try to attach new Cell
            CounterCell r = new CounterCell(x); // Optimistic create
            /** 通过CAS将cellsBusy设置为1，来表示正在操作CounterCell；操作完毕后，将cellsBusy设置为0，其他线程可以继续操作 */
            if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
                boolean created = false;
                try {    // Recheck under lock
                    CounterCell[] rs;
                    int m;
                    int j;
                    if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) {
                        rs[j] = r; /** 在下标j处，插入新建的CounterCell r */
                        created = true;
                    }
                } finally {
                    cellsBusy = 0;
                }
                if (created) {
                    break;
                }
                continue;           // Slot is now non-empty
            }
        }
        collide = false; /** collide:碰撞，cellsBusy不等于0时，表示有其他线程正在操作CounterCell，collide设置为false */
    }
    else if (!wasUncontended) { /** 如果wasUncontended=false（表示当前线程CAS竞争失败），则将wasUncontended重置为true */
        wasUncontended = true;      // CAS already known to fail，Continue after rehash
    }
    else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)) { /** 尝试将a的value值加x */
        break;
    }
    else if (counterCells != as || n >= NCPU) {
        collide = false;            // At max size or stale
    }
    else if (!collide) {
        collide = true;
    }
    /** 多线程之间设置CounterCell的value值时发生碰撞，那么扩展CounterCell的长度，以减少碰撞次数 */
    else if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
        try {
            if (counterCells == as) {// Expand table unless stale
                CounterCell[] rs = new CounterCell[n << 1]; /** 根据原数组长度扩张2倍*/
                for (int i = 0; i < n; ++i) {
                    rs[i] = as[i]; /** 进行数据迁移 */
                }
                counterCells = rs;
            }
        } finally {
            cellsBusy = 0;
        }
        collide = false;
        continue;                   // Retry with expanded table
    }
    /** 更改当前线程的探针哈希值 */
    h = ThreadLocalRandom.advanceProbe(h);
}

counterCells表示CounterCell的数组，那么我们先介绍一下CounterCell吧。

@sun.misc.Countended注解是做什么的？

Java8的@sun.misc.Contended注解（Contended：有争议的）

@sun.misc.Contended是Java8新增的一个注解，对某字段加上该注解则表示该字段会单独占用一个缓存行（Cache Line）。

这里的缓存行是指CPU缓存（L1、L2、L3）的存储单元，常见的缓存行大小为64字节。

（注：JVM添加-XX:-RestrictContended参数后@sun.misc.Contended注解才有效）

单独使用一个缓存行有什么作用？

答：避免伪共享

为了提高读取速度，每个CPU有自己的缓存，CPU读取数据后会存到自己的缓存里。而且为了节省空间，一个缓存行可能存储着多个变量，即伪共享。

但是这对于共享变量，会造成性能问题，如下所示：

当一个CPU要修改某共享变量A时会先锁定自己缓存里A所在的缓存行，并且把其他CPU缓存上相关的缓存行设置为无效。但如果被锁定或失效的缓存行里，还

存储了其他不相干的变量B，其他线程此时就访问不了B，或者由于缓存行失效需要重新从内存中读取加载到缓存里，这就造成了开销。所以让共享变量A单独

使用一个缓存行就不会影响到其他线程的访问。

Java8之前的方案是什么？

在Java8之前，是通过代码里手动添加属性的方式解决的，如：

class LongWithPadding {

long value;

long p0, p1, p2, p3, p4, p5, p6;

}

一个long占8个字节，再添加7个long属性就会变成64个字节，刚好是一个缓存行大小。

但是注意，Java7开始JVM做优化时可能会把不用的变量给去掉，所以这种方法并不推荐使用。

适用场景

主要适用于频繁写的共享数据上。如果不是频繁写的数据，那么CPU缓存行被锁的几率就不多，所以没必要使用了，否则不仅占空间还会浪费CPU访问操作数据的时间。

如果(a = as[(n - 1) & h]) == null，则表明随机数h待插入的下标位置没有元素。cellsBusy=0表示当前处理CounterCell是空闲的状态，那么就创建CounterCell，然后通过CAS的方式将cellsBusy赋值为1，表明现在正在处理CounterCell中，将其插入到conuterCells数组中后，将cellsBusy赋值为0，表明操作完毕。

if ((a = as[(n - 1) & h]) == null) { /** 如果随机数h待插入的下标位置没有元素 */
    if (cellsBusy == 0) {            // Try to attach new Cell
        CounterCell r = new CounterCell(x); // Optimistic create
        /** 通过CAS将cellsBusy设置为1，来表示正在操作CounterCell；操作完毕后，将cellsBusy设置为0，其他线程可以继续操作 */
        if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
            boolean created = false;
            try {    // Recheck under lock
                CounterCell[] rs;
                int m;
                int j;
                if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) {
                    rs[j] = r; /** 在下标j处，插入新建的CounterCell r */
                    created = true;
                }
            } finally {
                cellsBusy = 0;
            }
            if (created) {
                break;
            }
            continue;           // Slot is now non-empty
        }
    }
    collide = false; /** collide:碰撞，cellsBusy不等于0时，表示有其他线程正在操作CounterCell，collide设置为false */
}

如果wasUncontended=false（表示当前线程CAS竞争失败），则将wasUncontended重置为true。

else if (!wasUncontended) {
    wasUncontended = true;      // CAS already known to fail，Continue after rehash
}

如果随机数h待插入的下标位置存在CounterCell a，尝试将a的value值加x

else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)) { 
    break;
}

如果counterCells已经发生了变化（因为下面会有扩容的情况发生）

else if (counterCells != as || n >= NCPU) {
    collide = false;            // At max size or stale
}

多线程之间设置CounterCell的value值时发生了碰撞，那么扩展CounterCell的长度，以减少碰撞次数

else if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
    try {
        if (counterCells == as) {// Expand table unless stale
            CounterCell[] rs = new CounterCell[n << 1]; /** 根据原数组长度扩张2倍*/
            for (int i = 0; i < n; ++i) {
                rs[i] = as[i]; /** 进行数据迁移 */
            }
            counterCells = rs;
        }
    } finally {
        cellsBusy = 0;
    }
    collide = false;
    continue;                   // Retry with expanded table
}

b> case2：cellsBusy为0且counterCells为空

如果满足这种条件，那么创建一个长度为2的CounterCell数组counterCells，并将x赋值进数组，跳出循环

else if (cellsBusy == 0 && counterCells == as && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
    boolean init = false;
    try { // Initialize table
        if (counterCells == as) {
            CounterCell[] rs = new CounterCell[2]; /** 创建长度为2的数组 */
            rs[h & 1] = new CounterCell(x);
            counterCells = rs;
            init = true;
        }
    } finally {
        cellsBusy = 0;
    }
    if (init) {
        break;
    }
}

c> case3：尝试修改baseCount的值

如果修改baseCount成功，那么则跳出循环

else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x)) {
    break; // Fall back on using base
}

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

【解释】

里面逻辑比较简单，ConcurrentHashMap里总的kv数量就是：【baseCount数量】+【sum(CounterCells里所有CounterCell的value)】

7.2> case2：存储的总kv数量达到了阈值，执行扩容

源码如下所示：

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右移16位，则读取的就是原高16位的内容，即：table容量相关信息;不等于rs说明table容量发生了不一致的情况 */
                sc == rs + 1 ||
                sc == rs + MAX_RESIZERS ||
                (nt = nextTable) == null ||
                transferIndex <= 0) {
            break;
        }
        if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { /** 加入1个共同扩容的线程，即：sc+1*/
            transfer(tab, nt);
        }
    }
    /** 执行扩容操作 */
    else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) {
        transfer(tab, null);
    }
    /** 统计所有count值 */
    s = sumCount();
}

【解释】

逻辑包含两个内容：

case1：如果sc为负值，表明正在执行扩容操作中，那么也加入扩容的“大部队”中

case2：否则，表明table数组没有扩容，那么，发起扩容操作。

这部分内容涉及的内容U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)、U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)、transfer方法和sumCount方法在上面都介绍过了。那么我就不再介绍了。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-10-28，如有侵权请联系 cloudcommunity@tencent.com 删除

缓存