ConcurrentHashMap(JDK1.8)put分析（一）

爱被打了一巴掌 2023-09-28 08:39 4阅读 0赞

#### 目录 ####

*  ConcurrentHashMap整体结构
 *  1.计算hash值
 *  2.initTable
 *   *  2.1初始化table
     *  2.2 关于CAS
     *  2.3思考
 *  3.通过hash值定位key-value位置
 *  4.判断ConcurrentHashMap是否正在扩容
 *  5. 判断节点的类型，并添加节点；
 *   *  5.1表转换成红黑树
 *  6.addCount();
 *   *  6.1 ConcurrentHashMap多线程计数原理
     *  6.2 addCount的计数部分
     *  6.3 多线程计数的核心fullAddCount
     *   *  6.3.1为什么在线程在使用CounterCell对象计数时还是要使用CAS来更新值呢？
         *  6.3.2ConcurrentHashMap获取size
 *  7 ConcurrentHashMap扩容

## ConcurrentHashMap整体结构 ##

ConcurrentHashMap的数据结构与HashMap差不多，都是Node数组+红黑树+链表；ConcurrentHashMap中table的节点类型有3类：

*  Node节点，是链表类型的节点；这类节点hash 大于 0 ；
 *  在扩容时ConcurrentHashMap会有一个特殊的标志对象：ForwardingNode；hash值： MOVE(-1)
 *  在生成红黑树时，会生成一个TreeBin对象放在table中；hash值：TREEBIN（-2）

因此在put值时，会根据table中node的hash判断节点的类型；

*  hash>=0,是Node节点；
 *  hash=-2是TreeBin,表示该table元素是红黑树的节点；
 *  hash=-1是ForwardingNode；容器正在扩容
 *  hash = -3是ReservationNode，在调用computeIfAbsent方法时可能会使用的占位对象（本文不讨论computeIfAbsent方法，因此只关注前三种对象）

![01b8b6698914493988a5c0647e88ac18.png_pic_center][]

ConcurrentHashMap和HashMap的put值过程有些类似，ConcurrentHashMap的结构也是table + 链表+ 红黑树；在put值时，锁粒度是table的元素；也就是说，当put值时定位到table的第 i 个元素，那么就会给table\[i\]上锁；其他线程在put值时也定位到 i 时就需要等待获取锁；如果是其他位置则不需要等待锁，可以进行put操作。  
下面具体分析ConcurrentHashMap的put值过程

## 1.计算hash值 ##

ConcurrentHashMap在put值的时候会计算key的hash值，和HashMap类似；在ConcurrentHashMap计算hash值的是**spread**方法

public V put(K key, V value) {
        
            return putVal(key, value, false);
        }
    			||
    			||
    			\/
       final V putVal(K key, V value, boolean onlyIfAbsent) {
        
            if (key == null || value == null) throw new NullPointerException();
            int hash = spread(key.hashCode());
            ..
            ..
            ..
            }
            
    			||
    			||
    			\/
    
        static final int spread(int h) {
        
            return (h ^ (h >>> 16)) & HASH_BITS;
        }
        HASH_BITS = 0x7fffffff;(最大整数)

*  获取key的hashcode
 *  利用hashcode的高16位与低16位做" ^ "位运算，让高位也参与位运算，增加离散型。
 *  再与HASH\_BITS位运算，这是int最大正整数:最高位为0，其余位置为1，用来与hash值做 " & "运算；得到的结果保证最高位为0，其余位置不变；这样就保证得到的hash值是一个非负数；  
    举个例子：

![6fe92b897b0b4a618ae329181d8ffda4.png_pic_center][]

## 2.initTable ##

在put值时有几个判断，与HashMap类似

*  判断table是否已经初始化了；如果是没有初始化，就会先初始化table，再往table中添加值；
 *  根据hash值，计算出key-value在table的位置i，判断table\[i\]是否已经插入了值；如果没有插入值，就将该key-value插入到table\[i\]中；
 *  判断table\[i\]的hash值是否是MOVE,如果是MOVE表示正在扩容(ForwardingNode的hash值就是MOVE)，需要该线程帮忙将旧容器的值移动到新容器中
 *  将key-value插入到链表或者红黑树中
 *  判断是否需要扩容

### 2.1初始化table ###

因为是多线程环境，就需要考虑到如果有多个线程同时初始化table的情况；假如现在有三个线程：A,B,C同时判断到table == null。这个时候三个线程都会同时试图来初始化table，如果一个线程抢先修改了sizeCtl，他就可以初始化table，其余线程只需要等待初始化完成即可；

