万字长文图文阅读 JDK8 ConcurrentHashMap源码带你吊打面试官

红太狼 2022-12-21 04:40 143阅读 0赞

### 文章目录 ###

*   *  ConcurrentHashMap结构
     *  存放的key-value的Node节点
     *  继承结构
     *  核心属性
     *  看懂几个方法
     *  核心方法public V put(K key, V value)
     *  核心方法private final void addCount(long x, int check)
     *  核心方法 public int size()
     *  核心方法private final void addCount(long x, int check)
     *  size+1是如何做到线程安全的
     *  核心方法void transfer(Node K,V \[\] tab, Node K,V \[\] nextTab) 扩容，最难的方法
     *  什么时候才会触发分配任务
     *  怎么判断所有任务已经分配完毕。
     *  再看什么时候会退出自旋
     *  帮助扩容的时机
     *  最后我们来大概的看看同步代码块中的代码
     *  最后的最后做个总结

## ConcurrentHashMap结构 ##

数组+链表+红黑树。链表大于8转红黑树，红黑树节点数小于6退回链表。结构跟HashMap是一样的。只是支持并发的操作。  
![在这里插入图片描述][202011141615544.png_pic_center]

## 存放的key-value的Node节点 ##

结构几乎一样，但是有个很重要的地方，val，next都被volatile关键字修饰，保证在多个线程之间的可见性。

static class Node<K,V> implements Map.Entry<K,V> { 
            final int hash;
            final K key;
            volatile V val;
            volatile Node<K,V> next;
    }

树形结构的Node，继承Node多了几个红黑书树树特征的属性。

static final class TreeNode<K,V> extends Node<K,V> { 
            TreeNode<K,V> parent;  // red-black tree links
            TreeNode<K,V> left;
            TreeNode<K,V> right;
            TreeNode<K,V> prev;    // needed to unlink next upon deletion
            boolean red;
    }

## 继承结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center]

## 核心属性 ##

*  **volatile Node<K,V>\[\] table**;  
    结构中存放数据的数组，初始化时并不会立马初始化，而是等到第一次插入时延迟初始化。大小始终是2的幂
 *  **volatile Node<K,V>\[\] nextTable**;  
    这个属性在扩容是很重要， 只有在扩容时才会用得到。可以简单理解为扩容后需要放置的新数组。
 *  transient volatile long baseCount;  
    基本计算表中元素个数，主要在无竞争时使用， 通过CAS更新。但是map的size！=baseCount，具体往下看
 *  **volatile int transferIndex**;  
    扩容时，线程需要负责转移的开始下标，扩容时分配任务时会频繁用到。
 *  **transient volatile CounterCell\[\] counterCells**;  
    很重要，用来计算个数的元素，后续在size方法，以及addCount方法时会用到，后续详细讲解。
 *  **volatile int cellsBusy**;  
    当counterCells需要扩容了，修改counterCells时的标识，在addCount方法时讲解
 *  **transient volatile int sizeCtl**;  
    默认为0，用来控制table的初始化和扩容操作  
    1 代表table正在初始化  
    N 表示有N-1个线程正在进行扩容操作：  
    如果table未初始化，表示table需要初始化的大小。  
    如果table初始化完成，表示table的容量，默认table大小的0.75倍

**可以看到上述的属性都被volatile关键字所修饰，这样便可以保证属性在线程之间的可见性，如果在配合上CAS操作即可完成读-改-写的原子性。**[volatile关键字][volatile]

## 看懂几个方法 ##

*  tabAt  
    拥有volatile语义的读取数组中的相应索引的数据，直接读取主内存中的数组数据。
 *  casTabAt  
    使用CAS的方式保证 读-改-写的原子性，保证线程安全。Java的CAS会使用现代处理器上提供的高效机器级别的原子指令，指令以原子方式对内存执行读-改-写操作。CAS有3个操作数，内存地址，旧的预期值A，要修改的新值B。当且仅当预期值A和内存地址中的值相同时，将内存值修改为B，否则什么都不做（这个读-改-写是原子的）。
 *  setTabAt  
    对setTabAt的调用始终在锁定区域内发生，并不需要完整的volatile语义，当前被编码为volatile 是保守的。

static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { 
            return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
        }
    
        static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) { 
            return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
        }
    
        static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { 
            U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
        }

开始看map的几个核心方法的实现。

