刨析JDK1.8的ConcurrentHashMap

本篇文章将从concurrentHashMap的源码出发，了解它的实现原理。

文章结构

部分关键参数
构造方法
put
扩容讲解
get
总结
面试题

首先简单说说他的特性。

concurrentHashMap跟Hashtable具有相同的功能方法。可以看作是Hashtable的升级版，HashMap的线程安全版。
跟Hashtable相同，它的键或值不允许是null。
ConcurrentHashMap和HashMap一样都是采用拉链法处理哈希冲突，且都为了防止单链表过长影响查询效率，所以当链表长度超过某一个值时候将用红黑树代替链表进行存储，采用了数组+链表+红黑树的结构 $[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7m80bvPS-1644594019825)(C:\Users\14470\Desktop\java笔记\java JUC\ConcurrentHashMap刨析.assets\ydlvlwmr5z.jpeg)]$
ConcurrentHashMap与Hashtable比较

线程安全的实现：Hashtable采用对象锁(synchronized修饰对象方法)来保证线程安全，也就是一个Hashtable对象只有一把锁，如果线程1拿了对象A的锁进行有synchronized修饰的put方法，其他线程是无法操作对象A中有synchronized修饰的方法的(如get方法、remove方法等)，竞争激烈所以效率低下。而ConcurrentHashMap采用CAS + synchronized来保证并发安全性，且synchronized关键字不是用在方法上而是用在了具体的对象上，实现了更小粒度的锁，等会源码分析的时候在细说这个SUN大师们的鬼斧神工
数据结构的实现：Hashtable采用的是数组 + 链表，当链表过长会影响查询效率，而ConcurrentHashMap采用数组 + 链表 + 红黑树，当链表长度超过某一个值，则将链表转成红黑树，提高查询效率。

ConcurrentHashMap

我们先看一下ConcurrentHashMap实现了哪些接口、继承了哪些类，对ConcurrentHashMap有一个整体认知。

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-f6xXVtXD-1644594079042)(C:\Users\14470\Desktop\java笔记\java JUC\ConcurrentHashMap刨析.assets\df2yzdyumf.jpeg)]$

ConcurrentHashMap跟HashMap一样继承AbstractMap接口，

然后实现了ConcurrentMap接口，这个和HashMap不一样，HashMap是直接实现的Map接口。

部分关键参数

//node数组最大容量：2^30=1073741824  
private static final int MAXIMUM_CAPACITY = 1 << 30;

//默认初始表容量。必须是 2 的幂（即至少为 1）且最多为 MAXIMUM_CAPACITY
private static final int DEFAULT_CAPACITY = 16;

//最大可能（非二的幂）数组大小。 toArray 和相关方法需要
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

//负载因子
private static final float LOAD_FACTOR = 0.75f;

 // 链表转红黑树阀值,> 8 链表转换为红黑树  
static final int TREEIFY_THRESHOLD = 8;

//树转链表阀值，小于等于6（tranfer时，lc、hc=0两个计数器分别++记录原bin、新binTreeNode数量，<=UNTREEIFY_THRESHOLD 则untreeify(lo)）
static final int UNTREEIFY_THRESHOLD = 6;

  //存放node的数组  
transient  volatile  Node<K,V>[] table;  

  /*控制标识符，用来控制table的初始化和扩容的操作，不同的值有不同的含义  
  *当为负数时：-1 代表正在初始化，-N 代表有N-1 个线程正在 进行扩容  
  *当为  0  时：代表当时的table还没有被初始化  
  *当为正数时：表示初始化或者下一次进行扩容的大小  
*/
private  transient  volatile  int  sizeCtl;

几个重要的成员变量table、nextTable、baseCount、sizeCtl、transferIndex、cellsBusy

table：数据类型是Node数组，这里的Node和HashMap的Node一样都是内部类且实现了Map.Entry接口
nextTable：哈希表扩容时生成的数据，数组为扩容前的2倍
sizeCtl：多个线程的共享变量，是操作的控制标识符，它的作用不仅包括threshold的作用，在不同的地方有不同的值也有不同的用途。
- -1代表正在初始化
- -N代表有N-1个线程正在进行扩容操作
- 0代表hash表还没有被初始化
- 正数表示下一次进行扩容的容量大小
ForwardingNode：一个特殊的Node节点，Hash地址为-1，存储着nextTable的引用，只有table发生扩用的时候，ForwardingNode才会发挥作用，作为一个占位符放在table中表示当前节点为null或者已被移动