final V putVal(K key, V value, boolean onlyIfAbsent) {
        
            if (key == null || value == null) throw new NullPointerException();
            int hash = spread(key.hashCode());
            int binCount = 0;
            for (Node<K,V>[] tab = table;;) {
        
                Node<K,V> f; int n, i, fh;
                if (tab == null || (n = tab.length) == 0)
                    tab = initTable();
                    ..
                    ..
                    ..
                    ||
                    ||
                    \/
    
        /**
         * Initializes table, using the size recorded in sizeCtl.
         */
        private final Node<K,V>[] initTable() {
        
            Node<K,V>[] tab; int sc;
            while ((tab = table) == null || tab.length == 0) {
        
                if ((sc = sizeCtl) < 0)//有线程在初始化table，就让出cpu，等待table初始化就行；
                    Thread.yield(); // lost initialization race; just spin
                else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
        //修改sizeCtl的值，sizeCtl=-1 
                    try {
        
                        if ((tab = table) == null || tab.length == 0) {
        
                            int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                            @SuppressWarnings("unchecked")
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                            table = tab = nt;
                            sc = n - (n >>> 2);
                        }
                    } finally {
        
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            return tab;
        }

*  1.sizeCtl判断是否已经有线程初始化table，如果sizeCtl=-1，表示已经有线程对table初始化；这个时候会调用yield()让出cpu执行权
 *  2.通过CAS修改sizeCtl=-1，初始化table

这段代码，有意思的是进入到初始化table的分支时try\{\}finally\{\}代码块还要判断table==null，为什么呢？  
可能是对应这样一个场景：现在有 A ，B，C三个线程都判断table=null进入到initTable，A，B线程先获取到sizeCtl =0，此时A抢先修改到sizeCtl =-1开始初始化，而B线程修改失败，再次通过循环获取sizeCtl =-1，调用yield()放弃cpu执行权，等待A线程初始化table；A线程初始化完成之后修改sizeCtl 的值，修改后的sizeCtl >0；

而C线程比A，B线程慢一点，当C线程获取到sizeCtl的值时，A线程已经完成了table的初始化sizeCtl >0，C线程获取到的sc=sizeCtl >0,因此不会进入到休眠状态，会尝试修改sizeCtl 的值，这个时候没有其他线程竞争修改值，因此会修改成功；又会将sizeCtl 的值修改为 -1 ，**【此时：sc=A线程初始化之后sizeCtl 的值；sizeCtl =-1】**

进入到try代码块，判断table!=null,不会再重新初始化table，进入finally块，sizeCtl =sc；将sizeCtl 的值还原到A线程初始化时候的值；

添加的table=null的判断保证了只会有一个线程能初始化table；

流程：

![10b19c607100435a8eb6186a2f608cb9.png_pic_center][]

### 2.2 关于CAS ###

上面在多线程中竞争修改sizeCtrl的值没有使用**synchronized**，而是使用了Unsafe类的compareAndSwapInt也就是CAS；

U.compareAndSwapInt(object,offset,value,update);
    			||
    			||
    			||
    			\/
    U.compareAndSwapInt(this, SIZECTL, sc, -1);

看到这四个参数或许会感到有些蒙；这涉及到Java的实例对象在内存是如何存储的；  
Java对象有三部分：

*  头信息（hash,gc年龄，lock标志位，指向class对象的指针）
 *  实例数据，也就是成员变量
 *  填充对齐；（这部分不是必须的；如果：头信息+实例数据的字节数不是8的倍数，这个时候会填充字节让其满足8的倍数；）

下面用一个简单的图演示下：

![a1e9b8e8b8a84672a8e11e4263fb49c5.png_pic_center][]

在这个例子中，对象头+三个属性字段的字节数：24字节；刚好8的倍数，因此不用填充数据；从对象的内存布局，我们可以根据对象和对象属性的偏移量offset确定该对象的属性值存储位置；  
上面说了对象的布局，再来看上面的方法：**U.compareAndSwapInt(object,offset,value,update);** 就很容易理解了；  
举个例子：

U.compareAndSwapInt(user,12,34,48);
    对象user+offset 定位到user对象的id字段；
    value = 34,update =48;
    通过上面的定位，取到user对象的id值与value(34)比较；
    如果相等，就将user对象的id字段的值改为update(48);不相等，就不修改；

回到源码：

U.compareAndSwapInt(this, SIZECTL, sc, -1);
    this:是ConcurrentHashMap对象
    SIZECTL:SIZECTL = U.objectFieldOffset(k.getDeclaredField("sizeCtl"));===>获取到变量：sizeCtl的偏移量offset
    
    上面的代码：sc与sizeCtl比较，如果相等就将sizeCtl的值修改为-1；

### 2.3思考 ###

在try代码块中判断table是否已经初始化保证了只有一个线程能初始化table，这个和DCL(双重检测枷锁，差不多)；这段代码能修改吗？：

if ((sc = sizeCtl) < 0)
                    Thread.yield();
                	    ||
                	    ||
                	    ||
                	    \/
                if ((sc = sizeCtl) != 0)
                	Thread.yield();

因为只要**sizeCtl != 0**,就说明已经有线程在初始化table,其余线程都可以等待带线程初始化完成，这样当线程获初始化table的时就不用再判断table是否为null了，因为只能有一个线程能进入这个分支，对table进行初始化；  
这样看起来似乎没什么问题，但是忽略了一个问题，在ConcurrentHashMap的构造对象中有一个构造方法可以指定初始容量，而保存初始容量的变量就是sizeCtl ，也就是说如果指定了初始容量sizeCtl 值就 大于0；而指定初始容量时并不会初始化一个数组；因此不能修改成：(sc = sizeCtl) != 0，如果修改成这样，那么在指定初始容量之后，所有线程都不能初始化table了；  
同时也可以看到在初始化时利用了sizeCtl的值：

try {
        
                        if ((tab = table) == null || tab.length == 0) {
        
                        //sc是sizeCtl修改之前的值，如果指定了容量，那么值就保存在sizeCtl中；
                        //这里的(sc > 0),这个判断其实就是判断在创建ConcurrentHashMap时有没有指定初始容量；指定了就用指定的容量；
                        //没有指定容量就用默认值:DEFAULT_CAPACITY
                            int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//DEFAULT_CAPACITY=16;
                            @SuppressWarnings("unchecked")
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                            table = tab = nt;
                            sc = n - (n >>> 2);//sc = (3/4)*n
                        }
                    } finally {
        
                        sizeCtl = sc;//sizeCtl 扩容阈值 sizeCtl = 0.75 * n
                    }
                    break;

## 3.通过hash值定位key-value位置 ##

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
                }

这个分支其实就是判断table\[i\] = null，就将key-value值包装程成Node对象，放到table\[i\]中即可；

*  i = (n - 1) & hash；与HashMap一样，定位存储下标，因为使用了 **&** 运算，因此要求table的容量n一定是2的幂次倍；
 *  casTabAt（）将hash-key-value包装成node添加到table中，添加成功：返回true，进入分支break，结束put操作；不成功：进入下一次循环，继续put操作直到成功为止；

上面的casTabAt也使用到了Unsafe类的CAS方法，不再分析；简单说下定位 i 的计算；这个和hashMap中定位i的计算一样；正是因为使用了 **&** 这种位运算导致n的大小必须是2的幂次倍；否则会浪费table的存储空间，也会造成链表的长度快速增长；

2n \- 1 的数字二进制数有一个特点：**从最高位为1的位置 -> 最低位的每个位置都是1**；这样与hash值做 **&** 运算时的结果取值范围\[0，2n \-1\]的每一个数都能取到；反之，如果table的长度len不是 2n ，那么(len - 1)的二进制从最高位到最低位之间肯定有些位置的值为：0；导致(len-1) & hash的结果取值只会是\[0,len -1\]之间的部分数字，这样会使table的下标中有一些下标永远也获取不到，这就造成了这些位置的浪费；

**table长度为： 2n**  
![在这里插入图片描述][ed9f3d7b9d88465992bcccb2a12455c8.png_pic_center]  
**table长度不是： 2n**  
![在这里插入图片描述][e0127e8cae4b41dc8a83435eeb548dae.png_pic_center]

从上面举得2个例子就可以看出，table的长度是必须为2的次幂倍；否则会浪费掉大量的存储空间，同时可能会造成put，get效率降低；

## 4.判断ConcurrentHashMap是否正在扩容 ##

else if ((fh = f.hash) == MOVED)
                    tab = helpTransfer(tab, f);

在上一个分支，如果table\[i\] != null,就会在接下来的分支中，首先通过node节点的hash值判断，table是否处于扩容状态，**如果是扩容状态: hash == MOVE**;  
当判定在扩容时，会要求这个线程帮忙完成扩容:

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
        
            Node<K,V>[] nextTab; int sc;
            if (tab != null && (f instanceof ForwardingNode) &&
                (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
        
                int rs = resizeStamp(tab.length);
                while (nextTab == nextTable && table == tab &&
                       (sc = sizeCtl) < 0) {
        
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || transferIndex <= 0)
                        break;
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
        
                        transfer(tab, nextTab);
                        break;
                    }
                }
                return nextTab;
            }
            return table;
        }

扩容暂时在这里不分析，在后面会讲到；因为扩容是添加node导致，在添加元素之后会判断是否需要扩容；在前面提到过，在扩容时会产生：ForwardingNode；它的hash值就是MOVE(-1);由于扩容实在太复杂，并且扩容的原因并不在这里；扩容触发条件是添加node之后size到达扩容阈值触发扩容；因此把扩容的部分放在最后；（PS:扩容比较复杂，一直纠结这部分代码，容易引发头晕症状，还是继续下面简单的代码吧。。。。。。 ）

--------------------

![在这里插入图片描述][a035617d02a849689384cd8fc0c42d98.png_pic_center]

--------------------

## 5. 判断节点的类型，并添加节点； ##

判断节点类型，并添加值：

*  如果是链表，就将值打包成Node，添加到链表中；
 *  如果是红黑树，就将值打包成TreeNode添加到红黑树中；

这个过程和 HashMap一样；唯一的区别是，在多线程环境下，在确定table的下标之后，会获取table\[i\]对象的锁，只能有一个线程在table\[i\]所在的链表或者红黑树put值；其他线程要在table\[i\]中put值需要等待获取锁；

//	f = tabAt(tab, i = (n - 1) & hash)
    		
                else {
        
                    V oldVal = null;
                    synchronized (f) {
        //获取table[i]的对象锁
                        if (tabAt(tab, i) == f) {
        //判断是否被修改
                            if (fh >= 0) {
        //根据node的hash值判断是链表还是红黑树
                                binCount = 1;
                                for (Node<K,V> e = f;; ++binCount) {
        
                                    K ek;
                                   // 已经存在key，更新val，返回旧value
                                    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;
                               // key在链表中不存在，将key-value打包成Node插入到链表表尾
                                    if ((e = e.next) == null) {
        
                                        pred.next = new Node<K,V>(hash, key,
                                                                  value, null);
                                        break;
                                    }
                                }
                            }
                            else if (f instanceof TreeBin) {
        //红黑树
                                Node<K,V> p;
                                binCount = 2;
                                if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                               value)) != null) {
        
                                    oldVal = p.val;//p !=null ,说明在红黑树中已经存在key，只是更新了value；
                                    if (!onlyIfAbsent)
                                        p.val = value;
                                }
                            }
                        }
                    }
                    if (binCount != 0) {
        
                        if (binCount >= TREEIFY_THRESHOLD)
                            treeifyBin(tab, i);
                        if (oldVal != null)
                            return oldVal;
                        break;
                    }
                }
            }