## 核心方法public V put(K key, V value) ##

put时存在多种情况，并且需要支持并发线程安全。下面分步骤进行解析

*  数组table还未初始  
    直到put成功才退出

for (Node<K,V>[] tab = table;;) { 
        Node<K,V> f; 
        int n = tab.length;
        if (tab == null || tab.length == 0) { 
            tab = initTable();
        }     
    }

初始化代码，一进去就是个循环一直到table完成初始化。线程会根据 **sizeCtl（volatile保持线程可见性）** 的值，判断进行初始化操作，还是让当前线程从运行状态变为就绪状态，让出CPU。

*  sizeCtl小于0  
    说明已经有线程在做初始化的操作了。使当前线程从运行状态变为就绪状态让出CPU
 *  sizeCtl大于等于0  
    没有线程做初始化操作，sizeCtl表示table需要初始化的大小。进行CAS操作改变sizeCtl的值为-1，代表已有线程竞争到初始化table的权利。准备开始初始化。  
    初始化过程：  
    如果table未初始化sizeCtl表示table需要初始化的大小，New出数组。属性赋值给table。 table初始化完成，sizeCtl转而表示需要扩容的阈值，默认table大小的0.75倍。

private final Node<K,V>[] initTable() { 
            Node<K,V>[] tab; 
            //记录sizeCtl的值，sizeCtl后续CAS修改-1了
            int sc;
            //只要table没完成初始化就尝试初始化
            while ((tab = table) == null || tab.length == 0) { 
                if ((sc = sizeCtl) < 0)
                    //让出Cpu的执行
                    Thread.yield(); 
                //线程CAS,其中会有一个线程CAS成功
                else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { 
                    //CAS成功的线程开始初始化table
                    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;
                            //0.75倍
                            sc = n - (n >>> 2);
                        }
                    } finally { 
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            //直到table完成初始化退出循环，返回table
            return tab;
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 1]  
**从上述代码可以看到，数组的初始化过程只能一个线程完成，其中使用了sizeCtl来表示是否已经有线程在初始化了。  
sizeCtl是volatile的保证了线程的可见性，在修改时采用CAS方式保证只有一个线程能进行初始化的操作，其余线程都需要等待。**  
初始化完成后再接着往下看，会发现其他else 语句都进不去了，元素还没put进去呢，这不白瞎了。别忘记了开头进来是个一直的循环的行为，等table初始化完成后，循环还没结束。只有在元素插入后才会break跳出循环。table初始化完成后进入下一次循环。

*  存放时数组table相应位置为Null

else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { 
        if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))
            break;
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 2]  
线程2竞争CAS失败，依旧需要下次循环，下次循环时table此处已经存放了Node节点（忽略移除的情况），接下来再看线程2要怎么处理。

*  存放的结构是链表（数组中相应位置已有节点）  
    上述到线程1和线程2CAS竞争，线程2失败，再次进到循环处。前两个条件已经进不去了（table已初始化，table存在元素）先暂时忽略正在扩容的情况，后面说。线程2只能进到最后一个else。代码挺长耐心看。  
    代码第三行看到 synchronized 锁住头节点，其实只要看到就可以松一口气，因为其中是同步代码块同时只有一个线程可以进入，只要考虑单线程的就好了。基本和HashMap插入元素一样了。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 3]

//锁住头结点
    synchronized (f) { 
        if (tabAt(tab, i) == f) { 
            //头结点的Hash值如果小于0代表在扩容，这个后面扩容时说
            if (fh >= 0) { 
                //统计链表的节点个数
                binCount = 1;
                for (Node<K,V> e = f;; ++binCount) { 
                    K ek;
                    //存在key相同的情况，替换其中的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相等，构造节点插入到链表后
                    if ((e = e.next) == null) { 
                        pred.next = new Node<K,V>(hash, key,
                        value, null);
                        break;
                    }
                }
            }
        }
    }
    if (binCount != 0) { 
        //大于阈值8需要进行链表转红黑树
        if (binCount >= TREEIFY_THRESHOLD)
            treeifyBin(tab, i);
        if (oldVal != null)
            return oldVal;
        break;
    }
    addCount(1L, binCount);

*  存放的结构是 红黑树  
    锁住头节点，如果头节点是TreeBin结构，那么调用putTreeVal方法将节点插入红黑树。至于红黑树的插入与删除篇幅较长，下次在另一篇文章中介绍。