ConcurrentHashMap 针对 ForwardingNode、ReservationNode，以及树根结点都定义了特定的哈希值：

下面判断用到的很多哦。

/*
 * 节点哈希字段的编码. See above for explanation.
 */
/** ForwardingNode 结点的 hash 值 */
static final int MOVED     = -1; // hash for forwarding nodes  转发节点的哈希

static final int TREEBIN   = -2; // hash for roots of trees    树根的哈希

static final int RESERVED  = -3; // hash for transient reservations  临时预订的哈希
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash

构造方法

它的构造方法一共有5个，从数量上看就和HashMap、Hashtable(4个)的不同，多出的那个构造函数是public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)，即除了传入容量大小、负载因子之外还多传入了一个整型的concurrencyLevel，这个整型是我们预先估计的并发量，比如我们估计并发是30，那么就可以传入30。

其他的4个构造函数的参数和HashMap的一样，而具体的初始化过程却又不相同，HashMap和Hashtable传入的容量大小和负载因子都是为了计算出初始阈值(threshold)，而ConcurrentHashMap传入的容量大小和负载因子是为了计算出sizeCtl用于初始化table，这个sizeCtl即table数组的大小，不同的构造函数计算sizeCtl方法都不一样。

//1.  无参构造函数，什么也不做，此时sizeCtl参数为
//table的初始化放在了第一次插入数据时，默认容量大小是16。
//使用默认初始表大小 (16) 创建一个新的空映射
public ConcurrentHashMap() {
}
//2.  传入容量大小的构造方法
public ConcurrentHashMap(int initialCapacity) {
        if (initialCapacity < 0)
            throw new IllegalArgumentException();
    //如果传入的容量大小大于允许的最大容量值 则cap取允许的容量最大值 否则cap =
    //((传入的容量大小 + 传入的容量大小无符号右移1位 + 1)的结果向上取最近的2幂次方)，
    //即如果传入的容量大小是12 则 cap = 32 
    //(12 >>> 1)=1100>>>1=0110=7
    //=(12 + (12 >>> 1) + 1=19 并向上取2的幂次方即32)
    //，这里为啥一定要是2的幂次方，原因和HashMap的threshold一样，
    //都是为了让位运算和取模运算的结果一样。
    //MAXIMUM_CAPACITY即允许的最大容量值 为2^30。
        int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                   MAXIMUM_CAPACITY :
                   tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    //将上面计算出的cap 赋值给sizeCtl，注意此时sizeCtl为正数，代表进行扩容的容量大小。
        this.sizeCtl = cap;
}

//3.创建一个与给定Map具有相同映射Map

public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
    //置sizeCtl为默认容量大小 即16。
    this.sizeCtl = DEFAULT_CAPACITY;
    putAll(m);
}

//4.传入容量大小和负载因子的构造方法
//默认并发数大小是1。
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
    this(initialCapacity, loadFactor, 1);
}

//5. 传入容量大小、负载因子和并发数大小的构造方法	
public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
    //如果负载因子>0，传入容量大小<0,并发数大小<=0一项不符合就
    //直接抛出IllegalArgumentException非法参数异常
    if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
    
    //如果传入的容量大小小于并发数大小，咱们就把初始容量变成更大的并发数。这样做的原因是确保每一个Node只会分配给一个线程
    if (initialCapacity < concurrencyLevel)   // 
            initialCapacity = concurrencyLevel;   // 
/**
下面就时计算sizeCtl的值了，用于下一次扩容的时候就是初始化的时候。
*/
    long size = (long)(1.0 + (long)initialCapacity / loadFactor);
    
    //如果size比允许的最大容量值还大，那直接sizeCtl= 允许的最大容量值。
    // 否则对size进行tableSizeFor操作就是 size向上取2的幂次方
    //比如 size = 6   tableSizeFor((int)size)=9
    //    size = 9   tableSizeFor((int)size)=16
    //    size = 12   tableSizeFor((int)size)=32
    int cap = (size >= (long)MAXIMUM_CAPACITY) ?
            MAXIMUM_CAPACITY : tableSizeFor((int)size)=9;

    this.sizeCtl = cap;  //计算好之后便赋值给sizeCtl
}

