目录
- 一、ConcurrentHashMap使用场景
- 二、put方法的整体流程
- 三、第一种情况（初始化table数组）
- 四、第二种情况（寻址后的位置没有被占用）
- 五、第三种情况（如果寻址后的位置是正在迁移状态）
- 六、其他情况（ 将节点插入到链表中或者红黑树中。）
- 七、addCount将ConcurrentHashMap中存储的k,v总数+1
一、ConcurrentHashMap使用场景
-
我们平时最常用的HashMap其实不是线程安全的，而当我们有多线程使用场景的时候，即想线程安全，又想拥有Map的能力，我们可以选择HashTable，因为它是针对我们常用的方法上面加上了synchronize锁，但是在高并发的场景下，效率低是它的弊端。如果我们还非常在意效率，那么我们更好的选择是使用ConcurrentHashMap。
-
示例：
启动100个线程，每个线程循环100次，像容器中应该放入10000个元素。我们看到运行结果可以发现，HashMap并不是10000，这就说明，它在多线程并发的情况下，出现了线程不安全的问题。而ConcurrentHashMap返回的结果是没有问题的。
二、put方法的整体流程
1、put方法源码
public V put(K key, V value) {
return putVal(key, value, false);
}
2、在put方法中，调用了putVal方法。由于该方法代码较多，我们只保留框架性质的代码，这样会更方便我们理解。如下所示：
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
... ...
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
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; // no lock when adding to empty bin
}
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;
}
3、上面的代码，可以分为两部分内容，
（1）、第一部分：首先开启了无限循环，在里面进行了4中情况的判断。
-
第一种情况：如果table数组需要被创建。
如果table数组为null或者长度为0，则创建table数组。 -
第二种情况：如果寻址后的位置没有被占用。
创建Node节点，插入到这个位置 -
第三种情况：如果寻址后的位置是正在迁移状态。
加入到迁移大军中，帮助一起进行扩容迁移操作。 -
第四种情况：其他情况
将节点插入到链表中或者红黑树中。
（2）、 第二部分：执行addCount，将ConcurrentHashMap中存储的k,v总数+1。
- 下面将骤一的对上述的四个情况和addCount步骤进行分析。
三、第一种情况（初始化table数组）
3.1、initTable源码如下：
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
//table=null
while ((tab = table) == null || tab.length == 0) {
//sizeCtl =0
//如果sizeCtl为-1，则表示table数组正在被别的线程初始化
if ((sc = sizeCtl) < 0)
/**
* 其他竞争失败的线程，会在while循环中spin，直到table创建完毕
* 才能跳出while循环
*/
Thread.yield();
/**
* 如果我们是第一个执行初始化table数组，那么我们首先通过CAS把
* sizeCtl设置为-1表明名正在初始化中。
*
* 只有一条线程可以成功执行CAS操作，将 sizeCtl赋值为-1，其他竞争失败
* 的线程，在外面spin，直到table创建完毕才能跳出while循环。
*/
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
//table=null
if ((tab = table) == null || tab.length == 0) {
//sc =0 ,DEFAULT_CAPACITY=16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
// n =16
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//n>>>2=10000>>>2=00100=4 (16/2/2=4)
/**sc= (4-1)/4n =0.75*n*/
sc = n - (n >>> 2); //sc =16-4 =12
}
} finally {
//sc=12, sizeCtl =12
sizeCtl = sc;
}
break;
}
}
//返回size为16的空的Node数组
return tab;
}
3.2、sizeCtl
-
如果sizeCtl为-1，则表示table数组正在被别的线程初始化。默认sizeCtl=0，当table数组初始化或者扩容完毕的时候，sizeCtl会表示扩容阈值。
-
默认table数组的长度为16
3.3、流程解释
-
我们通过源码可以看到，如果table没有被初始化完毕的话，那么会一直在while循环中，直到table数组初始化完毕：
while ((tab = table) == null || tab.length == 0) -
假设现在有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数组 }
四、第二种情况（寻址后的位置没有被占用）
4.1、源码如下
/**
*目标位置为空，直接设置
*通过hash值对应的数组下标得到第一个节点；虽然table数组是线程间可见，但是数组元素未必。
*所以，以volatile读的方式来读取table数组中的元素，保证每次拿到的数组都是最新的
*i=[在数组中的位置]
*tabAt(tab, i = (n - 1) & hash)相当于tab[i]
*/
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//f =null i=[在数组中的位置]
//如果该下标返回的节点为空，则直接通过cas将新的值封装成node插入即可；
//如果cas失败，说明存在竞争，则进入下一次循环
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null))){
break;
}
}
- 其中，关键点是，我们要理解tabAt和casTabAt这两个方法的实现逻辑，那么在了解这两个方法之前，我们需要先了解两个变量的含义，即：ABASE和ASHIFT。
4.2、ABASE和ASHIFT 变量
-
源码部分
private static final long ABASE; static { try { ... 此处代码省略... Class<?> ak = Node[].class; /** *数组对象：对象头（8字节）* 指针（4字节）* 数组长度（4字节）=16，所以 *基础偏移量为16 * * public native int arrayBaseOffset(Class<?> var1); * 返回数组类型的[第一个元素]的偏移地址（基础偏移地址）。 * 如果arrayIndexScale方法返回的比例因子不为0，可以结合基础偏移地址 * 和比例因子访问数组的所有元素。 * Unsafe类中已经初始化了很多类似的常量如ARRAY_BOOLEAN_BASE_OFFSET等。 */ ABASE = U.arrayBaseOffset(ak);//16 /** * public native int arrayIndexScale(Class<?> var1); * 返回数组类型的[比例因子]（其实就是【数组中元素偏移地址的增量】， * 因为数组中的元素的地址是连续的）。 * 此方法不适用于数组类型为“narrow”类型的数组，“narrow”类型的数组类型 * 使用此方法会返回0（这里narrow应该是狭义的意思，但是具体指哪些类型 * 暂时不明确）。 * Unsafe类中已经初始化了很多类似的常量如ARRAY_BOOLEAN_INDEX_SCALE等。 **/ int scale = U.arrayIndexScale(ak);//4 if ((scale & (scale - 1)) != 0)// 4 & (4-1) =0100 & 0011 =0000 throw new Error("data type scale not a power of two"); /** *Integer.numberOfLeadingZeros(int i) *给定一个int类型数据，返回这个数据的二进制串中从最左边算起连续的“0”的 *总数量。 *而ASHIFT其实就是将scale数值转换为[按位左移对应的数值]， *即：通过scale =4，name计算出ASHIFT=2， *而N<<2其实就相当于N*2*2=N*4=N*scale *举例： * 如果scale =8（十进制）=1000（二进制） * name计算出Integer.numberOfLeadingZeros(scale)=28 * ASHIFT =31-28 =3 * N<<3 =n*2*2*2=N*8=N*scale */ //ASHIFT=31-29=2 ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); } catch (Exception e) { throw new Error(e); } }
4.3、tabAt 方法
-
源码部分
-
作用
获得tab数组下标为i位置上的Node元素 -
数组的寻址公式为
a[i]_address = base_address + i*data_type_size，通过该方式可以获得对应下标为i的值，即：获得tab[i]的值。 -
public native Object getObjectVolatile(Object o, long offset);
（1）、此方法和getObject功能类似，不过附加了volatile语义，也就是强制从主存中获取属性值
（2）、类似的方法有getIntVolatile、getDoubleVolatile等等。
（3）、这个方法要被使用的属性由volatile修饰，否则功能和getObject方法相同。
（4）、offset= ((long)i << ASHIFT) + ABASE，表示从ABASE开始，计算第i个元素的偏移量。
（5）、所以：tabAt(tab, i)就等同于tab[i]。
4.4、casTabAt 方法
-
源码部分
-
作用
如果tab数组下标为i的Node元素是c，则将c修改为v。 -
public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5);
（1）、针对Object对象进行CAS操作。即是对应Java变量引用var1，原子性地更新var1中偏移地址为offset的属性的值为var5，当且仅的偏移地址为offset的属性的当前值为var4才会更新成功返回true，否则返回false。
（2）、var1：目标Java变量引用。
（3）、var2：目标Java变量中的目标属性的偏移地址。
（4）、var4：目标Java变量中的目标属性的期望的当前值。
（5）、var5：目标Java变量中的目标属性的目标更新值。 -
类似的方法有compareAndSwapInt和compareAndSwapLong，在Jdk8中基于CAS扩展出来的方法有getAndAddInt、getAndAddLong、getAndSetInt、getAndSetLong、getAndSetObject，它们的作用都是：通过CAS设置新的值，返回旧的值。
4.5、流程解释
-
这部分属于判断中的第二种情况，通过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的代码段中。
五、第三种情况（如果寻址后的位置是正在迁移状态）
5.1、源码
-
源码如下
/** *插入元素遇到扩容，即MOVED =-1，表示插入元素时，无限循环内部正好在扩容；那么就 *调用helpTransfer(tab, f)一起扩容； *由于是在for（Node<K,V>[] tab =table;;）,所以当帮忙扩容完毕后，还会像新的 *table中插入元素的。 */ else if ((fh = f.hash) == MOVED){ tab = helpTransfer(tab, f); } -
MOVED变量的值是什么
-
当我们通过hash寻址到了我们应该插入的下标为i的位置上，已经存在了Node f，并且这个f的hash值等于-1，说明当前这个下标为i的位置，正在执行移动操作。那么，我们会通过执行helpTransfer方法来协助其他线程进行扩容操作。详细操作我们来看helpTransfer方法的具体实现。
5.2、helpTransfer方法
-
源码如下
-
helpTransfer方法作用
参与扩容数据迁移的操作。 -
源码解释
（1）、f.nextTable存储的是扩容后新的table数组。
（2）、int rs = resizeStamp(tab.length);返回的是旧数组的长度信息。
（3）、sc = sizeCtl) < 0 说明当前还是在对旧表操作中的状态，即：扩容数据转移还在操作中，没有操作完毕。
（4）、U.compareAndSwapInt(this, SIZECTL, sc, sc + 1);含义是，由于当前线程要帮忙去执行扩容和数据迁移操作，所以将总参与线程数+1。
（5）、执行transfer操作。该方法具体操作请参考【5.3.6> transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) 】
六、其他情况（ 将节点插入到链表中或者红黑树中。）
我们调用put方法的时候，此部分为最核心的处理逻辑了。因为当table表被初始化了，并且出现哈希冲突了，并且Node f这个位置并没有发生移动的情况下，都会走到这段代码段中。而这种情况又是最多发生的情况。所以，这部分我们要着重的仔细分析一番。
6.1、源码如下
6.2、流程解释
- 上面代码一共可以分为3部分内容，已经都用红框标注好了，分别是：
- 第一部分：向链表中插入Node的操作。
- 第二部分：向红黑树中插入Node的操作。（红黑树本次不涉及）
- 第三部分：扩容或者转换元素存储类型的操作。
- 当然，前提是针对f进行synchronize加锁。通过这段代码，我们可以看得出，ConcurrentHashMap是针对具体某个下标的Node进行并发竞争加锁的。极大的避免了由于加锁导致的效率低下的问题。
6.3、第一部分：向链表中插入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; } } } -
源码解释
（1）、fh 表示Node f的hash值，如果大于等于0，则表示正常的Node节点。
（2）、binCount=1对应链表中的第2个Node节点。
（3）、从链表的头节点遍历到末尾节点：如果f节点的hash值与put的key的hash值相同，并且两个key值也是相同，那么如果onlyIfAbsent=false则将新的value值替换旧的value值，否则不替换value值。执行完毕后，break跳出循环； 遍历到末尾节点，依然没有找到key值且hash值相同的Node，则将新Node加入到链表末尾。执行完毕后，break跳出循环。
6.4、第三部分：扩容或者转换元素存储类型的操作
如果Node链表长度大于等于9，则执行treeifyBin方法进行扩容或者转换元素存储操作。
6.4.1 、treeifyBin方法
-
源码部分
-
如果table数组长度小于64，则执行扩容操作。
-
如果数组长度大于等于64，则进行红黑树扩充。
6.4.2、 tryPresize方法
-
源码如下
首先，通过tableSizeFor根据size计算出2的n次方所有值中，所有大于size值中最小的那个值。具体的逻辑，参考【6.4.4> tableSizeFor(int c)】
-
源码逻辑解释
（1）、首先，正常情况下，sizeCtl表示table数组的阈值，所以肯定是大于0的。while循环里一共有3个判断逻辑：
（2）第一个判断：table数组没有初始化完毕。这块就是创建table数组，没什么别的复杂逻辑。
（3）、第二个判断：数组超过最大值，或者扩容发生越界。（MAXIMUM_CAPACITY=1<<30） 针对如上特殊情况，即直接break跳出循环。
（4）、第三个判断：table还是那个table，这个过程中没有被其他线程重建过。
6.4.2.1、 第一个判断（table数组没有初始化完毕）
-
源码如下
if (tab == null || (n = tab.length) == 0) { n = (sc > c) ? sc : c; /** 将sizeCtl设置为-1 */ if (U.compareAndSwapInt(this, SIZECTL, sc, -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设置为3/4的table数组长度 */ sizeCtl = sc; } } } -
执行过程如下所示
（1）、首先，将sizeCtl赋值为-1，则表示针对table正在操作中。
（2）、其次，创建table，且sc赋值为table数组的3/4长度。
（3）、最后，将sizeCtl赋值为扩容阈值，表示针对table的操作已经执行完毕。
6.4.2.2、第二个判断（数组超过最大值，或者扩容发生越界）
-
源码如下
6.4.2.3、第三个判断（table还是那个table，这个过程中没有被其他线程重建过
）
1、源码如下
2、resizeStamp方法的具体作用是返回table数组长度相关信息。源码如下所见：
3、resizeStamp方法的解释
- 其中，Integer.numberOfLeadingZeros(n)具体逻辑，请参考【6.4.5> numberOfLeadingZeros(int i)】
- 我么可以举例上面方面发计算过程：
假设n=16，那么二进制就是00010000，那么从左侧最高位开始计算，连续一共有27个0，那么Integer.numberOfLeadingZeros(16)就返回27。
RESIZE_STAMP_BITS=16，那么1<<(RESIZE_STAMP_BITS - 1)=1<<(16-1)=1<<15
我们在计算27 | 15，转换为二进制就是：00011011 | 00001000000000000000 = 00001000000000011011 - 综上所述，resizeStamp返回的结构由三部分组成，就是：
（1）、【高16位】16个0
（2）、【第16位】1
（3）、【第0~15位】"以二进制对table数组长度进行转换，然后计算从最左边算起连续的“0”的总数量"的二进制表现。
4、我们介绍完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开始计算。
6.4.3、transfer方法
- transfer这个方法主要是用来执行扩容和之后的数据迁移操作的。这个方法逻辑比较复杂，我们就不把这个方法整个的源码都粘过去了。而是分步骤来给大家标注讲解。
- 我们先来整体的介绍一下这个方法的主要流程：
（1）、步骤1：计算迁移时的步长（stride）。
（2）、步骤2：如果入参nextTab传入null，那么创建初始化nextTab。
（3）、步骤3：开启无限循环。首先：计算需要转移的节点范围。然后：将待转移的节点范围中的每个节点的数据进行转移。
6.4.3.1 步骤1：计算迁移时的步长
1、源码如下所示：
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) {
stride = MIN_TRANSFER_STRIDE;
}
2、扩容时，计算每次转移的固定节点数(步长)
- 如果NCPU大于1，则stride=n/8/NCPU，否则stride=n；但是，如果计算出来的stride小于16，那么stride就被赋值为16
- 其中：最小转移的节点数为：MIN_TRANSFER_STRIDE=16
3、我们假设现在有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。
6.4.3.2 步骤2：初始化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，那么就会执行上面这段代码。
（1）、首先，创建n的2倍长度的新数组。n如果等于32，那么nt数组的长度就是64。
（2）、其次，将新的table数组nt赋值给nextTable和nextTab变量。即：nextTable=nextTab=nt
（3）、最后，transferIndex=n=32。
6.4.3.3 步骤3：扩容及数据迁移
1、首先，创建ForwardingNode，源码如下所示：
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
2、数据结构如下所示：
3、开启无限循环
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循环 */
}
}
... 下面代码后续我们在分析 ...
}
4、解释
-
while循环中其实主要做了两件事：
（1）、第一件：i其实表示要数据迁移的数组下标。那么–i，其实相当于往前遍历一步，随着每次执行for循环，都会一步一步往前走。
（2）、第二件：如果多线程执行，会在CAS赋值transferIndex的时候发生碰撞。transferIndex表示下一个待迁移的边界。
（3）、所以，每个线程迁移数据的范围是从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：执行扩容迁移操作 ... } -
解释：
下面对四个步骤进行详细解析。
6.4.3.3.1 case1：扩容迁移结束逻辑
1、源码如下
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
}
}
2、解释
- 当扩容的活儿都分配完毕了，那么当前线程就不用执行扩容迁移行为了。那么此时i=-1
- 由于finishing为false，所以，第一次执行这段代码的时候，会直接跳过if(finishing)的代码块。
- 第二个if判断的意思就是，由于当前线程不需要再执行扩容迁移任务了，所以将总参与线程数减掉一个，sc和sizeCtl前面已经介绍过，它包含两部分含义：高16位表示数组长度信息，低16位表示参与扩容数据迁移的总线程数。
- finishing=true之后，再次执行此代码块，就会执行if(finishing)部分。
（1）、这段代码会将table数组赋值为扩容后的新数组nextTab（长度为旧数组的2倍）
（2）、根据新的数组nextTab，设置新的阈值。
6.4.3.3.2 case2：如果下标i处没有节点，则不需要进行扩容迁移操作
1、源码如下
else if ((f = tabAt(tab, i)) == null) { /** 如果下标i处没有节点，则不需要进行扩容迁移操作 */
advance = casTabAt(tab, i, null, fwd);
}
2、解释
- 这段由于不需要进行数据迁移了，所以将旧table的i位置赋值fwd节点，表明该位置已经处理过了。
6.4.3.3.3 case3：下标为i的这个位置已经被处理过了
1、源码如下
else if ((fh = f.hash) == MOVED) { /** 下标为i的这个位置已经被处理过了，设置advance = true，重新执行while(advance) */
advance = true; // already processed
}
2、解释
- 如果已经被处理，则跳过这个位置，继续往前遍历。
6.4.3.3.4 case4：执行扩容迁移操作
1、源码如下
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) {
... 红黑树处理逻辑（不涉及讲解部分） ...
}
}
}
2、解释
-
首先，锁住下标为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）
6.4.4、tableSizeFor方法
-
源码部分
-
函数的作用——返回指定容量的2的n次方。即:输入值为c，返回值为result：满足result=(2^k>=c)，k取最小值。
-
输入输出结果如下：
（1）、输入3、4、5时，返回4、4、8
（2）、输入11、12、13时，都返回16
（3）、输入200、210、220时，都返回256。 -
由于table数组长度都是2的n次方，且初始值为16，所以，可以通过高位连续多少个0来判断数组的长度是否相同。
-
如果是偶数，则原样输出
-
如果是奇数，则(n-1)*2 :
6.4.5、numberOfLeadingZeros方法
-
这个方法的作用是传入一个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; } -
方法源码看起来不太直观，我们通过图来了解一下
七、addCount将ConcurrentHashMap中存储的k,v总数+1
7.1、源码
-
源码如下
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)) { ... 第一种情况，计算当前存入key-value的总数 ... } if (check >= 0) { ... 第二种情况，存储的总kv数量达到了阈值，执行扩容 ... } } -
源码解释
（1）、第一种情况：计算当前存入key-value的总数
（2）、第二种情况：存储的总kv数量达到了阈值，执行扩容 -
addCount方法作用
维护ConcurrentHashMap中总的kv数量值，当存储的总kv值超过了阈值，那么会执行扩容操作。
7.2 第一种情况（计算当前存入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(); } -
源码解释
（1）、第一次执行这段方法的时候，counterCells默认等于null。
（2）、假设现在有线程A和线程B两个线程同时执行addCount的这段代码，那么(as = counterCells) != null判断结果为false。
（3）、s=b+x，其中b是baseCount，x是addCount的第一个入参，表示总容量需要增加n个kv。
（4）、!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x))操作只有一条线程可以执行成功，假设，线程A执行将baseCount修改为s成功，则不用进入if判断的方法体中。而线程B则需要进入if的方法体中。那么，这里大家需要记住一点，总量是有两部分组成的，baseCount就是计数其中之一。
（5）、线程B由于执行CAS操作失败，那么对于这种所有竞争失败的线程，都会执行fullAddCount方法。这个方法里面就是计算总量的第二部分——CounterCell.value。
7.2.1、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; } } -
源码解释
（1）、fullAddCount代码较多，总体可以分为两部分：第一部分，如果获得随机数为0，则初始化当前线程的探针哈希值。第二部分，开启无限循环，利用CounterCell进行计数。
（2）、 第一部分比较简单。只需要记住一下几个方法的作用：
ThreadLocalRandom.getProbe()：用来获得随机数
ThreadLocalRandom.localInit()：初始化当前线程的探针哈希值
ThreadLocalRandom.advanceProbe(h)：更改当前线程的探针哈希值
（3）、第二部分，利用CounterCell进行计数一共分为3种情况：
case1：counterCells不为空且数组里面有元素。
case2：cellsBusy为0且counterCells为空。
case3：尝试修改baseCount的值。
（4）、这里我通过一张图，先介绍一下代码里的逻辑：
7.2.1.1、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吧。
/** *计算单元格表 *最初大小是2，如果线程较多，那么为了避免冲突，会进行2倍扩容。 *最终计算ConcurrentHashMap中所有元素的时候，其中一步，就是会统计 counterCells数组中所有元素的value值之和 */ private transient volatile CunterCell[] counterCells;@sun.misc.Contended static dinal class CounterCell{ volatile Long value; CounterCell(long x){ value=x; } } -
@sun.misc.Countended注解是做什么的？
（1）、Java8的@sun.misc.Contended注解（Contended：有争议的）
（2）、@sun.misc.Contended是Java8新增的一个注解，对某字段加上该注解则表示该字段会单独占用一个缓存行（Cache Line）。
（3）、这里的缓存行是指CPU缓存（L1、L2、L3）的存储单元，常见的缓存行大小为64字节。
（4）、（注：JVM添加-XX:-RestrictContended参数后@sun.misc.Contended注解才有效） -
单独使用一个缓存行有什么作用
（1）、避免伪共享。为了提高读取速度，每个CPU有自己的缓存，CPU读取数据后会存到自己的缓存里。而且为了节省空间，一个缓存行可能存储着多个变量，即伪共享。
（2）、 但是这对于共享变量，会造成性能问题，例如：当一个CPU要修改某共享变量A时会先锁定自己缓存里A所在的缓存行，并且把其他CPU缓存上相关的缓存行设置为无效。但如果被锁定或失效的缓存行里，还储了其他不相干的变量B，其他线程此时就访问不了B，或者由于缓存行失效需要重新从内存中读取加载到缓存里，这就造成了开销。所以让共享变量A单独使用一个缓存行就不会影响到其他线程的访问。 -
Java8之前的方案是什么
（1）、在Java8之前，是通过代码里手动添加属性的方式解决的，如：class LongWithPadding { long value; long p0, p1, p2, p3, p4, p5, p6; }（2）、一个long占8个字节，再添加7个long属性就会变成64个字节，刚好是一个缓存行大小。
（3）、但是注意，Java7开始JVM做优化时可能会把不用的变量给去掉，所以这种方法并不推荐使用。
（4）、适用场景主要适用于频繁写的共享数据上。如果不是频繁写的数据，那么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 }
7.2.1.2、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; } }
7.2.1.3、case3：尝试修改baseCount的值
-
如果修改baseCount成功，那么则跳出循环
else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x)) { break; // Fall back on using base }
7.2.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.3 第二种情况（存储的总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方法在上面都介绍过了。