//锁住头节点，形成同步代码块
    synchronized (f) { 
        if (f instanceof TreeBin) { 
            Node<K,V> p;
            binCount = 2;
            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,value)) != null) { 
                oldVal = p.val;
                if (!onlyIfAbsent)
                    p.val = value;
            } 
        }
    }
    addCount(1L, binCount);

**对于上述 synchronized (f)，做一个补充，因为只是锁住了单个节点，并不会影响操作其他链表或者其他红黑树的线程。只要线程tabAt(tab, i = (n - 1) & hash)获取的节点不一致，就不会产生冲突。**

## 核心方法private final void addCount(long x, int check) ##

再线程添加完节点后最后都调用了addCount(1L, binCount);到目前为止，是不是没有发现扩容在何处触发，因为我们说当hashMap存放的元素>容量\*负载因子（默认0.75）就会扩容，那么这个方法一定是在put增加元素时检查是否需要扩容。  
一起看看这个方法干嘛用的，是否有触发扩容。再看这个增加Map存放元素个数方法前，先看看size()方法，是怎么计数的。

## 核心方法 public int size() ##

先大概看一下源码，发现并不是直接返回baseCount这个属性。而是遍历counterCells数组中所有的值累加到baseCount上返回。

public int size() { 
            long n = sumCount();
            return ((n < 0L) ? 0 :(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int)n);
        }

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;
        }

那么为什么这个中的所有值和baseCount累加在一起就等于Map的size呢（**其实在并发情况下这个数字仍旧可能是不准确的，CounterCell遍历到中间的时候，其他线程做了put操作，修改了baseCount或者CounterCell数组前部分的值**），带着这个疑问可以看addCount方法了。先看看CounterCell的结构很简单，只有一个volatile修饰的value值。**这个时候再回头看addCount(long x, int check)方法**

@sun.misc.Contended static final class CounterCell { 
            volatile long value;
            CounterCell(long x) {  value = x; }
        }

## 核心方法private final void addCount(long x, int check) ##

private final void addCount(long x, int check) { 
            CounterCell[] as; long b, s;
            //如果counterCells不为空，说明不是第一次add，直接进入if
            //只有ounterCells为空，才会CAS竞争baseCount
            //CAS操作失败进入if 避免多个线程竞争baseCount
            if ((as = counterCells) != null ||
                !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { 
                CounterCell a; 
                long v; 
                int m;
                boolean uncontended = true;
                //以下条件只要有一个为true，就会执行fullAddCount(x, uncontended)，贼长的一个方法。
                //CounterCell数组为空
                //随机得到的CounterCell数组中的一个节点为Null（ThreadLocalRandom.getProbe() 生成每个线程的一个随机数，&m 取余CounterCell数组的长度得到的节点）
                //去CAS修改节点中的值失败
                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;
                }
            }
        }

从判断是否要进入fullAddCount();是个递进的条件，CounterCell数组为空，需要进入fullAddCount做初始化数组的动作。数组不为空随机取得节点为空，需要做初始化节点的动作。只有在已经存在节点，那么尝试CAS修改值。如果并发严重的话才会考虑进到fullAddCount方法做扩容或者其他操作。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 4]  
大致流程图如上，如果CounterCell为Null，说明还没有线程往Map中put元素，直接CAS baseCount（可能此时的线程竞争不激烈）。如果成功了则表示添加成功，失败的话说明还有其他线程在操作baseCount。准备开始修改CounterCell数组中节点的值，随机生成一个随机数取余CounterCell数组长度，如果节点为Null或者CAS节点值修改失败进入fullAddCount。再来看fullAddCount函数。

从上图可以看出，

*  当直接CAS操作BASECOUNT失败
 *  CounterCell为Null（未初始化）
 *  随机值取余CounterCell表长度下标元素为Null
 *  以及CAS修改下标元素值失败  
    都会进入fullAddCount方法。一起看看fullAddCount做了什么。如果你不想看代码可以直接看下面的图和总结。