构造方法简单总结：

都是为了算出sizeCtl值，也就是初始化table数组大小。

一共有5个构造方法。

无参构造函数，什么也不做，此时sizeCtl参数为0，初始化操作在put操作中实现
传入容量大小的构造方法，对参数校验后(>0)，传入的容量大小大于MAXIMUM_CAPACITY即允许的最大容量 2^30，则cap取允许的容量最大值。比他小咱们就做tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1))操作，对initialCapacity + (initialCapacity >>> 1) + 1 向上取最近的2的幂次方，最后将值给sizeCtl。
创建一个与给定Map具有相同映射Map。
传入容量大小和负载因子的构造方法，它调用第5个方法，并发数大小=1。
传入容量大小、负载因子和预期并发数大小的构造方法。具体看上面。

put操作

put

public V put(K key, V value) {
    return putVal(key, value, false);//调用了putVal方法
}

它就调用了putVal方法，咱们来看putVal方法

putVal

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();//如果有一个为null，那么直接报错。所以可以得出concurrentHashMap中Key，value不能为空。
    int hash = spread(key.hashCode());//计算key的hash值，参与插入位置的计算
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {//循环尝试,for循环内它是if..。else if...else格式
        Node<K,V> f; int n, i, fh;
       
        //如果table数组为null，说明还没有table数组呢，那咱们就开始初始化数组。
        if (tab == null || (n = tab.length) == 0)
       //初始化之后，会直接进行下次循环。
            tab = initTable();
        
        //如果i位置没有数据(即该下标上的链表为空)，就直接CAS无锁插入
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//获取对应位置Node节点(链表头节点)(Volatile的)
            // casTabAt方法CAS无锁的添加到空箱,  如果插入成功了则跳出for循环，插入
            //失败(其他线程抢先插入了)，那么继续下次循环。
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;               
        }
        
        //如果该下标上的节点的哈希地址为-1(即链表的头节点为ForwardingNode节点)，则表示
        //table正在进行扩容（transfer），那咱们先等着吧。
        //注意的：：ConcurrentHashMap初始化和扩容不是用同一个方法，而
        //HashMap和Hashtable都是用同一个方法，并且当前线程会去协助扩容，扩容过程后面介绍。
        else if ((fh = f.hash) == MOVED)//-1
            tab = helpTransfer(tab, f);//返回扩容完成后的table。
        
        //上面情况都没有，那么我们就将进入到链表中，将新节点插入或者覆盖旧值。
        else {
            V oldVal = null;
            
         // only针对首个节点进行加锁操作，一次锁的资源小了，减少了锁的粒度。也正是因为这个提高了ConcurrentHashMap的效率，提高了并发度。
            synchronized (f) {
                if (tabAt(tab, i) == f) {  //做一次校验看看头节点一样不
                    if (fh >= 0) {//哦~不知道fh怎么来的？看26行。节点的哈希地址大于等于0，则表示这是个链表
                        binCount = 1;
                         //遍历链表，大家应该非常熟悉了吧。
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {//如果发现有key一样的，那咱们就把value值替换掉。
                                oldVal = e.val;
                                if (!onlyIfAbsent)
                                    e.val = value;
                                break;//结束推出链表循环
                            }
                            
                            // 如果没有找到值为key的节点，直接新建Node并加入链表尾部即可。
                            Node<K,V> pred = e;
                            if ((e = e.next) == null) {//到链表尾部了。  
                                pred.next = new Node<K,V>(hash, key,
                                                          value, null);/将节点插入链表尾部  
                                break;//结束推出链表循环
                            } 
                        }
                    }
                    else if (f instanceof TreeBin) {//如果首节点为TreeBin类型，说明为红黑树结构，执行putTreeVal操作
                        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;
                        }
                    }
                }
            }
            //这时候同步代码块已经结束，元素插入完毕，释放了锁。
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)//如果链表的长度大于树形阈值 8 时就会进行红黑树的转换
                    treeifyBin(tab, i); 
                if (oldVal != null)
                    return oldVal;//如果key对应的以前有值，那么返回oldVal
                break;
            }
        }
    }
    addCount(1L, binCount);//统计size，可能会扩容transfer操作。
    return null;
}