为了更好的看清上面代码的整体逻辑，去掉部分代码，留下一个大的框架：

//当有一个线程 ： A ，锁定table[i]时，其余线程如果也是锁定这个位置，就必须等A线程，put完成之后；
    			//获取到table[i]的锁，才能在这个位置put值；
    			synchronized (f) {
        
    			//判断节点没有变动；如果变动了，就计入下一次循环；
    			//f发生改变的几种情况：
    			//1:有可能链表变成红黑树，
    			//2:可能容器在扩容；
    			//3:如果其他线程在之前进行了remove操作，导致f被删除，这种情况也不能直接put；
    			//还有一种情况：就是key-value的值被修改，这种情况对下一个线程put值没有影响；
    			//因此可以看到在很多地方，对节点进行操作前，都会先判断节点有没有改变；
    				if (tabAt(tab, i) == f) {
        
    					if(fh>=0){
        
    						binCount=1;
    						链表，会将新node添加到链表末尾，在这个过程中binCount会记录链表的长度，用来判断是否需要将链表修改为红黑树；
    						还有一种情况是这个key已经存在，就直接更新value；
    					}else if(f instanceof TreeBin){
        
    						binCount=2;
    						红黑树，这里的binCount就是只用来表示该线程抢到锁，已经put值了；
    					}
    				}
    			}
    		//说明已经添加了node；binCount=0是因为线程没有put值,
    		//f已经被其他线程删除，或者是正在扩容，或者是由链表改成了红黑树。。
    				if (binCount != 0) {
        
                        if (binCount >= TREEIFY_THRESHOLD)//这里是用来判断链表需不需要改成红黑树;
                            treeifyBin(tab, i);
                        if (oldVal != null)//key已经存在了，会将value更新；并返回旧值oldValue
                            return oldVal;
                        break;
                    }