private final void fullAddCount(long x, boolean wasUncontended) { 
        int h;
        //如果当前线程的probe的值为0，则初始化当前线程的probe的值,probe就是随机数
        if ((h = ThreadLocalRandom.getProbe()) == 0) { 
            //force initialization
            ThreadLocalRandom.localInit();
            //重新获取probe，随机数
            h = ThreadLocalRandom.getProbe();
            // 重新生成了probe，未冲突标志位设置为true
            wasUncontended = true;
        }
    
        boolean collide = false;
        //自旋
        for (;;) { 
            CounterCell[] as; 
            CounterCell a; 
            int n; 
            long v;
    
            // 如果counterCells已经完成初始化过，可以先看初始化部分
            if ((as = counterCells) != null && (n = as.length) > 0) { 
    
                //随机数快速取余表长度,定位数组中的节点
                if ((a = as[(n - 1) & h]) == null) { 
                    //该位置不存在节点元素
                    //cellsBusy=0表示counterCells不在初始化或者扩容状态下
                    if (cellsBusy == 0) { 
                        //构造一个CounterCell的值
                        CounterCell r = new CounterCell(x); 
    
                        //cellsBusy=0表示counterCells不在初始化或者扩容状态下,并且没有其他线程在添加节点
                        //通过cas设置cellsBusy标识，防止其他线程来对counterCells并发处理
                        if (cellsBusy == 0 &&
                            U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { 
                            boolean created = false;
                            try {                
                                CounterCell[] rs; int m, j;
                                //将初始化的r对象的元素个数放在对应下标的位置
                                //cas成功后，检查一些必要条件是否依旧成立
                                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;
                        }
                    }
                    collide = false;
                } else if (!wasUncontended)
                    wasUncontended = true;
                //由于指定下标位置的cell值不为空，则直接通过cas进行原子累加，如果成功，则直接退出
                else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
                    break;
                // CounterCells大于CPU核心数（线程的并发数不会超过cpu核心数）
                else if (counterCells != as || n >= NCPU)
                    //设置当前线程的本次循环失败不进行其他操作，下次循环再看情况做选择
                    collide = false;  
                //上述步骤依旧未完成，线程的并发数小于cpu核心数，恢复collide状态，标识下次循环会进行操作
                else if (!collide)
                    collide = true;
    
                //进入这个步骤，说明CounterCell数组容量不够，
                //线程竞争比较大了，才会到这步
                //所以先设置一个标识表示为正在扩容
                else if (cellsBusy == 0 &&
                         U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { 
                    try { 
                        if (counterCells == as) { 
                            //扩容一倍 2变成4，这个扩容比较简单
                            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;
                }
                //继续下一次自旋
                h = ThreadLocalRandom.advanceProbe(h);
            }
            //counterCellsw未初始化，进行初始化
            //cellsBusy=0表示没有线程执行初始化操作
            //cellsBusy volatile变量保持线程可见性
            //通过cas更新cellsbusy的值标注当前线程正在做初始化操作
            else if (cellsBusy == 0 && 
                     counterCells == as &&
                     U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { 
                boolean init = false;
                // Initialize table
                try {                            
                    if (counterCells == as) { 
                        //初始化容量为2
                        CounterCell[] rs = new CounterCell[2];
                        //将个数 放在数组随机的位置
                        rs[h & 1] = new CounterCell(x);
                        //赋值给counterCells
                        counterCells = rs;
                        //设置初始化完成标识
                        init = true;
                    }
                } finally { 
                    //初始化结束恢复标识
                    cellsBusy = 0;
                }
                if (init)
                    break;
            }
            //竞争激烈，以上情况都不满足，其它线程占据cell 数组，直接累加在base变量中
            else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
                break;
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 5]  
ConcurrentHashMap单纯做一个size+1的操作都是这般鬼斧神工，太难了。不知道你看完上面的图没有，**做个总结。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 6]

## size+1是如何做到线程安全的 ##

绕来绕去的，心中是不是在想直接CAS操作baseCount累加就不用这么麻烦了，**但是这会造成多个线程在put完元素，同时竞争baseCount，导致性能低下。**  
选择size=baseCount+counterCells数组中的所有值加一块，这样结合上图就会发现，多个线程可以同时竞争baseCount，counterCells的每个数据节点（改变counterCells需要竞争cellsBusy），这样是不是有可能10个线程竞争一个baseCount和多个counterCells数组节点（默认数组长度为2，当竞争激烈时，会有线程扩容成2倍），10个竞争3个，是不是比10个竞争1个好呢。  
下面看下，什么情况下会CAS竞争什么节点

1.  counterCells未初始化，CAS竞争cellsBusy置为1，开始初始化
2.  线程竞争激烈，counterCells扩容成2倍，CAS竞争cellsBusy置为1
3.  counterCells相应下标元素为Null，CAS竞争cellsBusy置为1，需要对counterCells做修改
4.  counterCells相应下标元素不为Null，CAS竞争相应下标元素，+x  
    **瞬间由多个线程竞争一个baseCount，到多个线程竞争多个元素，并且当竞争激烈时，还会进行扩容，可以竞争元素增多。**  
    还有一点值得注意的地方，为什么counterCells相应下标元素不为Null时才可以CAS修改其中的值。如果需要添加节点到数组中需要CAS竞争cellsBusy置为1

你以为**addCount**方法结束了吗，上文说了**还要判断是否需要扩容**，到现在目前为止只是完成了size的+1操作，后面还有一半的代码，来判断是否需要扩容。

private final void addCount(long x, int check) { 
             /******省略size+1部分的代码**************/
            //统计map中有多少个元素
            long    s = sumCount();
            //remove也会调用addCount，但是check为-1代表不需要检查扩容
            if (check >= 0) { 
                Node<K,V>[] tab;
                 Node<K,V>[] nt; 
                int n;
                int sc;
                //1.为什么是循环而不是if
                	//首先内部的几处CAS都有可能会失败，失败了就需要再做一次判断
                	//其次考虑一种情况，是否会存在，该线程刚刚完成扩容，又有大量的元素进到map中
                	//while中完成扩容后，最后一行代码又计算了一下map的元素个数，看是否还需要扩容
                //2. 在初始化table完成后就会修改sizeCtl为扩容的阈值
                while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                       (n = tab.length) < MAXIMUM_CAPACITY) { 
                    int rs = resizeStamp(n);
                    //已经存在线程扩容了（回顾一下sizeCtl各个数字的含义）
                    if (sc < 0) { 
                        // 没看懂，前面几个关于sc的判断没看懂，有明白的辛苦评论告知
                        // 能进入if的说明sc小于0，但是还存在一个情况就是进入if后扩容完成了
                        //在扩容的分配任务中可以看到对nextTable，transferIndex 的修改
                        if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                            sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                            transferIndex <= 0)
                            break;
                        //CAS修改使得sizeCtl表示的正在扩容的线程+1
                        if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                            //CAS正在扩容的线程+1成功后，开始扩容操作
                            transfer(tab, nt);
                    //CAS修改SIZECTL的值，CAS成功则说明当前线程为第一个进入扩容的线程
                    }else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))
                        //进行扩容
                        transfer(tab, null);
                    s = sumCount();
                }
            }
        }

至此要进入最核心的代码transfer扩容了。做好战斗准备

## 核心方法void transfer(Node K,V \[\] tab, Node K,V \[\] nextTab) 扩容，最难的方法 ##

ConcurrentHashMap是支持多个线程扩容的，**在什么时候触发多个线程后续再分析**。  
当有多个线程扩容时，需要先分配任务，将将数组拆分成多块，指定线程处理某一些桶的扩容。例如这样  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 7]  
下面先分析线程是怎么确定自己需要扩容的桶的。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { 
            //确认一个线程需要处理多少个桶
            int n = tab.length;
            //线程单次需要处理的桶的数量
            int stride;
            //如果CPU个数大于1，需要处理的数量： 将 length / 8 然后除以 CPU核心数,如果得到的结果小于 16，那么就使用 16。
            //让每个 CPU 处理的桶一样多，避免出现转移任务不均匀的现象 
            if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
                stride = MIN_TRANSFER_STRIDE;
             
            //nextTab为Null扩容线程，不为Null是帮助扩容线程
            if (nextTab == null) {             // initiating
                try { 
                    //初始化一下新的数组
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
                    nextTab = nt;
                } catch (Throwable ex) {       // try to cope with OOME
                    sizeCtl = Integer.MAX_VALUE;
                    return;
                }
                //赋值，这样其他线程访问nextTable时就不会为空了 
                nextTable = nextTab;
                //这个下标很重要，这里是等于旧的数组的容量。代表的含义是未处理的最后一个位置。
                transferIndex = n;
            }
            //新数组的容量，也很重要，先记着
            int nextn = nextTab.length;
            //上文说过的一个标记的用的Node节点，hash值为-1
            ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
            //也是一个标识，标识是否前进到下一个桶的位置（是否处理下一个桶）
            boolean advance = true;
            //当前线程的扩容任务是否结束
            boolean finishing = false;
            //开始自旋处理
            for (int i = 0, bound = 0;;) { 
                Node<K,V> f; 
                int fh;
                while (advance) { 
                    int nextIndex, nextBound;
                    //这个两个if暂时不要看，看完最后一个分支再看
                    //只需要记得其中完成了--i，nextIndex = transferIndex（旧的table的长度 ）
                    //（看完最后一个分支再看）当经历过下面的任务分配后,
                    // 如果--i，大于等于bound上限，说明已经分配了任务并且没执行完不用再执行分配任务了
                    if (--i >= bound || finishing)
                        advance = false;
                    else if ((nextIndex = transferIndex) <= 0) { 
                        i = -1;
                        advance = false;
                    }
                    //CAS设置transferIndex的值（未分配处理的最后一个结点的位置），如果CAS成功就计算需要处理的桶的开始和结束位置。
                    //其中两个分重要的属性nextBound，nextIndex
                    //处理的桶下限：nextBound = 数组长度-单次需要处理的桶个数
                    //处理的桶上限：i = 数组的最后一个
                    else if (U.compareAndSwapInt
                             (this, TRANSFERINDEX, nextIndex,
                              nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { 
                        bound = nextBound;
                        i = nextIndex - 1;
                        //任务分配完成，结束while循环
                        advance = false;
                    }
                }
           }
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 8]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 9]  
**分配任务的重点在于，先CAS修改transferIndex的值，修改成功再分配。这样在其他线程读取到transferIndex的值时，再去分配任务时就不会造成任务重复分配**分配完任务后，看线程是怎么处理这些分配到自己头上的这些桶的。先看一个Node结构，其HASH值为-1，用来做特定标识的。标记当前桶已经被处理了

static final class ForwardingNode<K,V> extends Node<K,V> { 
            final Node<K,V>[] nextTable;
            ForwardingNode(Node<K,V>[] tab) { 
                //Hash值MOVED为负一
                super(MOVED, null, null, null);
                this.nextTable = tab;
            }
    }

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

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { 
            /*************************以下代码为分配任务*************************/
            int n = tab.length, stride;
            if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
                stride = MIN_TRANSFER_STRIDE; // subdivide range
            if (nextTab == null) {             // initiating
                try { 
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
                    nextTab = nt;
                } catch (Throwable ex) {       // try to cope with OOME
                    sizeCtl = Integer.MAX_VALUE;
                    return;
                }
                nextTable = nextTab;
                transferIndex = n;
            }
            int nextn = nextTab.length;
            ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
            boolean advance = true;
            boolean finishing = false; // to ensure sweep before committing nextTab
            /*************************以上代码为分配任务*************************/
            for (int i = 0, bound = 0;;) { 
                //先不用去看这个if，这个是判断当前线程是否已经将任务处理完毕
                if (i < 0 || i >= n || i + n >= nextn) { 
                    int sc;
                    if (finishing) { 
                        nextTable = null;
                        table = nextTab;
                        sizeCtl = (n << 1) - (n >>> 1);
                        return;
                    }
                    if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { 
                        if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                            return;
                        finishing = advance = true;
                        i = n; // recheck before commit
                    }
                }
                //如果这个桶是空的，直接CAS一个Hash值为-1的ForwardingNode节点
                //代表这个桶已经处理完毕，然后进入下一次循环，i在while循环中会-1
                else if ((f = tabAt(tab, i)) == null)
                    advance = casTabAt(tab, i, null, fwd);
                //桶不为空，但是是ForwardingNode节点，直接进入下次循环（处理下个桶）
                else if ((fh = f.hash) == MOVED)
                    advance = true; // already processed
                else { 
                    //开始锁住头节点，往新的数组迁移数组（里面细节先不看之后都是同步代码块。先分析一下整体流程）
                    synchronized (f) { 
                       //判断是链表还是树，按结构迁移数据
                    }
                }
            }
        }

至此为止扩容部分大部分结束。其实简单来说，就两个步骤  
![在这里插入图片描述][20201121150304841.png_pic_center]  
但是每个步骤拆开细节都很多。

## 什么时候才会触发分配任务 ##

## 怎么判断所有任务已经分配完毕。 ##

//分配任务的代码
     while (advance) { 
        int nextIndex, nextBound;
        if (--i >= bound || finishing)
            advance = false;
        else if ((nextIndex = transferIndex) <= 0) { 
            i = -1;
            advance = false;
        }
        else if (U.compareAndSwapInt
                 (this, TRANSFERINDEX, nextIndex,
                  nextBound = (nextIndex > stride ?
                               nextIndex - stride : 0))) { 
            bound = nextBound;
            i = nextIndex - 1;
            advance = false;
        }
    }

> 1.处理位置i要大于下限bound=》分配给当前线程的任务没处理完。  
> 2. transferIndex（代表还未分配的桶的最后一个的位置）大于0=》还有未处理的桶  
> 什么时候transferIndex会被设置成等于0呢，执行到CAS处，当transferIndex不大于stride（每次被分配的个数），如果CAS成功说明我将剩下所有的任务都分配给A线程去执行了，那么这个时候B线程进来发现transferIndex等于0，**那就是任务分配完了**。设置i = -1;advance = false;**（i=-1这个很重要）**  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 10]

## 再看什么时候会退出自旋 ##

整个transfer方法看下来，只有这个if语句内有return语句。（忽略异常return）。上面一个点说到了当判断transferIndex等于0时说明任务分配完了，i = -1。还有就是while内 if (–i >= bound || finishing)，–i递减到-1。才会满足以下条件。说明已经将所有任务分配完，并且当前线程的任务已经执行完。

//后面两个判断条件没看懂，有看懂的辛苦评论区告知
    if (i < 0 || i >= n || i + n >= nextn) { 
        int sc;
        //什么时候finishing会为true呢，只有执行了下面的CAS才会为true
        //所以先看看下面的额CAS干什么了
        //经过下面的CAS后，finishing设置为True，并且先设置i=n，进行下次循环
        //这个时候只有在--i到-1时，再进到这个if语句，也就是将所有元素再检查一遍后
        //才会再进到这个方法，这时finishing为true退出自旋，完成扩容。
        if (finishing) { 
            nextTable = null;
            table = nextTab;
            sizeCtl = (n << 1) - (n >>> 1);
            return;
        }
        //CAS设置正在扩容的线程-1
        if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { 
            //看不懂为什么这么操作，有知道的辛苦评论区告知
            if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                return;
            //设置扩容已经结束
            finishing = advance = true;
            //把i重新设置为N,再重新检查一遍，结束本次循环
            i = n; // recheck before commit
        }
    }

i = n，这个设定很巧妙。当所有任务都分配出去后，线程便会进到这个方法中，但是想退出自旋，完成扩容，还需要做最后一件事，检查一遍所有的桶是否都已经完成迁移。以及可能帮助承接最后一个任务的线程完成剩余作业。当检查完毕，–i到负数时，再次进入这个判断条件退出扩容。

## 帮助扩容的时机 ##

在Put元素时如果节点的Hash为-1说明，正在扩容，而且该桶的元素已经发生了迁移，帮助扩容。helpTransfer跟addCount中判断帮助扩容差不多就不再进入helpTransfer函数了。

f = tabAt(tab, i = (n - 1) & hash)
    else if ((fh = f.hash) == MOVED) { 
        tab = helpTransfer(tab, f);
    }

在addCount后，需要扩容，并且已经有其他线程在扩容了（sizeCtl的值已经为负数了 ）

if (sc < 0) { 
        if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
            transfer(tab, nextTable);
    }

## 最后我们来大概的看看同步代码块中的代码 ##

不是很重要，毕竟在synchronized中，是同步代码。其操作与HashMap中迁移数据类似  
[HashMap源码阅读][HashMap]

synchronized (f) { 
        if (tabAt(tab, i) == f) { 
            Node<K,V> ln, hn;
            if (fh >= 0) { 
                //节点的hash值&table容量，要么等于0，要么不为0
                int runBit = fh & n;
                //记录节点，这个节点的意思代表后面的所有hash&N都跟lastRun一样了
                Node<K,V> lastRun = f;
                //遍历链表
                for (Node<K,V> p = f.next; p != null; p = p.next) { 
                    int b = p.hash & n;
                    //只要后一个节点的p.hash & n不等于前一个节点，就更新
                    //这样在后面lastRun后面的节点都跟lastRun一致
                    if (b != runBit) { 
                        runBit = b;
                        lastRun = p;
                    }
                }
                //如果runBit==0保留在原处的节点
                //这点在HashMap中有过说明
                if (runBit == 0) { 
                    ln = lastRun;
                    hn = null;
                }else { 
                    //位置变为当前位置+table长度
                    hn = lastRun;
                    ln = null;
                }
                //将节点分开，但是终止条件是p != lastRun;因为lastRun节点后面的存放的位置与lastRun一致，在前面的代码已经处理过
                for (Node<K,V> p = f; p != lastRun; p = p.next) { 
                    int ph = p.hash; 
                    K pk = p.key;
                    V pv = p.val;
                    //为什么ph & n==0就放在原处，否则i+n在hashmap一文中说过了，不再赘述
                    if ((ph & n) == 0)
                        ln = new Node<K,V>(ph, pk, pv, ln);
                    else
                        hn = new Node<K,V>(ph, pk, pv, hn);
                }
                //具有volatile语义的设置节点到新的数组
                setTabAt(nextTab, i, ln);
                setTabAt(nextTab, i + n, hn);
                // 将旧的链表设置成占位符
                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) { 
                        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);
                setTabAt(tab, i, fwd);
                advance = true;
            }
        }
    }