好！咱们先来总结一下put流程：

校验Key，value是否为空。如果有一个为null，那么直接报NullPointerException异常。所以可以得出concurrentHashMap中Key，value不能为空。
循环尝试插入。进入循环。
case1:如果没有初始化就先调用initTable（）方法来进行初始化过程
case2:根据Hash值计算插入位置(n – 1) & hash=i。如果没有hash冲突,也就是说插入位置上面没有数据，就直接casTabAt（）方法将数据插入。
case3:插入位置上有数据。数据的头节点的哈希地址为-1(即链表的头节点为ForwardingNode节点)，则表示其他线程正在对table进行扩容（transfer），就先等着,等其他线程扩容完了咱们再尝试插入。
case4:上面情况都没有。就对首节点加synchronized锁来保证线程安全，两种情况，一种是链表形式就直接遍历到尾端插入，一种是红黑树就按照红黑树结构插入，结束加锁。
如果Hash冲突时会形成Node链表，在链表长度超过8，Node数组超过64 时会将链表结构转换为红黑树的结构。
break退出循环。
调用addCount（）方法统计Map已存储的键值对数量size++，检查是否需要扩容，需要扩容就扩容。

了解了它的基本流程之后，我们来看其中的一些调用方法。

initTable     初始化数组
tabAt		  获取对应位置Node节点
casTabAt	  CAS添加节点
     
helpTransfer  	帮助扩容，如果线程进入到这边说明已经有其他线程正在做扩容操作，这个是一个辅助方法
transfer   		扩容操作（重点）
addCount  		ConcurrentHashMap的键值对数量size+1，并判断是否需要扩容

咱们一个个说。

initTable

第一次进入putVal循环发现table数组是null的时候调用。

initTable()方法初始化一个合适大小的table数组，然后设置sizeCtl值(下一次扩容阈值=数组长度*0.75)。

我们知道ConcurrentHashMap是线程安全的，即支持多线程的，那么一开始很多个线程同时执行put()方法，而table又没初始化，那么就会很多个线程会去执行initTable()方法尝试初始化table，而put方法和initTable方法都是没有加锁的(synchronize)，

那SUN的大师们是怎么保证线程安全的呢？

/**
 * 初始化表table数组 大小为sizeCtl值。
 */