流程：

![在这里插入图片描述][274695fe463d4113a8e7e75f14fc6b1e.png_pic_center]

### 5.1表转换成红黑树 ###

上面流程图中判断：f是TreeBin对象时表示红黑树节点；但红黑树的节点是由TreeNode保存；那么在链表转换成红黑树的转换过程是什么？  
在上述流程图中，可以看到链表转换成红黑树是由treeifyBin方法中进行的；treeifyBin的流程

*  判断table长度;如果len < 64,就将容器扩大2倍；
 *  len>=64;先将Node链表 -> TreeNode双向链表;再生成一个TreeBin对象放在table中；TreeBin的hash=-2；TreeBin的构造方法中将：链表 -> 红黑树；并且这个TreeBin对象包含链表转换前的头节点，以及转换后红黑树的根节点；

关于红黑树的生成不再分析。。。。。  
【后续更新RBT原理及代码 ([二叉搜索树BST][BST]，[AVL][]，[RBT][])】

![在这里插入图片描述][01f7c17c388e445196b47aecebfc1187.webp_pic_center]

## 6.addCount(); ##

这段代码在多线程环境下计数，并判断是否需要扩容。这部分代码实在太多，首先是在多线程环境下如何计数？再一个就是多线程如何扩容？后续再分析吧。

--------------------

![在这里插入图片描述][c8ffdfd7194845e2bf8f473733e32261.png_pic_center]

--------------------

### 6.1 ConcurrentHashMap多线程计数原理 ###

先说下多线程计数的原理吧，这里没有直接使用CAS来更新size的值；是为什么呢？

