JDK1.8中ConcurrentHashMap源码解析

x33g5p2x  于2021-11-21 转载在 Java  
字(24.4k)|赞(0)|评价(0)|浏览(205)

一、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方法在上面都介绍过了。

相关文章

微信公众号

最新文章

更多

目录