private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    //先看看table是否被初始化，如果已经初始化了，结束while循环
    //如果table为null那么一直while循环尝试初始化，直到完成
    while ((tab = table) == null || tab.length == 0) {\
        
        if ((sc = sizeCtl) < 0) // sizeCtl< 0说明，其他线程抢先的正在初始化或者扩容，
            Thread.yield(); 	// 初始化竞争失败；让出cpu，等着就完了。运行态转成就绪态
        //sizeCtl >= 0
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { //Unsafed.compareAndSwapInt以CAS操作CASsizeCtl=-1,表示初始化状态。
            try {
                if ((tab = table) == null || tab.length == 0) {//再次确认当前table为null即还未初始化，这个判断不能少。
                    //如果sc(sizeCtl)大于0，则n=sc，否则n=默认的容量大
                    小16，
                    //这里的sc=sizeCtl=0，即如果在构造函数没有指定容量
                    大小，
                    //否则使用了有参数的构造函数，sc=sizeCtl=指定的容量大小。
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; //new Node数组(table)
                    table = tab = nt;
                    
                    sc = n - (n >>> 2); //计算阈值，n - (n >>> 2) = 0.75*n，大于这个阈值，就会发生扩容。
                 
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

initTable()初始化方法总结

initTable()用于里面table数组的初始化，值得一提的是table初始化是没有加锁的，那么如何处理并发呢？
由上面代码可以看到，当要初始化时会通过CAS操作将sizeCtl置为-1，而sizeCtl由volatile修饰，保证修改对后面线程可见。
这之后如果再有线程执行到此方法时检测到sizeCtl为负数，说明已经有线程在给扩容了，这个线程就会调用Thread.yield()让出一次CPU执行时间。

initTable流程总结：

如果table==null，进入循环。
case1: sizeCtl< 0 说明其他线程抢先对table初始化或者扩容，就调用Thread.yield(); 让出一次cpu，等下次抢到cpu再循环判断。
case2: 以CAS操作CASsizeCtl=-1,表示当前线程正在初始化。下面就开始初始化。
判断sizeCtl的值。 sc(sizeCtl)大于0，则容量大小=sc，
sc(sizeCtl)<=0，即如果在使用了有参数的构造函数，sc=sizeCtl=指定的容量大小,否则n=默认的容量大小16。
用上面求出的容量大小new出table数组。
计算阈值，sizeCtl = n – (n >>> 2) = 0.75*n。

tabAt AND casTabAt

都是调用了Unsafe类中的方法。大家感兴趣可以看我的这篇文章
Unsafe详解

@SuppressWarnings("unchecked")
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);
}

helpTransfer 和 transfer

/**
 * Helps transfer if a resize is in progress.
 */
//协助扩容方法
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
     //如果当前table不为null 且 f为ForwardingNode节点 且新的table即nextTable存在的情况下才能协助扩容，该方法的作用是让线程参与扩容的复制。
    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) {	//如果小于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)) { //更新sizeCtl的值，+1，代表新增一个线程参与扩容
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}

transfer()方法为ConcurrentHashMap扩容操作的核心方法。JDK1.8中，ConcurrentHashMap最复杂的部分就是扩容/数据迁移，涉及多线程的合作和rehash。我们先来考虑下一般情况下，如何对一个Hash表进行扩容。

Hash表的扩容，一般都包含两个步骤：

①table数组的扩容
table数组的扩容，一般就是新建一个2倍大小的槽数组，这个过程通过由一个单线程完成，且不允许出现并发。

②数据迁移
所谓数据迁移，就是把旧table中的各个槽中的结点重新分配到新table中*。

这一过程通常涉及到槽中key的rehash(重新Hash)，因为key映射到桶的位置与table的大小有关，新table的大小变了，key映射的位置一般也会变化。

ConcurrentHashMap在处理rehash的时候，并不会重新计算每个key的hash值，而是利用了一种很巧妙的方法。我们在上一篇说过，ConcurrentHashMap内部的table数组的大小必须为2的幂次，原因是让key均匀分布，减少冲突，这只是其中一个原因。另一个原因就是：

当table数组的大小为2的幂次时，通过key.hash & table.length-1这种方式计算出的索引i，当table扩容后（2倍），新的索引要么在原来的位置i，要么是i+n。

我们来看个例子：
$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DXdfscVZ-1644669997297)(C:\Users\14470\Desktop\java笔记\java JUC\ConcurrentHashMap刨析.assets\bVbf0jg.png)]$

上图中：
扩容前，table数组大小为16，key1和key2映射到同一个索引5；
扩容后，table数组的大小变成 2*16=32 ，key1的索引不变，key2的索引变成 5+16=21。

而且还有一个特点，扩容后key对应的索引如果发生了变化，那么其变化后的索引最高位一定是1（见扩容后key2的最高位）。

这种处理方式非常利于扩容时多个线程同时进行的数据迁移操作，因为旧table的各个桶中的结点迁移不会互相影响，所以就可以用**“分治”**的方式，将整个table数组划分为很多部分，每一部分包含一定区间的桶，每个数据迁移线程处理各自区间中的结点，对多线程同时进行数据迁移非常有利，后面我们会详细介绍。

什么时候会扩容？

当往hashMap中成功插入一个key/value节点时，有两种情况可能触发扩容动作：
1、如果新增节点之后，所在链表的元素个数达到了阈值 8，则会调用treeifyBin方法把链表转换成红黑树，不过在结构转换之前，会对数组长度进行判断，实现如下：如果数组长度n小于阈值MIN_TREEIFY_CAPACITY，默认是64，则会调用tryPresize方法把数组长度扩大到原来的两倍，并触发transfer方法，重新调整节点的位置。