先来看下使用直接使用CAS有什么利弊？

![在这里插入图片描述][99b10a1de8764b8484a5c25b822da31b.png_pic_center]

在多线程环境下，直接使用单一变量用CAS更新size虽然可以保证数据的正确性，但是效率不高；如果线程竞争比较激烈多个线程同时更改size，就需要多个线程排队更改值；可以看到并发特别高的话，这样的做法效率就不高了，会让其他线程消耗cpu做空循环，占用了cpu又没做事；  
因此，ConcurrentHashMap采用了另一种高效的做法；设计了一种数据结构：基础值 + 数组；

int baseCount;
    	
    	CounterCell[] counterCells;
    
        @sun.misc.Contended static final class CounterCell {
        
            volatile long value;
            CounterCell(long x) {
         value = x; }
        }

如果是多个线程同时put值需要更新size，进入到addCount中；这个时候仍然会先使用CAS更新baseCount，但是只有一个线程能更新成功；其余线程会分别将值更新到counterCells，如下图：  
![在这里插入图片描述][d794cc38acce46a2806bd3f382a31e3c.png_pic_center]

这就是ConcurrentHashMap在多线程环境下更新size的方法；相比直接使用CAS对一个变量更新，这种方法显然更高效；

--------------------

ConcurrentHashMap的更新size的大体原理就是这样,但细节处有所不同；CounterCells数组是一个懒加载，也就是说，没有多个线程同时竞争修改baseCount时，不会生成CounterCells数组，直接用CAS修改baseCount；当有多个线程竞争修改baseCount时才会生成CounterCells数组，每个线程在各自的CounterCell中计数互不干扰；

![524a5b1fb9f4431785dd4464e1941255.png_pic_center][]

这里说明一下，即使生成了CounterCells数组，也不会立即将CounterCells数组中所有元素都初始化一个CounterCell对象；ConcurrentHashMap的设计突出一个懒加载，生成数组时的策略是这样，生成在CounterCell对象时也是需要用到时才生成CounterCell对象用于处理线程的计数；

### 6.2 addCount的计数部分 ###

addCount源码：

private final void addCount(long x, int check) {
        
            CounterCell[] as; long b, s;//b=baseCount,s = size;
            /**
    		case 0:
    				如果生成了counterCells数组，就直接通过线程定位到cell，在其中累加值;
            case 1:
            		如果counterCells没有初始化(没有生成counterCells数组)就通过CAS更新baseCount的值;
            case 2: 		
            		counterCells==null && 更新baseCount的值失败;  一定有其他线程同时竞争更改baseCount的值，这个时候会生成
            		counterCells数组；让更新失败的线程，在counterCells数组中更新累加;
            
            **/
            if ((as = counterCells) != null ||
                !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        
                CounterCell a; long v; int m;
                boolean uncontended = true;
                /**
    			1.as == null || (m = as.length - 1) < 0  ===> counterCells还没有初始化，说明是通过CAS更改baseCount失败从而进入
    				到该分支，这个时候会进入到fullAddCount中具体操作；
    			2.	(a = as[ThreadLocalRandom.getProbe() & m]) == null ; 
    				说明已经初始化了counterCells；
    				【ThreadLocalRandom.getProbe() & m】通过线程私有的Random随机数生成器生成的随机数来确定处理线程的CounterCell;
    				最后得到的处理对象：a==null，说明counterCells[i]==null没有对象；进入fullAddCount中累加;		
    			3. 如果a != null,则用CAS来修改CounterCell中value的值；如果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))) {
        
                    fullAddCount(x, uncontended);
                    return;
                }
                if (check <= 1)
                    return;
                s = sumCount();
            }
    *******************************************************************************************
    				//扩容部分暂且省略。  
        }

流程图：

![在这里插入图片描述][e72fd41ad012447d89eb542800c2ca52.png_pic_center]

### 6.3 多线程计数的核心fullAddCount ###

上图中可以看到，fullAddCount是核心方法，很多分支都会经过这个方法；关于cunterCells数组的初始化，扩容，以及cunterCell计数都在这里面；  
在看fullAddCount方法之前简单介绍fullAddCount处理的流程：

*  判断counterCells数组是否为空
    
     *  counterCells为空，初始化counterCells数组；并生成CounterCell对象用于计数；
     *  counterCells不为空，获取到CounterCell对象计数；如果对象为空，new一个对象来处理；如果计数失败，说明有多个线程在使用同一个CounterCell对象计数；这个时候将会扩容counterCells数组；（扩容长度最大到cpu的个数不再扩容）扩容之后再定位处理线程的计数；

下图是fullAddCount简易流程图：  
![在这里插入图片描述][7215d7e9d65d4d7c954117be2689c1b8.png_pic_center]

fullAddCount源码：

private final void fullAddCount(long x, boolean wasUncontended) {
        
            int h;
            if ((h = ThreadLocalRandom.getProbe()) == 0) {
        
                ThreadLocalRandom.localInit();      // force initialization
                h = ThreadLocalRandom.getProbe();
                wasUncontended = true;
            }
            boolean collide = false;                // True if last slot nonempty
            for (;;) {
        
                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, 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)       // CAS already known to fail
                        wasUncontended = true;      // 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);
                }
                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;
                }
                else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
                    break;                          // Fall back on using base
            }
        }

保留代码整体框架，省略掉部分代码:

private final void fullAddCount(long x, boolean wasUncontended) {
        
            int h;
            //每一个线程都可以通过ThreadLocalRandom，生成一个随机数，用来确定数组下标；
            if ((h = ThreadLocalRandom.getProbe()) == 0) {
        
                ThreadLocalRandom.localInit();      // force initialization
                h = ThreadLocalRandom.getProbe();
                wasUncontended = true;
            }
                       
            for (;;) {
        
                CounterCell[] as; CounterCell a; int n; long v;
                if ((as = counterCells) != null && (n = as.length) > 0) {
        
                	数组非空的处理逻辑；
    
                }
                /**
    			*
    			*初始化counterCells数组
    			*和initTable差不多，通过一个标志cellsBusy 的状态 来确定，有没有 线程初始化counterCells 数组
    			*
    			**/
                else if (cellsBusy == 0 && counterCells == as &&
                         U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
        
                    boolean init = false;
                    try {
                                   // Initialize table
                        if (counterCells == as) {
        //判断是否已经初始化，因为cellsBusy 的状态会还原到0；
                            CounterCell[] rs = new CounterCell[2];//length = 2，从2开始。每次扩容都增加一倍；
                            rs[h & 1] = new CounterCell(x);// h & (n-1) = h & 1;定位下标；new对象累加值；
                            counterCells = rs;
                            init = true;
                        }
                    } finally {
        
                        cellsBusy = 0;//状态还原
                    }
                    if (init)
                        break;
                }
                //当多个线程同时初始化counterCells 时，通过CAS竞争只有一个线程成功获取到初始化的权力，初始化数组；
                //其余没有初始化数组的线程进入这个分支，尝试更新baseCount;没有成功进入下一次循环继续尝试累加。
                else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
                    break;                          // Fall back on using base
            }
        }

