JDK1.8中ConcurrentHashMap源码解析

- 一、ConcurrentHashMap使用场景
- 二、put方法的整体流程
- 三、第一种情况（初始化table数组）
- - 3.1、initTable源码如下：
  - 3.2、sizeCtl
  - 3.3、流程解释
- 四、第二种情况（寻址后的位置没有被占用）
- - 4.1、源码如下
  - 4.2、Abase和ASHIFT 变量
  - 4.3、tabAt 方法
  - 4.4、casTabAt 方法
  - 4.5、流程解释
- 五、第三种情况（如果寻址后的位置是正在迁移状态）
- - 5.1、源码
  - 5.2、helpTransfer方法
- 六、其他情况（将节点插入到链表中或者红黑树中。）
- - 6.1、源码如下
  - 6.2、流程解释
  - 6.3、第一部分：向链表中插入Node的操作
  - 6.4、第三部分：扩容或者转换元素存储类型的操作
  - - 6.4.1 、treeifyBin方法
    - 6.4.2、 tryPresize方法
    - - 6.4.2.1、第一个判断（table数组没有初始化完毕）
      - 6.4.2.2、第二个判断（数组超过最大值，或者扩容发生越界）
      - 6.4.2.3、第三个判断（table还是那个table，这个过程中没有被其他线程重建过
    - 6.4.3、transfer方法
    - - 6.4.3.1 步骤1：计算迁移时的步长
      - 6.4.3.2 步骤2：初始化nextTab
      - 6.4.3.3 步骤3：扩容及数据迁移
      - 6.4.3.3.1 case1：扩容迁移结束逻辑
        6.4.3.3.2 case2：如果下标i处没有节点，则不需要进行扩容迁移操作
        6.4.3.3.3 case3：下标为i的这个位置已经被处理过了
        6.4.3.3.4 case4：执行扩容迁移操作
    - 6.4.4、tableSizeFor方法
    - 6.4.5、numberOfLeadingZeros方法
- 七、addCount将ConcurrentHashMap中存储的k,v总数+1
- - 7.1、源码
  - 7.2 第一种情况（计算当前存入key-value的总数）
  - - 7.2.1、fullAddCount方法
    - - 7.2.1.1、case1：counterCells不为空且数组里面有元素
      - 7.2.1.2、case2：cellsBusy为0且counterCells为空
      - 7.2.1.3、case3：尝试修改baseCount的值
    - 7.2.2、sumCount方法
  - 7.3 第二种情况（存储的总kv数量达到了阈值，执行扩容）

一、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方法。由于该方法代码较多，我们只保留框架性质的代码，这样会更方便我们理解。如下所示：

    final V putVal(K key, V value, boolean onlyIfAbsent) {
        ... ...
        for (Node[] tab = table;;) {
            Node 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(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[] initTable() {
        Node[] tab; int sc;
        //table=null
        while ((tab = table) == null || tab.length == 0) {
            //sizeCtl =0
            //如果sizeCtl为-1，则表示table数组正在被别的线程初始化
            if ((sc = sizeCtl) < 0)
                
                Thread.yield();
             
            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[] nt = (Node[])new Node[n];
                        table = tab = nt;
                        //n>>>2=10000>>>2=00100=4 (16/2/2=4)
                        
                        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[] initTable() {
        Node[] 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)) {
            	 
               
               ...  此处代码省略 ...
               
               ...  此处代码省略 ...
            }
        }
        return tab; // 返回创建好的table数组
    }

四、第二种情况（寻址后的位置没有被占用） 4.1、源码如下

 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//f =null i=[在数组中的位置]
   //如果该下标返回的节点为空，则直接通过cas将新的值封装成node插入即可；
   //如果cas失败，说明存在竞争，则进入下一次循环
    if (casTabAt(tab, i, null,
                   new Node(hash, key, value, null))){
                       break; 
    }                 
}

其中，关键点是，我们要理解tabAt和casTabAt这两个方法的实现逻辑，那么在了解这两个方法之前，我们需要先了解两个变量的含义，即：Abase和ASHIFT。

4.2、Abase和ASHIFT 变量

源码部分

 private static final long Abase;
 static {
        try {
           ... 此处代码省略...
            Class ak = Node[].class;
            
            Abase = U.arraybaseOffset(ak);//16
            
            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");
             
            //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(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、源码

源码如下

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[] tab, Node[] 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 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 pred = e;
        if ((e = e.next) == null) {
            pred.next = new Node(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;
    
    if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { 
        try {
            if (table == tab) {
                Node[] nt = (Node[])new Node[n];
                table = nt;
                sc = n - (n >>> 2); 
            }
        } finally {
            
            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[] 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[] nt = (Node[])new Node[n << 1]; 
        nextTab = nt; 
    } 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 fwd = new ForwardingNode(nextTab);

2、数据结构如下所示：

3、开启无限循环

boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
    Node f;
    int fh;
    while (advance) {
        int nextIndex; 
        int 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; 
        }
    }
    ... 下面代码后续我们在分析 ...
}

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); 
        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) { 
    advance = casTabAt(tab, i, null, fwd);
}

2、解释

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

6.4.3.3.3 case3：下标为i的这个位置已经被处理过了

1、源码如下

else if ((fh = f.hash) == MOVED) { 
    advance = true; // already processed
}

2、解释

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

6.4.3.3.4 case4：执行扩容迁移操作

1、源码如下

synchronized (f) {
    if (tabAt(tab, i) == f) {
        Node ln; 
        Node hn; 
        if (fh >= 0) {
            int runBit = fh & n; 
            Node lastRun = f;
            
            for (Node 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 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(ph, pk, pv, ln);
                }
                else { 
                    hn = new Node(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 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;
        
        
        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
        }
        
        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
        }
        
        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
        }
        
        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))) {  
        fullAddCount(x, uncontended); 
        return;
    }
    if (check <= 1) {
        return;
    }
    
    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 (;;) {
    
    if ((as = counterCells) != null && (n = as.length) > 0) {
        f ((a = as[(n - 1) & h]) == null) {
          ...如果随机数h待插入的下标位置没有元素,则插入CounterCell ...
        }
        else if (!wasUncontended) { 
            wasUncontended = true;      
        }
        else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + 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的长度，以减少碰撞次数 ...
        }
        ... ...
    }
    
    else if (cellsBusy == 0 && counterCells == as && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
        ... 创建一个长度为2的CounterCell数组，将x赋值进数组，跳出循环 ...
    }
    
    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) { 
        if (cellsBusy == 0) {            // Try to attach new Cell
            CounterCell r = new CounterCell(x); // Optimistic create
            
            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; 
                        created = true;
                    }
                } finally {
                    cellsBusy = 0;
                }
                if (created) {
                    break;
                }
                continue;           // Slot is now non-empty
            }
        }
        collide = false; 
    }
    else if (!wasUncontended) { 
        wasUncontended = true;      // CAS already known to fail，Continue after rehash
    }
    else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + 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)) {
        try {
            if (counterCells == as) {// Expand table unless stale
                CounterCell[] rs = new CounterCell[n << 1]; 
                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吧。

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) { 
    if (cellsBusy == 0) {            // Try to attach new Cell
        CounterCell r = new CounterCell(x); // Optimistic create
        
        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; 
                    created = true;
                }
            } finally {
                cellsBusy = 0;
            }
            if (created) {
                break;
            }
            continue;           // Slot is now non-empty
        }
    }
    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]; 
            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]; 
            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 == 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);
    }
    
    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方法在上面都介绍过了。

JDK1.8中ConcurrentHashMap源码解析

Java相关栏目本月热门文章