2、调用put方法新增节点时，在结尾会调用addCount方法记录元素个数，并检查是否需要进行扩容，当数组元素个数达到阈值时，会触发transfer方法，重新调整节点的位置。

扩容状态下其他线程对集合进行插入、修改、删除、合并、compute 等操作时遇到 ForwardingNode 节点会触发扩容。

putAll 批量插入或者插入节点后发现存在链表长度达到 8 个或以上，但数组长度为 64 以下时会触发扩容。

注意：桶上链表长度达到 8 个或者以上，并且数组长度为 64 以下时只会触发扩容而不会将链表转为红黑树。

扩容原理

/**
 * 数据转移和扩容.
 * 每个调用tranfer的线程会对当前旧table中[transferIndex-stride, transferIndex-1]位置的结点进行迁移
 *
 * @param tab     旧table数组
 * @param nextTab 新table数组
 */
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    //根据服务器CPU数量来决定每个线程负责的bucket桶数量，避免因为扩容的线程过多反而影响性能。
    //如果CPU数量为1，则stride=1，否则将需要迁移的bucket数量(table大小)除以CPU数量，平分给
    //各个线程，但是如果每个线程负责的bucket数量小于限制的最小是(16)的话，则强制给每个线程
    //分配16个bucket数。
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range
    
    //如果nextTable还未初始化，则初始化nextTable，这个初始化和iniTable初始化一样，只能由
    //一个线程完成。
    if (nextTab == null) {   // nextTab为null，那么就初始化它
        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; //[transferIndex-stride, transferIndex-1]表示当前线程要进行数据迁移的桶区间
    }
    int nextn = nextTab.length;
    
     // ForwardingNode结点，当旧table的某个桶中的所有结点都迁移完后，用该结点占据这个桶
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // 标识一个桶的迁移工作是否完成，advance == true 表示可以进行下一个位置的迁移
    boolean advance = true;
    
     //循环的关键变量，最后一个数据迁移的线程将该值置为true，并进行本轮扩容的收尾工作
    boolean finishing = false; 
    
   //下个循环是分配任务和控制当前线程的任务进度，这部分是transfer()的核心逻辑，描述了如何与其他线程协同工作。
    for (int i = 0, bound = 0;;) { //进入循环
        Node<K,V> f; int fh;
        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;
            }
        }
        //i<0说明已经遍历完旧的数组tab；i>=n什么时候有可能呢？在下面看到i=n,所以目前i最大应该是n吧。
        //i+n>=nextn,nextn=nextTab.length，所以如果满足i+n>=nextn说明已经扩容完成
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            if (finishing) {
                nextTable = null;
                table = nextTab; 
                sizeCtl = (n << 1) - (n >>> 1);
                return;
            }
            //利用CAS方法更新这个扩容阈值，在这里面sizectl值减一，说明新加入一个线程参与到扩容操作,参考sizeCtl的注释
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                    return;
                finishing = advance = true;//finishing和advance保证线程已经扩容完成了可以退出循环
                i = n; //先退出，重新检查遍
            }
        }
        else if ((f = tabAt(tab, i)) == null)//如果tab[i]为null，那么就把fwd插入到tab[i]，表明这个节点已经处理过了
            advance = casTabAt(tab, i, null, fwd);
        else if ((fh = f.hash) == MOVED)//那么如果f.hash=-1的话说明该节点为ForwardingNode，说明该节点已经处理过了
            advance = true; // already processed

        //迁移过程（对当前指向的bucket），这部分的逻辑与HashMap类似，拿旧数组的容量当做一
        //个掩码，然后与节点的hash进行与&操作，可以得出该节点的新增有效位，如果新增有效位为
        //0就放入一个链表A，如果为1就放入另一个链表B，链表A在新数组中的位置不变（跟在旧数
        //组的索引一致），链表B在新数组中的位置为原索引加上旧数组容量。
        else {
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;
                    if (fh >= 0) {// 桶的hash>=0，说明是链表迁移
                        /**
                         * 下面的过程会将旧桶中的链表分成两部分：ln链和hn链
                         * ln链会插入到新table的槽i中，hn链会插入到新table的槽i+n中
                         */
                        int runBit = fh & n;
                        Node<K,V> lastRun = f;
                        for (Node<K,V> p = f.next; p != null; p = p.next) {
                            int b = p.hash & n;
                            if (b != runBit) {
                                runBit = b;
                                lastRun = p;
                            }
                        }
                        if (runBit == 0) {
                            ln = lastRun;
                            hn = null;
                        }
                        else {
                            hn = lastRun;
                            ln = null;
                        }
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
                            int ph = p.hash; K pk = p.key; V pv = p.val;
                            if ((ph & n) == 0)
                                ln = new Node<K,V>(ph, pk, pv, ln);
                            else
                                hn = new Node<K,V>(ph, pk, pv, hn);
                        }
                        //
                        setTabAt(nextTab, i, ln);
                        setTabAt(nextTab, i + n, hn);
                        setTabAt(tab, i, fwd);//把已经替换的节点的旧tab的i的位置用fwd结点替换，fwd包含nextTab
                        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;
                            }
                        }
                        //判断扩容后是否还需要红黑树结构
                        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);  // 设置ForwardingNode占位
                        advance = true;         // 表示当前旧桶的结点已迁移完毕
                    }
                }
            }
        }
    }
}