counterCells 数组非空时的处理逻辑源码：  
counterCells 数组非空时处理过程的简易流程图：

![在这里插入图片描述][95c3b9a514014a0eb467a52e9cffe8f9.png_pic_center]

//通过h（线程生成的随机数）定位到数组下标，获取CounterCell 对象,为空就生成一个对象；
                    if ((a = as[(n - 1) & h]) == null) {
        
                        if (cellsBusy == 0) {
                    // Try to attach new Cell
                            CounterCell r = new CounterCell(x); // Optimistic create
                            //cellsBusy == 0，没有线程在修改counterCells数组，线程通过CAS修改cellsBusy 的值修改成功；
                            //线程获取到修改权
                            if (cellsBusy == 0 &&
                                U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
        
                                boolean created = false;
                                try {
                       // Recheck under lock
                                    CounterCell[] rs; int m, j;
                                    if ((rs = counterCells) != null &&
                                        (m = rs.length) > 0 &&
                                        rs[j = (m - 1) & h] == null) {
        //将生成的CounterCell对象给counterCells数组赋值；
                                        rs[j] = r;
                                        created = true;
                                    }
                                } finally {
        
                                    cellsBusy = 0;
                                }
                                if (created)//累加值成功，退出；
                                    break;
                                continue;           //累加值失败，下一次循环继续尝试累加值；
                            }
                        }
                        collide = false;
                    }
                  //在addCount中，CAS竞争修改CounterCell对象的value值，失败进入这个方法的；wasUncontended=false;
                  //如果下一轮CAS竞争失败，最后重新生成一个随机数；重新定位；
                    else if (!wasUncontended)       // CAS already known to fail
                        wasUncontended = true;      // Continue after rehash
    				//CAS修改value值，成功退出循环；竞争失败，可能触发扩容
                    else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
                        break;
                        //n >= NCPU 数组长度 大于等于cpu个数时，不再扩容；
                    else if (counterCells != as || n >= NCPU)
                        collide = false;           
                    else if (!collide)
                        collide = true;
                    else if (cellsBusy == 0 &&
                             U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
        //线程CAS修改cellsBusy 的值成功，获取到扩容权限
                        try {
        
                            if (counterCells == as) {
        // Expand table unless stale
                                CounterCell[] rs = new CounterCell[n << 1];//n << 1;扩大2倍；
                                for (int i = 0; i < n; ++i)
                                    rs[i] = as[i];//将原数组的值拷贝到新数组
                                counterCells = rs;
                            }
                        } finally {
        
                            cellsBusy = 0;//重新设置为0；
                        }
                        collide = false;
                        continue;                   // Retry with expanded table
                    }
                    //重新生成一个随机数，再定位一个位置；
                    //可能是线程CAS修改值失败；
                    //或者是new CounterCell对象时，CAS竞争失败；
                    //竞争扩容失败
                    //都会重新生成一个h，重新定位到数组的其他位置尝试累加值；
                    h = ThreadLocalRandom.advanceProbe(h);

