• 源码解析系列：ConcurrentHashMap(2) - put方法和扩容

    文章目录

    前言

    这篇文章主要介绍ConcurrentHashMap的put方法。文章参考视频：concurrentHashMap源码解析。文章对于树化这一些操作不会详细说明，有兴趣可以去参考之前的HashMap的文章：

    第一篇文章：源码解析系列：HashMap(1)
    第二篇文章：源码解析系列：HashMap(2)
    第三篇文章：源码解析系列：HashMap(3)
    第四篇文章：源码解析系列：HashMap(4)
    第五篇文章：源码解析系列：HashMap(5)

    1. Put方法

    //put方法
public V put(K key, V value) {
	//调用putVal方法，onlyIfAbsent = false，说明可以被替换
 	return putVal(key, value, false);
}

//putVal方法
final V putVal(K key, V value, boolean onlyIfAbsent) {
	 //判断参数异常错误
     if (key == null || value == null) throw new NullPointerException();
     //基于key计算hash值，并进行一定的扰动
     int hash = spread(key.hashCode());
     //记录元素个数，如果超过8个就会变成红黑树
     int binCount = 0;
     //tab：数组
     //n：数组长度
     //f：要求的加入的hash所在的桶的首个节点元素
     //i：通过hash求出的要加入的下标
     //fh：f的hash值
     for (Node<K,V>[] tab = table;;) {
         Node<K,V> f; int n, i, fh;
         //如果数组是空的或者数组长度 == 0，没有元素，那么进行初始化操作
         if (tab == null || (n = tab.length) == 0)
             tab = initTable();
         //否则通过hash值求出所在的下标位置判断有没有元素
         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
         	 //使用cas操作把新的node节点添加到首个节点位置
             if (casTabAt(tab, i, null,
                          new Node<K,V>(hash, key, value, null)))
                 //跳出循环
                 break;                   // no lock when adding to empty bin
         }
         //如果首个节点的hash值是MOVED（-1），证明此时有其他线程正在进行扩容操作，那么需要协助扩容
         else if ((fh = f.hash) == MOVED)
         	 //协助扩容
             tab = helpTransfer(tab, f);
         else {
         	 //hash计算的桶位置元素不为空，且当前没有处于扩容操作，进行元素添加
             V oldVal = null;
             //加锁，防止多个线程都进来，注意这里是锁头节点，不是锁整个table，粒度更小了
             synchronized (f) {
             	 //再次检查首部元素有没有被修改，因为可能两个线程进入了else里面
             	 //而一个线程做了某些操作之后首个元素改变了，比如去树化等操作，导致前后结构不一致
             	 //所以下面还是需要再次判断一下的
                 if (tabAt(tab, i) == f) {
                 	 //如果hash值 > 0, 代表这是一个链表元素
                     if (fh >= 0) {
                     	 //从1开始
                         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;
                             }
                         }
                     }
                     //否则要判断f是什么元素节点
                     //TREEBIN(树): -2
                     //RESERVED(临时节点): -3
                     else if (f instanceof TreeBin) {
                         Node<K,V> p;
                         //设置binCount = 2，就是为了下面防止重复树化了
                         binCount = 2;
                         //调用树的添加方法，里面涉及到左旋右旋等操作
                         if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                        value)) != null) {
                             //获取旧节点
                             oldVal = p.val;
                             //put是可以进去的，那么替换新的value
                             if (!onlyIfAbsent)
                                 p.val = value;
                         }
                     }
                 }
             }
             //判断节点个数
             if (binCount != 0) {
             	 //如果大于等于8个，注意上面从0开始计数，而0对应第一个节点，所以这里实际上是 >= 9
                 if (binCount >= TREEIFY_THRESHOLD)
                 	 //树化操作
                     treeifyBin(tab, i);
                 //返回旧的数值
                 if (oldVal != null)
                     return oldVal;
                 break;
             }
         }
     }
     addCount(1L, binCount);
     return null;
 }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105

    上面是put方法的总体概述，下面是初始化方法，扩容方法放到后面说

    private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    //while循环
    while ((tab = table) == null || tab.length == 0) {
    	//如果sizeCtl < 0, 证明有其他线程正在初始化了，那么当前线程就不能初始化了
        if ((sc = sizeCtl) < 0)
            Thread.yield(); // lost initialization race; just spin
        //否则尝试把SIZECTL改成-1，代表正在初始化
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
            	//再次判断是不是为null了，避免多个线程都进入来到了else if 这里导致多次初始化
                if ((tab = table) == null || tab.length == 0) {
                	//判断sc > 0, 注意sc = sizeCtl，初始的时候等于传入的数组长度，如果自定义初始长度，那么就等于16
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    //创建node数组，长度为sc或者默认的16
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    //赋值给table
                    table = tab = nt;
                    //设置为原来的0.75
                    sc = n - (n >>> 2);
                }
            } finally {
            	//sizeCtl = sc，也就是说初始化之后sizeCtl = 0.75 initialCapacity, 也就是扩容因子 * 数组长度
            	//为什么要用 n - (n >>> 2) ? 因为位运算比较快
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32

    2. 其他工具方法

    1. spread

    //使用h的高16位和低16位进行异或操作，然后再和0x7fffffff相与，目的就是为了保证求出的这个hash值不为负数
static final int spread(int h) {
    return (h ^ (h >>> 16)) & HASH_BITS;
}

    • 1
    • 2
    • 3
    • 4

    0x7ffffff = 01111111111111111111111111111111，与操作之后第一位肯定不为1，所以肯定不是一个负数。



    2. casTabAt

    static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                              		  Node<K,V> c, Node<K,V> v) {
    //调用本地的compareAndSwapObject方法，比较并交换，把v的值赋值给c的值
    return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}

    • 1
    • 2
    • 3
    • 4
    • 5



    3. resizeStamp

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

    • 1
    • 2
    • 3
    • 4

    Integer.numberOfLeadingZeros：用于返回无符号整数从左往右数第一个0开始到最左边的0的个数，比如0000 0000 1000 0001 => 8。1 << (RESIZE_STAMP_BITS - 1 就是往左移15位，方法会生成一个扩容戳。这样移动之后第16位一定是一个1，就是为了在后面第一次扩容的时候再往前移动16位，使得最高位为1，保证这个数字小于0，因为这个数小于0就代表了正在扩容。

    当线程扩容的时候会进行rs << RESIZE_STAMP_SHIFT) + 2操作，而 rs 就是 resizeStamp(tab.length)，此时这个结果整体再向左移动16位，最高位肯定为1，是一个负数。为什么要+2呢，因为前一篇文章说过 sizeCtl = -1的时侯表示正在初始化，而 sizeCtl < -1 的时候表示数组正在扩容，并且线程长度为 -(1 + n)，-2 = -(1 + 1)，所以低16位代表了线程数目+1，2就表示有一个线程正在扩容。


    3. 扩容方法

    下面就顺着 put 方法的扩容来看

    1. 计数

    在第一次 PUT 或者之后的 PUT 达到了扩容阈值的时候，就会开始扩容，但是在这之前会先计数一次。也就是 PUT 方法的最后有一个addCount(1L, binCount)。

    注意下面的 addCount 方法中传入的参数是 x = 1 以及 check = 节点个数(链表) 或者 2(树)，总之 check > 0。这个计数的原理和 LongAdder 的原理很类似。在整个 addCount 方法中，可以分为三部分：

    • ① CounterCell数组不为空，优先利用数组中的CounterCell记录数量
    • ② 如果数组为空，尝试对baseCount进行累加，失败后，会执行fullAddCount逻辑
    • ③ 如果是添加元素操作，会继续判断是否需要扩容

    从上面的三点，我们可以得出两个结论，第一个 if 做的是更新值的操作，第二个 if 做的是判断扩容的操作。在第一个 if 中有一个判断，如果 check <= 1, 那么直接 return，什么情况下会 return 呢，追踪源码可得：在 replaceNode、clear会传入一个 -1，这时候直接返回，也就是说如果是对于remove操作或者 clear这类删除的操作，是不会考虑扩容的。但是在 compute 方法中如果是树节点，那么传入的 check = 1，这里不知道为什么不需要判断是否需要扩容。

    private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        //下面的if的逻辑：
        //1. 如果counterCells ！= null，那么优先使用counterCells来计数，第一次进入为null
        //2. 继续对baseCount进行添加，注意下面的BASECOUNT设置为baseCount+1，x = 1
        //3. 如果对baseCount添加失败，证明此时有其他线程可能正在添加，那么当前线程进入if，利用counterCells 进行添加数据
        if ((as = counterCells) != null ||
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            //a: counterCells 的一个随机下标
            //v：a的值
            //m：counterCells数组长度-1，为什么-1呢，是因为counterCells数组长度也是2的n次方，-1之后值就是xxxx....xx11111,
            //	 大大提高了counterCells 计算的利用率，每个下标都有可能取到。
            CounterCell a; long v; int m;
            //标识是否有多线程竞争，初始默认为true
            boolean uncontended = true;
            //下面就是调用对counterCells进行加法的过程
            //1. 当as == null，第一次进入要初始化，调用fullAddCount
            //2. 当 (m = as.length - 1) < 0，同样代表了数组为空，调用fullAddCount
            //3. 当获取的桶位元素为空，也就是说随机add的下标为null，也要调用fullAddCount
            //4. 如果上面三个都满足，证明此时可以累加了，但是CAS累加失败，证明多线程竞争严重，进入fullAddCount
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                //里面就是对这个数组的特殊处理，uncontended = false，因为上面CAS失败了
                fullAddCount(x, uncontended);
                return;
            }
            //就是上面说的，如果是remove这些方法，直接return了
            if (check <= 1)
                return;
            //sumCount里面就是把baseCount和counterCells数组相加，最终的结果就是map的size
            s = sumCount();
        }
        //进入下面判断扩容的条件
        //1. put方法
        //2. 在上面if 判断成功了，也就是说利用baseCount添加成功了，就会来到下面
        if (check >= 0) {
        	//tab: 原来的数组
        	//nt: 新建的table
        	//n: 旧数组的长度
        	//sc: sizeCtl
            Node<K,V>[] tab, nt; int n, sc;
            //sizeCtl在这里是扩容的阈值，如果 s >= 扩容阈值，代表要扩容了
            //数组不为空，更加要扩容
            //数组的长度 < 1 << 30的时候才扩容，因为扩容是原来的2被，如果大于1 << 30，那么扩容之后就大于Integer.MAX_VALUE, 溢出
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
                //同样求出扩容戳，根据数组长度求出来，这是一个很大的正数，上面说过第16位一定是1
                int rs = resizeStamp(n);
                //sc < 0, 表明有线程正在扩容，会帮助扩容
                if (sc < 0) {
                	//下面第一个if就是判断到底扩容完成了没有
                	//1. (sc >>> RESIZE_STAMP_SHIFT) != rs
                	//		sc = (rs << RESIZE_STAMP_SHIFT) + 2，+2可以忽略不记，因为sc右移16位之后，后面16位的数值基本起不到作用，
                	//		所以这里只要有线程在扩容默认就是false，什么时候为true呢？当扩容完成了，sc被赋值为新的阈值，这时候就是不等于了
                	//2. sizeCtl == rs + 1，意思就是扩容完成了，这里为什么等于rs + 1没有找到很好的解释
                	//3. sc == rs + MAX_RESIZERS，表示线程数已经超过最大的线程数了，不需要再使用多线程扩容
                	//4. (nt = nextTable) == null表示扩容完成，扩容后的数组已经被销毁
                	//5. transferIndex <= 0表示任务已经领完了，不需要线程再来领任务
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        //完成扩容，break
                        break;
                    //没有完成扩容的话，就协助扩容，并且sizeCtl+1，表示协助扩容的线程数+1
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    	//这里传入的nextTab = nt，因为已经有线程在扩容了。
                        transfer(tab, nt);
                }
                //没有其他线程在进行扩容，达到扩容阈值后，给sizeCtl赋了一个很大的负数
                //然后这个线程开始扩容，注意这里SIZECTL从sc改成了(rs << RESIZE_STAMP_SHIFT) + 2
                //RESIZE_STAMP_SHIFT = 16
                //上面rs第16位为0，再次左移16位之后，最高位为1，是一个负数
                //详细的解析可以看上面的resizeStamp方法，总之这里就是设置了线程数目为1，初始值为-2，表示有一个线程在扩容
                //当然这个初始值是低16位的初始值
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    //进行扩容操作，注意这传入的nextTab = null, 因为是第一次
                    transfer(tab, null);
                //计算总数，再次while判断，扩容成功侯，while循环就判断失败了，因为此时s肯定小于新的阈值，退出
                s = sumCount();
            }
        }
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85

    完成上面的代码解析后，下面来总结下addCount计数流程：

    1. 首先判断 counterCells 数组是不是为 null，如果是就尝试调用 baseCount 计数，如果不是就优先使用 counterCells 计数。
      • 如果 baseCount 计数失败，证明此时有其他线程正在尝试 CAS 对 baseCount add 操作，那么这时候使用 counterCells 计数
      • 计数的时候首先要计算出要添加的位置
      • 然后判断，如果数组为空或者使用 CAS 在这个位置 add 失败，证明数组这个下标位置竞争比较大，此时进入 fullAddCount 流程
      • 否则就在数组的某一个位置 add 成功
    2. 到这里第一步维护 map 的节点数目完成，至于 map 的节点个数就是等于 baseCount 和数组的总和
    3. 下面需要判断数组是否需要扩容，如果需要
      • 首先判断是否有线程已经在扩容了，如果是就协助扩容
      • 如果没有线程在扩容，那么就进入扩容

    画个图就是：
    在这里插入图片描述

    注意一下：这里的线程协助扩容的代码和下面的 helpTransfer 是一模一样的。最后简单总结下addCount流程，就是首先去维护节点个数，维护的方式就是通过 baseCount 和 counterCells 数组下标计数的方式，线程找到某一个位置，然后这个位置的值 + 1，就算维护成功了。维护完节点个数之后就要来判断是否需要扩容，如果需要并且已经有线程在扩容了，就协助扩容，否则就自己设置一些初始变量，开始扩容。


    2. fullAddCount流程

    从上面可以看出来，当数组为空的时候或者使用CAS失败的时候会进入这个流程，那么前面所说的就意味者进入这个流程代表了：第一次扩容，或者 baseCount 添加失败且数组没有初始化，或者线程竞争比较激烈。那么在里面肯定就是要解决这些问题，所以里面必定会涉及到数组的初始化和如何解决线程竞争激烈的问题。那么下面就来看看这里面的代码，注意进入这里的时候传入的 wasUncontended = false, x = 1。

    在看代码之前首先需要知道的是，fullAddCount 本质就是对于 CounterCell 数组取一个随机值来进行 CAS 操作，然后根据冲突程度和数组长度判断要不要去扩容。

    private final void fullAddCount(long x, boolean wasUncontended) {
	
    int h;
    //获取当前线程的hash值
    if ((h = ThreadLocalRandom.getProbe()) == 0) {
    	//如果为0，PROBE这个变量没有被初始化
        ThreadLocalRandom.localInit();      //初始化操作
        h = ThreadLocalRandom.getProbe();	//再次获取
        wasUncontended = true;				//设置为true，
    }
    //标识是否有冲突，如果最后一个桶不是null，那么为true
    boolean collide = false;                // True if last slot nonempty
    //下面开始死循环
    for (;;) {
    	//as：数组
    	//a：找到添加的位置
    	//n：长度
    	//v：baseCount
        CounterCell[] as; CounterCell a; int n; long v;
        //1. 首先判断如果数组为不为空才进入数组的添加流程
        if ((as = counterCells) != null && (n = as.length) > 0) {
        	//随机求出一个下标进行add，如果是null，那么就对这个下标进行初始化
            if ((a = as[(n - 1) & h]) == null) {
            	//等于0表示没有其他线程在修改
                if (cellsBusy == 0) {
                	//创建一个CounterCell对象，赋值x
                    CounterCell r = new CounterCell(x);
                    //cellsBusy == 0表明没有其他线程在初始化操作，然后当前线程尝试CAS修改为1表明有线程操作
                    if (cellsBusy == 0 &&
                        U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
                        //成功之后，开始初始化
                        boolean created = false;
                        try {
                        	//再次检查其他线程有没有已经初始化，因为可能多个线程来到上面的这个if，而这多个线程都是来初始化的
                            CounterCell[] rs; int m, j;
                            if ((rs = counterCells) != null &&
                                (m = rs.length) > 0 &&
                                rs[j = (m - 1) & h] == null) {
                                //确认完还没有初始化之后，进行初始化
                                rs[j] = r;
                                //标志位设置为true
                                created = true;
                            }
                        } finally {
                        	//相当于释放锁
                            cellsBusy = 0;
                        }
                        //退出循环
                        if (created)
                            break;
                        //否则初始化失败，证明肯定有其他线程初始化了，那么继续for循环找位置添加数据
                        continue; 
                    }
                }
                //上面判断cellsBusy == 0失败，证明有线程竞争
                collide = false;
            }
            //wasUncontended为false有两种含义
            //1.线程取随机变量的时候PROBE有没有初始化
            //2.在进入这个方法之前是调用了CAS对CounterCell赋值，但是失败了
            //所以下面标识为true，重新去求hash值然后确定新下标进行add
            else if (!wasUncontended)       // CAS already known to fail
                wasUncontended = true;      // Continue after rehash
            //重新计算了hash值后，对应的桶位依然不为空，先对value尝试累加
            //如果累加成功了，那么结束循环
            //否则继续下面的流程
            else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
                break;
            //再次判断如果counterCells != as, 表明数组可能被其他线程改变了，具体为扩容
            //或者n >= NCPU，数组长度大于CPU个数，由于线程数 <= CPU个数。此时再多线程也没有什么用了
            //重新计算线程hash值
            //这一步和下面的一步就是为了阻止无限扩容，来到了这一步如果长度大于CPU核数了，那么扩容就没什么必要了
            else if (counterCells != as || n >= NCPU)
            	//设置为没有冲突
                collide = false;            // At max size or stale
            //当没有冲突，修改为有冲突，并重新计算线程hash，作为下一次扩容的标识
            else if (!collide)
                collide = true;
            //否则线程桶位不是空，而且修改不成功，并且现在数组长度也没有超过CPU的核数，那么就尝试去扩容
            else if (cellsBusy == 0 &&
            		 //扩容就要尝试去修改CELLSBUSY为1
                     U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
                try {
                	//如果counterCells == as，表明其他线程还没有扩容这个数组
                	//有可能这个数组被其他线程扩容完成了，那么当前的数组就不需要再次扩容了
                	//因为这个线程如果再扩容，不确定长度是不是超过了
                    if (counterCells == as) {
                    	//n << 1 = 2 * n, 长度扩展为原来的2倍
                        CounterCell[] rs = new CounterCell[n << 1];
						//然后把旧数组的值移动到新数组上
                        for (int i = 0; i < n; ++i)
                            rs[i] = as[i];
                        //赋值给counterCells
                        counterCells = rs;
                    }
                } finally {
                	//最后设置为0，释放锁
                    cellsBusy = 0;
                }
                //设置为false，然后和上面一样，再走一遍流程
                collide = false;
                continue;              
            }

			//重新计算hash值，和上面几个else if同级
            h = ThreadLocalRandom.advanceProbe(h);
        }
        //2. 数组为空，那么首先就要初始化cellsBusy 
        //初始化之前需要把cellsBusy用CAS修改为1，cellsBusy也可以看成锁，1表示有线程正在操作
        else if (cellsBusy == 0 && counterCells == as &&
                 U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
            //是否初始化了
            boolean init = false;
            try { 
            	//初始化
            	//首先判断counterCells == as，是为了防止过程中别的线程修改了
                if (counterCells == as) {
                	//初始大小为2
                    CounterCell[] rs = new CounterCell[2];
                    //初始赋值
                    rs[h & 1] = new CounterCell(x);
					//赋值给counterCells 
                    counterCells = rs;
                    //设置为true
                    init = true;
                }
            } finally {
            	//解锁了
                cellsBusy = 0;
            }
            if (init)
            	//退出for循环，回到上一层，结束整个方法
                break;
        }
        //数组为空并且有其他线程在修改数组，那么线程尝试再次修改baseCount，如果修改成功，就break，否则循环
        else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
            break;                         
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138
    • 139

    看完整个代码，这里面都没有用到加锁的操作，而是用了cellsBusy 去模拟加锁的过程，并且里面的操作都是用的CAS，每一步都尽量考虑周全，下面就来概括下整个 fullAddCount 流程。

    1. 首先初始化一下线程的PROBE，这个值就是用来获取数组的随机下标的
    2. 判断如果 counterCells 数组没有初始化，并且抢到了 cellsBusy ，就开始初始化
    3. 如果counterCells 数组没有初始化，但是多线程下没有抢到 cellsBusy 锁，那么就再次尝试对 baseCount 进行 CAS操作
    4. 如果counterCells 数组初始化了，那么进入 for 循环流程
      • 如果通过随机 hash 获取的下标数据为 null，那么就会尝试去抢到锁然后添加上数据节点 CounterCell 到数组的下标位置，初始值为x
      • 如果通过随机 hash 获取的下标数据不是 null，那么就判断有没有发生争用，如果没有，把 wasUncontended 设置为 true，重写计算 hash，对新的下标数组进行操作
      • 如果上面都不成立，证明此时要操作的下标有数据了，使用CAS尝试去加一下看能不能
      • 如果加不成功，就判断这时候数组有没有被其他线程修改过了或者说数组的长度有没有大于 CPU 核数，如果大于了，那么就没必要扩容。而是把冲突值collide设置为false
      • 判断冲突值，如果是false，修改成 true，作为下一次扩容的标识
      • 最后，由于前面都不成立，冲突值为true，并且数组长度也没有大于CPU的核数，那么就尝试获取锁开始扩容



    3. 数组扩容transfer

    经过上面大的两个方法之后，Map中节点数目已经维护完成了，下面就要继续判断是不是需要扩容，看 addCount 方法，进入扩容前，如果是没有线程在扩容，那么 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2，这就是初始值。如果有线程在扩容，那么就在 sizeCtl 原来的基础上 +1，代表协助扩容的线程数 + 1。

    首先要知道的是，扩容的模式就是协助扩容，也就是说每一个线程都负责一个区域的部分数据。

    private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
	 //n: 数组长度
	 //stride: 细分范围，每个线程要负责迁移的节点个数
     int n = tab.length, stride;
     //首先如果是单核CPU，也就是单线程，那么就拿数组长度和最小值16比较，也就是说最少都得负责16个
     //其次如果是多核的CPU，那么每个线程负责的数目是 ⇒ 总数/(8 * CPU个数)和16的最大值
     //所以最少也得负责16个
     if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
         stride = MIN_TRANSFER_STRIDE; // subdivide range
     //如果是扩容线程，此时新数组为null，第一次进入的时候这里就是null
     if (nextTab == null) {            // initiating
         try {
             @SuppressWarnings("unchecked")
             //初始化了
             Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
             nextTab = nt;
         } catch (Throwable ex) {
         	 //溢出了，赋值一个最大值
             sizeCtl = Integer.MAX_VALUE;
             return;
         }
         //新的数组
         nextTable = nextTab;
         //记录线程开始迁移的桶位，从后往前迁移
         transferIndex = n;
     }
     //新的数组的长度
     int nextn = nextTab.length;
     //已经迁移的桶位，会用这个节点占位（这个节点的hash值为-1--MOVED）
     ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
     boolean advance = true;	//用来标识扩容的下标是否还需要推荐的 
     boolean finishing = false; //用来标识扩容完成的，每个线程都完成后才会标识为true
     for (int i = 0, bound = 0;;) {
     	 //f: 第i个节点的首个元素
     	 //fh: f的哈希值
         Node<K,V> f; int fh;
         //开始默认是true
         //这个while里面就是在分配任务
         while (advance) {
             int nextIndex, nextBound;
             //i记录当前正在迁移桶位的索引值
             //bound可以看作是当前线程可以处理的区间的最小下标
             
             //1. --i >= bound成立，那么就代表了，此时线程任务还没有完成，那么不需要分配任务，退出while循环就行
             //2. finishing表示迁移完成了，那么结束分配任务
             //3. 如果 --i < bound, 那么表示当前线程任务已经完成了，那么需要判断下面的两个else if，如果还能派任务，就派任务
             //也就是说线程执行完一个任务之后还会回来领取另一个任务
             if (--i >= bound || finishing)
             	 //标识false，while循环结束
                 advance = false;
             //如果后续没有元素要迁移了，那么设置 i = -1，并且 advance  = false
             else if ((nextIndex = transferIndex) <= 0) {
                 i = -1;
                 advance = false;
             }
             //计算下一次任务的桶位，并把这个值赋值给transferIndex
             //计算过程就是nextIndex - stride，也就是transferIndex - stride
             //第一个线程进入的时候transferIndex = 数组长度，从最后一个开始
             //第二个线程进入的时候bound = transferIndex - stride，也就是数组长度 - 负责的桶位个数
             else if (U.compareAndSwapInt
                      (this, TRANSFERINDEX, nextIndex,
                       nextBound = (nextIndex > stride ?
                                    nextIndex - stride : 0))) {
                 //bound记录了本次迁移终点下标，迁移的过程是从后往前的
                 bound = nextBound;
                 //i是记录本次迁移最大的下标，
                 i = nextIndex - 1;
                 //advance = false，本次任务分配完成了
                 advance = false;
             }
         }
         // i < 0 证明是全部迁移完成了
         // i >= n 应该一直为false吧，n是数组长度，除非扩容成功了？可能是一个线程扩容成功了，另一个线程才到这个位置，避免多此扩容
         // i+n >= nextn: nextn是新数组的长度，如果i + 旧数组长度 >= 新数组长度 ⇒ i >= 旧数组长度，和上面的一样了？
         // 满足上面三个条件就是说没有更多需要迁移的桶位  
         if (i < 0 || i >= n || i + n >= nextn) {
             int sc;
             //如果已经完成了，就是说整个数组都完成迁移了
             if (finishing) {
             	 //这个变量是扩容完成就删掉了
                 nextTable = null;
                 //赋值新数组
                 table = nextTab;
                 //(n << 1) - (n >>> 1) = 1.5n, 也就是2*n * 0.75
                 //因为位运算更加快
                 sizeCtl = (n << 1) - (n >>> 1);
                 return;
             }
             //如果没有完成，只是单个线程任务完成迁移任务，那么就设置 sizeCtl = sizeCtl-1,代表一个线程已完成
             if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
             	 //判断当前所有扩容任务线程是否都执行完成
             	 //因为第一个线程扩容时设置的初始值是：(rs << RESIZE_STAMP_SHIFT) + 2 ⇒ (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2
             	 //所以正常情况下是不相等的
                 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                 	 //如果不相等，证明此时还有其他线程在帮忙扩容，那么当前线程结束就行
                     return;
                 //如果相等，那么证明已经没有线程帮忙了，这个是最后一个，因为初始值就是上面说的，此时设置finishing = true
                 //advance= true， 最后结束前检查一下
                 finishing = advance = true;
                 i = n; //需要再次循环一下检查整张表
             }
         }
         //如果没有迁移完成并且当前节点是null，证明已经被迁移过了或者是原本就没有结点数据
         //这时候把一个fwd节点设置到这个节点上面，表示这个地方已经迁移了，fwd的hash = MOVED(-1)
         //同时当其他线程添加节点到这里的时候就知道，这里已经被迁移过了，那么这个线程就去帮忙迁移其他的线程
         else if ((f = tabAt(tab, i)) == null)
             advance = casTabAt(tab, i, null, fwd);
         //如果节点不为空并且是MOVED，证明正在迁移，所以设置advance = true，帮忙分配任务
         else if ((fh = f.hash) == MOVED)
             advance = true; // 设置为true，上面的while循环就可以进去分配任务了
         else {
         	//否则就开始进行迁移，需要对f进行加锁，粒度更小
             synchronized (f) {
             	 //再次判断，防止其他线程已经迁移了
                 if (tabAt(tab, i) == f) {
                     Node<K,V> ln, hn;
                     //hash >0, 证明这是一个正常的结点
                     //下面整个迁移过程就和HashMap类似了
                     if (fh >= 0) {
                     	 //hash & n
                     	 //因为put的时候是根据hash & n-1put的，比如n = 16
                     	 //那么put的时候就是xxx...xxx1111 & hash
                     	 //现在hash & n = xxxx...xxx1 0000 & hash
                     	 //其实这里用新数组长度-1 & hash是一样的，都是最新的第5位产生区别
                         int runBit = fh & n;
                         //从头开始遍历
                         Node<K,V> lastRun = f;
                         for (Node<K,V> p = f.next; p != null; p = p.next) {
                         	 //next.hash & n
                             int b = p.hash & n;
                             if (b != runBit) {
                                 runBit = b;
                                 lastRun = p;
                             }
                         }
                         //经过上面的for循环之后，lastRun记录最后一个和之前不同的值
                         //比如原来链表的hash & n之后是 1 0 0 1 0 0 0 0，那么runBit = 0, lastRun 是第五个，因为第五个之后都是相同的了
                         if (runBit == 0) {
                         	 //如果是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)
                                 //=0就是低位
                                 ln = new Node<K,V>(ph, pk, pv, ln);
                             else
                                 //否则就是高位
                                 hn = new Node<K,V>(ph, pk, pv, hn);
                         }
                         //设置ln为低位
                         setTabAt(nextTab, i, ln);
                         //设置hn为高位
                         setTabAt(nextTab, i + n, hn);
                         //原来的数组设置上fwd节点
                         setTabAt(tab, i, fwd);
                         //再次分配任务
                         advance = true;
                     }
                     else if (f instanceof TreeBin) {
                         TreeBin<K,V> t = (TreeBin<K,V>)f;
                         TreeNode<K,V> lo = null, loTail = null;
                         TreeNode<K,V> hi = null, hiTail = null;
                         int lc = 0, hc = 0;
                         for (Node<K,V> e = t.first; e != null; e = e.next) {
                             int h = e.hash;
                             TreeNode<K,V> p = new TreeNode<K,V>
                                 (h, e.key, e.val, null, null);
                             if ((h & n) == 0) {
                             	 //如果hash & n == 0, 就是低位
                             	 //如果hash & n == 1, 就是高位
                                 if ((p.prev = loTail) == null)
                                 	 //低位头节点
                                     lo = p;
                                 else
                                     loTail.next = p;
                                 loTail = p;
                                 ++lc;
                             }
                             else {
                                 if ((p.prev = hiTail) == null)	
                                 	 //高位头节点
                                     hi = p;
                                 else
                                     hiTail.next = p;
                                 hiTail = p;
                                 ++hc;
                             }
                         }
                         //如果低位的树节点 <= 6, 去树化为链表
                         ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                             (hc != 0) ? new TreeBin<K,V>(lo) : t;
                         //高位也一样
                         hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                             (lc != 0) ? new TreeBin<K,V>(hi) : t;
                         //设置低位
                         setTabAt(nextTab, i, ln);
                         //设置高位
                         setTabAt(nextTab, i + n, hn);
                         //设置原数组为fwd
                         setTabAt(tab, i, fwd);
                         advance = true;
                     }
                 }
             }
         }
     }
 }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138
    • 139
    • 140
    • 141
    • 142
    • 143
    • 144
    • 145
    • 146
    • 147
    • 148
    • 149
    • 150
    • 151
    • 152
    • 153
    • 154
    • 155
    • 156
    • 157
    • 158
    • 159
    • 160
    • 161
    • 162
    • 163
    • 164
    • 165
    • 166
    • 167
    • 168
    • 169
    • 170
    • 171
    • 172
    • 173
    • 174
    • 175
    • 176
    • 177
    • 178
    • 179
    • 180
    • 181
    • 182
    • 183
    • 184
    • 185
    • 186
    • 187
    • 188
    • 189
    • 190
    • 191
    • 192
    • 193
    • 194
    • 195
    • 196
    • 197
    • 198
    • 199
    • 200
    • 201
    • 202
    • 203
    • 204
    • 205
    • 206
    • 207
    • 208
    • 209
    • 210
    • 211
    • 212
    • 213
    • 214
    • 215

    代码挺复杂，最后我们也总结一些整个流程：

    1. 首先根据CPU分配任务的个数，最小是16

    2. 如果是新数组位null，第一个线程进来扩容，那么就创建新数组，然后记录迁移的位置

    3. 开始 for 循环中的 while 循环，分配迁移范围

      • 判断自己的任务完成没有，如果没有完成就退出while循环，不需要再分配范围
      • 如果没有元素迁移，设置标记下标为-1，后续会判断
      • 如果整个数组还没有迁移完成并且自己当前没有任务或者已经弄完一个任务了，那么重新分配范围开始下一个迁移任务

    4. 判断一下还有没有更多的需要迁移

      • 如果没有了，判断一下全部迁移任务完成没，完成了就赋值新数组，然后新阈值，退出方法
      • 如果没有了，但是还没有全部完成，只是当前线程的任务完成了，那么把 sizeCtl 值 -1，代表一个线程任务完成

    5. 有更多的节点需要迁移，如果这个位置没有元素，那么设置一个fwd节点，表示数组正在迁移

    6. 如果这个节点已经被迁移，那么重新分配范围，因为这里的位置已经由其他线程负责

    7. 上面4、5、6都不满足，证明此时线程能够迁移这个节点

      • 对迁移的首节点加锁
      • 然后把链表或者树分为低位和高位
      • 最后分别把低位和高位分配到新数组的两个位置

    8. 完成扩容迁移



    4. 协助扩容helpTransfer

    协助扩容，如果是头节点的hash值是moved，那么表示这个节点正在扩容，那么其他线程就进入协助扩容流程一起帮助扩容操作。在旧数组被迁移过后会把原来的节点设置为ForwardingNode，里面的hash值是MOVED(-1)。协助扩容里面的内容和addCount的基本一模一样，这里不多说了。

    //协助扩容
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
	//nextTab：记录新的扩容数组
	//sc：sizeCtl
    Node<K,V>[] nextTab; int sc;
    //如果tab != null && f 是ForwardingNode节点，证明正在扩容
    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;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    而发生协助扩容的时机就是当线程发现数组的节点是一个 fwd 节点或者说在调整 map 里面的节点个数的时候，会去协助扩容。





    如有错误，欢迎指出！！！！

  • 相关阅读:
    企业的固定资产管理怎么操作
    金仓数据库兼容Oracle exp/imp的导出导入工具手册（4. 功能与实践 ）
    OkHttp原理分析总结
    Apache InLong 1.2 单机部署问题记录
    基于SSM+SpringBoot+Vue+ElementUI的校园疫情防控管理系统
    计算机毕业设计之java+ssm果蔬经营平台系统
    不要再使用Load方式加载数据到Hive了，这种方式很low，你造吗？
    Parquet 文件生成和读取
    使用node-pty报错Uncaught Error: This socket has been ended by the other party
    AI+医疗：使用神经网络进行医学影像识别分析 ⛵
  • 原文地址：https://blog.csdn.net/laohuangaa/article/details/126165663