tranfer方法的开头，会计算出一个stride变量的值，这个stride其实就是每个线程处理的桶区间，也就是步长：

// stride可理解成“步长”，即数据迁移时，每个线程要负责旧table中的多少个桶
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
    stride = MIN_TRANSFER_STRIDE;

首次扩容时，会将table数组变成原来的2倍。
注意上面的transferIndex变量，这是一个字段，table[transferIndex-stride, transferIndex-1]就是当前线程要进行数据迁移的桶区间：

/**
 * 扩容时需要用到的一个下标变量.
 */
private transient volatile int transferIndex;

整个transfer方法几乎都在一个自旋操作中完成，从右往左开始进行数据迁移，transfer的退出点是当某个线程处理完最后的table区段——table[0,stride-1]。

transfer方法主要包含4个分支，即对4种不同情况进行处理。
扩容总结

首先根据运算得到需要遍历的次数i，然后利用tabAt方法获得i位置的元素：

如果这个位置为空，就在原table中的i位置放入forwardNode节点，这个也是触发并发扩容的关键点；
如果这个位置是Node节点（fh>=0），如果它是一个链表的头节点，就构造一个反序链表，把他们分别放在nextTable的i和i+n的位置上
如果这个位置是TreeBin节点（fh<0），也做一个反序处理，并且判断是否需要untreefi，把处理的结果分别放在nextTable的i和i+n的位置上
遍历过所有的节点以后就完成了复制工作，这时让nextTable作为新的table，并且更新sizeCtl为新容量的0.75倍，完成扩容。

多线程遍历节点，处理了一个节点，就把对应点的值set为forward，另一个线程看到forward，就向后继续遍历，再加上给节点上锁的机制，就完成了多线程的控制。这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。

图解扩容

触发扩容的操作：

假设目前数组长度为8，数组的元素的个数为5。再放入一个元素就会触发扩容操作。

总结一下扩容条件：

(1) 元素个数达到扩容阈值。

(2) 调用 putAll 方法，但目前容量不足以存放所有元素时。

(3) 某条链表长度达到8，但数组长度却小于64时。

CPU核数与迁移任务hash桶数量分配(步长)的关系

单线程下线程的任务分配与迁移操作

多线程如何分配任务？

普通链表如何迁移？

首先锁住数组上的Node节点，然后和HashMap1.8中一样，将链表拆分为高位链表和低位链表两个部分，然后复制到新的数组中。

什么是 lastRun 节点？

红黑树如何迁移？

hash桶迁移中以及迁移后如何处理存取请求？

多线程迁移任务完成后的操作

get操作

get方法无疑是在HahsMap还是在ConcurrentHashMap中最好理解的方法了。
返回指定键映射到的值，如果此映射不包含该键的映射，则返回null 。