总结：

fullAddCount简略流程图：

![在这里插入图片描述][de0abdcaa9974b8dbce29cca0de7df13.png_pic_center]

fullAddCount完整详细的流程图：

![在这里插入图片描述][5226e23951714070ae65cf1de29e8810.png_pic_center]

#### 6.3.1为什么在线程在使用CounterCell对象计数时还是要使用CAS来更新值呢？ ####

原因有以下2点：

*  在counterCells数组长度大于并发线程个数时：两个线程生成的随机数h不同，但是有可能定位到数组的同一个下标；这个时候如果两个线程同时进入到fullAddCount更新size就会产生冲突；这个情况下，要判断数组长度是否小于cpu个数；如果小于cpu个数首先尝试扩容；扩容失败时，线程会重新产生一个随机数，来获取一个新的下标解决冲突问题；
 *  在counterCells数组长度小于并发线程个数时；必然造成多个线程同时使用一个CounterCell；会通过扩容来解决冲突；（这种情况，必定是数组长度小于cpu个数）

![在这里插入图片描述][e8a807be14b148d8a24d44b391b7ca45.png_pic_center]

#### 6.3.2ConcurrentHashMap获取size ####

ConcurrentHashMap不是直接通过获取一个变量来获取size的；因为记录的方法：维护一个变量 baseCount + CounterCell数组；因此在获取size时，需要将counterCells数组中value的值累加，再加上 baseCount；  
获取size的方法源码：