## 最后的最后做个总结 ##

ConcurrentHashMap使用了volatile关键字保证线程之间可见性，以及CAS，synchronized来做到支持并发的。  
其中Put方法时，先去判断table是否已经初始化，如果还未初始化线程CAS修改SIZECTL的值为-1进行初始化，如果此时其他线程读到SIZECTL为负数会让出Cpu的执行。  
存放元素时如果table中相应位置为Null则会CAS修改节点，如果该节点的Hash值为-1说明正在扩容，并且这个桶已经扩容完成了就会帮助扩容。  
如果CAS失败就会进入下一次的循环，这时如果该位置存在节点就用synchronized锁住这个节点再判断是红黑树结构还是链表结构插入元素。  
在插入元素后要进行size+1操作，以及判断是否需要扩容。  
在做size+1的操作时，先CAS修改BaseCount属性，如果修改失败就会转而去操作CounterCell数组，这样就把所有线程竞争baseCount变成了竞争baseCount和CounterCell数组中的每个节点。  
当然如果线程之间竞争激烈会有线程CAS修改cellsBusy的值为1，去扩容CounterCell。  
当执行完size+1操作后，计算map的size，将baseCount+CounterCell数组中每个节点的值，如果这个值大于map容量的0.75则需要扩容。  
扩容支持多线程扩容，扩容第一步会为每个线程分配任务，需要处理桶的长度，线程会判断分配的任务范围下每个位置的节点的情况，如果是null CAS为ForwardingNode一个Hash值为-1的标识节点。  
如果不为空也不为ForwardingNode节点，就用synchronized锁住头节点类似HashMap一样将元素移动到新的table。  
当transferIndex=0时，说明任务分配完毕，会CAS设置SIZECTL-1，代表正在扩容的线程减少一个，CAS成功后会再检查一遍所有得元素是否都已经转移成功再退出扩容

[202011141615544.png_pic_center]: /images/20221120/a10a45fb916242bca8ce8ad009abefd2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/48c8001bed8e4b5883a916c187967ebd.png
[volatile]: https://blog.csdn.net/wsdfym/article/details/102878329
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/ca23ae777d4e40c3a4016cf262200859.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/2556b6af89d9483ebad4fae77292645d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/1c303bbd5a6744fa92e44018beafc3cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221120/d5dd7633f48a452593292e380cb5c90e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221120/3d7d696b605c4ffb918abea71b227824.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221120/36c23bf12a394f5ba2a883d31db3a572.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221120/10fdc1f69b2541fdb78429129f9a8af5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221120/e0141cbef6e64b3dbcc4c8daad284419.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221120/291b8731cf024a3a90d54796d6ae2697.png
[2020112113581883.png_pic_center]: /images/20221120/8b2aa1718c1047ea9dfac5b7bfbcb977.png
[20201121150304841.png_pic_center]: /images/20221120/822fda2d593e4eeba253cfc1a77dffc4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221120/9fc4db999bc24da39108958738a18202.png
[HashMap]: https://blog.csdn.net/wsdfym/article/details/109232470