和HashMap一样用equals来判断是否相等，也就是说必须Hash索引和equals一样，才算相同。

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    int h = spread(key.hashCode());//运用key的hashCode()计算出哈希地址
    
    //table不为空 且 table长度大于0 且 计算出的下标上用tabAt CAS取值不为空，也就是这个位置上有人，那么就进去找。
    if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) {  
        
        //如果哈希地址、键key相同则表示查找到，返回value，这里查找到的是头节点，就不用往下找了。
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        //eh=-2ash值<0 可能是Tree树形，可能e节点为ForwardingNode结点。
        //如果eh=-1就说明e节点为ForwardingNode,这说明什么，说明这个节点已经不存在了，被另一个线程正则扩容所以要查找key对应的值的话，直接到新newtable找
        //如果是eh=-2 说明e节点为Tree根结点，那么后面就以红黑树的方式查找。
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;//eh=-1这里调用的find其实就是ForwardingNode中的find方法了。
        //排除完上面情况，咱就正常找
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

总结流程：

调用spread()方法计算key的hashCode()获得哈希地址。
计算出键key所在的下标，算法是(n – 1) & h，如果table不为空，且下标上的bucket不为空，则到bucket中查找。
如果bucket的头节点的哈希地址小于0，有可能e结点是树的根节点，那么就按照红黑树的方式查找。还有可能说明e节点可能为ForwardingNode,这说明什么，说明这个节点已经不存在了，被另一个线程正则扩容所以要查找key对应的值的话，直接到新newtable找。
找到则返回该键key的值，找不到则返回null。

可能有人要问了？put的时候加锁了，get读的时候不需要锁吗？

这要归功于使用的tabAt中的了Unsafe的getObjectVolatile，因为table是volatile类型，所以对tab[i]的原子请求也是可见的。因为如果同步正确的情况下，根据happens-before原则，对volatile域的写入操作happens-before于每一个后续对同一域的读操作。所以不管其他线程对table链表或树的修改，都对get读取可见。

总结与思考

其实可以看出JDK1.8版本的ConcurrentHashMap的数据结构已经接近HashMap，相对而言，ConcurrentHashMap只是增加了同步的操作来控制并发，从JDK1.7版本的ReentrantLock+Segment+HashEntry，到JDK1.8版本中synchronized+CAS+HashEntry+红黑树,相对而言，总结如下思考

JDK1.8的实现降低锁的粒度，JDK1.7版本锁的粒度是基于Segment的，包含多个HashEntry，而JDK1.8锁的粒度就是HashEntry（首节点）
JDK1.8版本的数据结构变得更加简单，使得操作也更加清晰流畅，因为已经使用synchronized来进行同步，所以不需要分段锁的概念，也就不需要Segment这种数据结构了，由于粒度的降低，实现的复杂度也增加了
JDK1.8使用红黑树来优化链表，基于长度很长的链表的遍历是一个很漫长的过程，而红黑树的遍历效率是很快的，代替一定阈值的链表，这样形成一个最佳拍档
JDK1.8为什么使用内置锁synchronized来代替重入锁ReentrantLock，我觉得有以下几点
1. 因为粒度降低了，在相对而言的低粒度加锁方式，synchronized并不比ReentrantLock差，在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界，更加的灵活，而在低粒度中，Condition的优势就没有了
2. JVM的开发团队从来都没有放弃synchronized，而且基于JVM的synchronized优化空间更大，使用内嵌的关键字比使用API更加自然
3. 在大量的数据操作下，对于JVM的内存压力，基于API的ReentrantLock会开销更多的内存，虽然不是瓶颈，但是也是一个选择依据

1.扩容期间在未迁移到的hash桶插入数据会发生什么？

答：只要插入的位置扩容线程还未迁移到，就可以插入，当迁移到该插入的位置时，就会阻塞等待插入操作完成再继续迁移。

2.正在迁移的hash桶遇到 get 操作会发生什么？

答：在扩容过程期间形成的 hn 和 ln链是使用的类似于复制引用的方式，也就是说 ln 和 hn 链是复制出来的，而非原来的链表迁移过去的，所以原来 hash 桶上的链表并没有受到影响，因此如果当前节点有数据，还没迁移完成，此时不影响读，能够正常进行。如果当前链表已经迁移完成，那么头节点会被设置成fwd节点，此时get线程会帮助扩容。