public int size() {
        
            long n = sumCount();
            return ((n < 0L) ? 0 :
                    (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                    (int)n);//size超过int类型的范围，返回Integer.MAX_VALUE；没超过：long -> int;返回原值
                    
    				}
    
        final long sumCount() {
        
            CounterCell[] as = counterCells; CounterCell a;
            long sum = baseCount;
            if (as != null) {
        //counterCells数组不为空，就累加counterCells数组的每一个CounterCell对象中value值；
                for (int i = 0; i < as.length; ++i) {
        
                    if ((a = as[i]) != null)
                        sum += a.value;
                }
            }
            return sum;
        }

size的计算公式：

size =  ∑ 0 n − 1 C o u n t e r C e l l . v a l u e \\huge\\textcolor\{8B7D7B\}\{\\sum\_\{\\boldsymbol\{0\}\}^\{\\boldsymbol\{n-1\}\}\\boldsymbol\{CounterCell.value\}\}\\huge ∑0n−1CounterCell.value \+ baseCount

## 7 ConcurrentHashMap扩容 ##

宕机。待续。。

--------------------

在ConcurrentHashMap中，扩容是由多个线程同时参与的，这样会比较高效的完成数据的迁移。但这样就会有比较多的难点：

*  多线程下，在扩容时保证只能一个线程创建新容器；
 *  多线程数据迁移时，如何协作多个线程同时迁移数据？

在addCount方法中，更新元素个数之后，得到元素个数size与扩容阈值sizeCtrl比较；size >= sizeCtrl && len < 最大容量值MAXIMUM\_CAPACITY会扩容；  
源码：

//s:元素个数，sc = sizeCtl，n = table.length 
            if (check >= 0) {
        
                Node<K,V>[] tab, nt; int n, sc;
              //  size >= sizeCtrl  && len < 最大容量值MAXIMUM_CAPACITY 就会触发扩容；
                while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                       (n = tab.length) < MAXIMUM_CAPACITY) {
        
                    int rs = resizeStamp(n);
                    if (sc < 0) {
        //当已经有线程扩容的时候，后进入的线程会看到sc（sizeCtrl ） < 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();
                }
            }

--------------------

![在这里插入图片描述][11bf91c2ed4d48b7b7f6d8cd1ffc71ff.png_pic_center]

![在这里插入图片描述][fa95b1fd7558454d9be111f41cc8eb02.gif_pic_center]  
对put的分析到此结束；关于扩容方法transfer的分析，以及在扩容前的判断条件的分析会在下一篇文章中具体分析。。。

sizeCtrl在ConcurrentHashMap中的多个地方出现，并且含义各不相同，做一个总结：

*  在创建ConcurrentHashMap对象时指定了容量大小，会将这个值存储在sizeCtrl中；（容量值会被[tableSizeFor][] 修正为：2n，然后存入sizeCtrl）
 *  在初始化数组时，会将sizeCtrl的值设为负数表示有线程在初始化数组；
 *  在扩容时，将sizeCtrl的值设置为负数，表示有线程在创建扩容后的数组；
 *  在初始化或扩容完成后，将szieCtrl的值设置为：0.75 \* n；用做扩容阈值

--------------------

后续：[ConcurrentHashMap的扩容分析链接][ConcurrentHashMap]

--------------------

参考文章：

*  [并发之Striped64][Striped64]
 *  [扩容分析][Link 1]

[01b8b6698914493988a5c0647e88ac18.png_pic_center]: https://img-blog.csdnimg.cn/01b8b6698914493988a5c0647e88ac18.png#pic_center
[6fe92b897b0b4a618ae329181d8ffda4.png_pic_center]: https://img-blog.csdnimg.cn/6fe92b897b0b4a618ae329181d8ffda4.png#pic_center
[10b19c607100435a8eb6186a2f608cb9.png_pic_center]: https://img-blog.csdnimg.cn/10b19c607100435a8eb6186a2f608cb9.png#pic_center
[a1e9b8e8b8a84672a8e11e4263fb49c5.png_pic_center]: https://img-blog.csdnimg.cn/a1e9b8e8b8a84672a8e11e4263fb49c5.png#pic_center
[ed9f3d7b9d88465992bcccb2a12455c8.png_pic_center]: https://img-blog.csdnimg.cn/ed9f3d7b9d88465992bcccb2a12455c8.png#pic_center
[e0127e8cae4b41dc8a83435eeb548dae.png_pic_center]: https://img-blog.csdnimg.cn/e0127e8cae4b41dc8a83435eeb548dae.png#pic_center
[a035617d02a849689384cd8fc0c42d98.png_pic_center]: https://img-blog.csdnimg.cn/a035617d02a849689384cd8fc0c42d98.png#pic_center
[274695fe463d4113a8e7e75f14fc6b1e.png_pic_center]: https://img-blog.csdnimg.cn/274695fe463d4113a8e7e75f14fc6b1e.png#pic_center
[BST]: https://blog.csdn.net/m0_37550986/article/details/125617417?spm=1001.2014.3001.5502
[AVL]: https://blog.csdn.net/m0_37550986/article/details/125629666?spm=1001.2014.3001.5502
[RBT]: https://blog.csdn.net/m0_37550986/article/details/125690095?spm=1001.2014.3001.5502
[01f7c17c388e445196b47aecebfc1187.webp_pic_center]: https://img-blog.csdnimg.cn/01f7c17c388e445196b47aecebfc1187.webp#pic_center
[c8ffdfd7194845e2bf8f473733e32261.png_pic_center]: https://img-blog.csdnimg.cn/c8ffdfd7194845e2bf8f473733e32261.png#pic_center
[99b10a1de8764b8484a5c25b822da31b.png_pic_center]: https://img-blog.csdnimg.cn/99b10a1de8764b8484a5c25b822da31b.png#pic_center
[d794cc38acce46a2806bd3f382a31e3c.png_pic_center]: https://img-blog.csdnimg.cn/d794cc38acce46a2806bd3f382a31e3c.png#pic_center
[524a5b1fb9f4431785dd4464e1941255.png_pic_center]: https://img-blog.csdnimg.cn/524a5b1fb9f4431785dd4464e1941255.png#pic_center
[e72fd41ad012447d89eb542800c2ca52.png_pic_center]: https://img-blog.csdnimg.cn/e72fd41ad012447d89eb542800c2ca52.png#pic_center
[7215d7e9d65d4d7c954117be2689c1b8.png_pic_center]: https://img-blog.csdnimg.cn/7215d7e9d65d4d7c954117be2689c1b8.png#pic_center
[95c3b9a514014a0eb467a52e9cffe8f9.png_pic_center]: https://img-blog.csdnimg.cn/95c3b9a514014a0eb467a52e9cffe8f9.png#pic_center
[de0abdcaa9974b8dbce29cca0de7df13.png_pic_center]: https://img-blog.csdnimg.cn/de0abdcaa9974b8dbce29cca0de7df13.png#pic_center
[5226e23951714070ae65cf1de29e8810.png_pic_center]: https://img-blog.csdnimg.cn/5226e23951714070ae65cf1de29e8810.png#pic_center
[e8a807be14b148d8a24d44b391b7ca45.png_pic_center]: https://img-blog.csdnimg.cn/e8a807be14b148d8a24d44b391b7ca45.png#pic_center
[11bf91c2ed4d48b7b7f6d8cd1ffc71ff.png_pic_center]: https://img-blog.csdnimg.cn/11bf91c2ed4d48b7b7f6d8cd1ffc71ff.png#pic_center
[fa95b1fd7558454d9be111f41cc8eb02.gif_pic_center]: https://img-blog.csdnimg.cn/fa95b1fd7558454d9be111f41cc8eb02.gif#pic_center
[tableSizeFor]: https://blog.csdn.net/m0_37550986/article/details/117257269?spm=1001.2014.3001.5501
[ConcurrentHashMap]: https://blog.csdn.net/m0_37550986/article/details/125230667?spm=1001.2014.3001.5501
[Striped64]: https://www.cnblogs.com/gosaint/p/9129867.html
[Link 1]: https://www.cnblogs.com/gunduzi/p/13651664.html