并发集合：ConcurrentHashMap解析

1、ConcurrentHashMap 介绍

1.1、ConcurrentHashMap 概述

ConcurrentHashMap 是线程安全的HashMap，但最早的线程安全的HashMap 是 HashTable

，HashTable 现在已经弃用，因为它是使用synchronized 来保证线程安全，性能比较低，现在

使用最多的是 ConcurrentHashMap，在jdk1.7里面 ConcurrentHashMap 使用分段锁Segment

来保证线程安全；但分段锁Segment 在数据量比较大的时候，性能提升并不是太大，在jdk1.8

中为了进一步提升性能，采用 “CAS+synchronized” 桶锁的方式来保证线程安全；

1.2、ConcurrentHashMap 存储结构

ConcurrentHashMap 在jdk1.8中，底层存储结构其实就是：数组（桶）+链表+红黑树；

ConcurrentHashMap 存储数据时，先通过hash算法找到该数据要存储到那个数组元素中

（即桶中），将数据封装成Node节点以链表的形式保存到该桶中，若桶中的节点个数大于8，

且数组长度大于64，为了提升查询效率，则桶中的链表会转换成红黑树。

ConcurrentHashMap 存储结构大致如下图所示：

2、ConcurrentHashMap 核心属性

2.1、ConcurrentHashMap 核心属性

//数组table的最大长度
private static final int MAXIMUM_CAPACITY = 1 << 30;
//默认table数组长度
private static final int DEFAULT_CAPACITY = 16;
//链表长度大于8，则被转换成红黑树
static final int TREEIFY_THRESHOLD = 8;
//如果红黑树的长度小于6，则需要把红黑树转为链表
static final int UNTREEIFY_THRESHOLD = 6;
//最小转化为红黑树的table数组长度
static final int MIN_TREEIFY_CAPACITY = 64;
//最小迁移长度
private static final int MIN_TRANSFER_STRIDE = 16;
//sizeeCtl中用于生成戳记的位数。对于32位数组，必须至少为6。
private static int RESIZE_STAMP_BITS = 16;
//低16位全是1
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;//值也是16
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;//-1 表示当前位置的数据正在扩容
static final int MOVED     = -1; 
//-2 表示当前位置下挂载的是一个红黑树
static final int TREEBIN   = -2; 
//-3 表示当前索引位置被占用（预留索引为止）
static final int RESERVED  = -3; 
//保证hash值一定是正数
static final int HASH_BITS = 0x7fffffff; //CPU内核数 
static final int NCPU = Runtime.getRuntime().availableProcessors();//数组，即桶， 数组的每个位置存放的是node节点
transient volatile Node<K,V>[] table;//扩容时用到的新数组
private transient volatile Node<K,V>[] nextTable;//基本计数器值，记录当前元素个数，主要在没有竞争时使用，但在table初始化竞争期间也用作回退。通过CAS更新。
private transient volatile long baseCount;/**是数组table初始化或扩容时的一个控制变量： -1：表示当前数组table正在初始化 小于-1：表示当前数组table正在扩容的线程个数（如果一个线程正在扩容，则值为-2，如果2个线程正在扩容，则值为-3） 0：表示当前线程还没有初始化 大于0：表示当前数组 table 扩容的阈值，或者是当前数组的初始化大小
*/
private transient volatile int sizeCtl;//迁移数据时用到的标识
private transient volatile int transferIndex;//CounterCell 数组
private transient volatile CounterCell[] counterCells;

2.2、ConcurrentHashMap 内部类 TreeBin

TreeBin 表示一个红黑树，在TreeBin内部除了红黑树的节点外，还维护了一个单项链表；

当根据key获取数据时，若有其他线程正在往红黑树中写数据（即此时红黑树的锁由写线程

持有），则会从单项链表中查找数据。

TreeBin 核心属性和构造函数如下：

 /*** TreeBin 锁操作没有基于AQS，仅仅是对一个变量操作和一些业务判断实现。* 每次读线程操作 lockState+4* 每次写线程操作 lockState+1* 如果读线程正在持有锁，就先对 lockState+2，waiter指向当前线程，并挂起当前线程*/static final class TreeBin<K,V> extends Node<K,V> {/*** 红黑树根节点*/TreeNode<K,V> root;/*** 双向链表第一个节点，即头节点*/volatile TreeNode<K,V> first;/*** 等待获取写锁的线程* todo 注意：*      若有读线程正在读取红黑树的数据，则写线程会阻塞，这是为了保证红黑树的平衡*      如果有写线程正在往红黑树中写数据，此时读线程不会阻塞，读线程会读取双向链表*      读-读 操作不会阻塞*/volatile Thread waiter;/*** 当前 TreeBin 的锁状态*/volatile int lockState;/*** lockState 的常量值，对锁状态进行运算的一些值* 1：有线程正在获取写锁* 2：有写线程在等待获取写锁* 4：读线程，在红黑树检索中，需要先对 lockState+READER 操作*    这个只会在读操作中遇到*/static final int WRITER = 1; static final int WAITER = 2; static final int READER = 4; /**** 将双向链表转为红黑树的操作* 参数：*    b: 双向链表的第一个节点*/TreeBin(TreeNode<K,V> b) {//调用父类Node构造函数，创建一个Node节点，并将Node的hash值设置为-2super(TREEBIN, null, null, null);this.first = b;//r最后会被赋值给跟节点TreeNode<K,V> r = null;/*** 遍历双向链表，将双向链表转换为红黑树*/for (TreeNode<K,V> x = b, next; x != null; x = next) {next = (TreeNode<K,V>)x.next;//先将每个节点的左右子节点清空x.left = x.right = null;//如果r等于null，表示是第一个节点，则设置第一个节点为根节点（根节点父节点为null）//但这个根节点并不一定是最终的根节点if (r == null) {x.parent = null;x.red = false;r = x;}else {//这里已经有根节点，新插入的节点要作为父节点的左子树或右子树//获取节点的keyK k = x.key;//获取当前节点的hash值int h = x.hash;Class<?> kc = null;for (TreeNode<K,V> p = r;;) {/*** dir: 若为-1，表示要插入其父节点的左边（即左子树）；若为1，表示要插入其父节点的右边（即右子树）* ph: 父节点的hash值*/int dir, ph;//父节点的keyK pk = p.key;//若父节点的hash值大于当前节点的hash值，则表示当前节点插入父节点p的左边，// 若父节点的hash值小于当前节点的hash值若则插入父节点p的右边//若父节点的hash值等于当前节点的hash值if ((ph = p.hash) > h)dir = -1;else if (ph < h)dir = 1;//若父节点的hash值等于当前节点的hash值，则基于Comparable判断dirde 值else if ((kc == null &&(kc = comparableClassFor(k)) == null) ||(dir = compareComparables(kc, k, pk)) == 0)dir = tieBreakOrder(k, pk);//获取当前父节点TreeNode<K,V> xp = p;//将p指向p的left或right，若p的left或right节点等于Null，表示该位置是空的，当前节点正好可以插入到该位置if ((p = (dir <= 0) ? p.left : p.right) == null) {//将当前节点x的父节点设置为px.parent = xp;//插入当前节点xif (dir <= 0)xp.left = x;elsexp.right = x;//插入一个数据（节点后）需要做一次平衡操作r = balanceInsertion(r, x);break;}}}}//设置树的根节点this.root = r;assert checkInvariants(root);}
}

3、ConcurrentHashMap 存储数据流程

3.1、put(K key,V value) 方法

put 方法是ConcurrentHashMap 中基本的存储数据的方法，若key不存在，则直接添加，若

key 已存在，则拿当前传入的value值覆盖已经存在的value的值

put 方法如下：

3.2、putIfAbsent(K key, V value) 方法

putIfAbsent 方法也是存储数据的方法，与put的方法的区别是，若key已经存在，则什么也

不做，并返回已存在的数据；若key不存在时，则正常添加。

putIfAbsent 方法结构如下：

3.3、putVal(K key, V value, boolean onlyIfAbsent) 方法

putVal方法是ConcurrentHashMap 存储数据的具体实现方法，put 和 putIfAbsent 都是

通过 putVal 来完成数据存储的。

putVal方法代码如下：

  /*** todo 核心方法* 存放数据** 参数：*    onlyIfAbsent: true=表示当2次调用该方法，传入的key一致时，第二次什么也不做，并返回已存在的数据,key不存在则正常添加*                  false=表示当2次调用该方法，传入的key一致时，第二次会覆盖原来的数据，并返回已存在的数据，若 key不存在则正常添加*//** Implementation for put and putIfAbsent */final V putVal(K key, V value, boolean onlyIfAbsent) {if (key == null || value == null) throw new NullPointerException();/*** 计算hash，后期当前key-value要存储到那个数组索引为止（即存储到哪个桶中）*/int hash = spread(key.hashCode());//一个标识，hash冲突可以用到int binCount = 0;//循环遍历数组（桶的每个位置），这是个死循环for (Node<K,V>[] tab = table;;) {Node<K,V> f; //数组table当前i位置的数据node/*** n: 数组长度* i: 当前node节点要存放的索引为止* fh: 数组table中i位置数据的hash值*/int n, i, fh;//判断当前数组 table 是不是还没初始化if (tab == null || (n = tab.length) == 0)//初始化数组tabletab = initTable();else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {/*** (n - 1) & hash : 根据hash 计算当前node节点需要存放的索引位置i* 然后再根据方法 tabAt 获取到位置i的数据*///执行到这里，表示当前数组在索引i位置上没有数据（即不存在hash冲突），// 基于CAS将当前数据的node节点保存在位置i处，若保存成功，则退出循环if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))break;                   // no lock when adding to empty bin}else if ((fh = f.hash) == MOVED)//协助扩容tab = helpTransfer(tab, f);else {/*** todo 注意：*     执行到这里说明存在hash冲突，即数组tale中位置i处有数据*/V oldVal = null;//加锁，以table[i]位置的Node作为锁对象synchronized (f) {//判断当前table[i]位置的数据是否被修改（并发下 table[i] 位置的数据可能被修改）if (tabAt(tab, i) == f) {//再次判断节点的hash值是否大于等于0，if (fh >= 0) {//表示f是一个普通的链表//binCount 设置为1，用于记录链表的长度，//当链表长度大于8，俩表需要转换为红黑树binCount = 1;for (Node<K,V> e = f;; ++binCount) {K ek;//判断当前table[i]位置的链表的每个数据的hash值与当前put数据的hash是否一致，且key是否一致、//这说明key一样，这时需要考虑是否覆盖key的原有数据，然后退出循环if (e.hash == hash &&((ek = e.key) == key ||(ek != null && key.equals(ek)))) {//获取旧的key的值oldVal = e.val;//判断是否覆盖相同key的值，若 onlyIfAbsent=false,则覆盖老的数据if (!onlyIfAbsent)e.val = value;//退出循环break;}//将当前Put数据的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) {//表示f是一个树节点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) {//若链表长度大于等于8，则将链表转换为红黑树或扩容//如果数组table的长度没达到64的话，会先将数组table扩容if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);//如果出现了数据覆盖的情况，则返回覆盖之前的数据if (oldVal != null)return oldVal;break;}}}addCount(1L, binCount);return null;}

3.4、spread(int h) 方法

该方法功能是根据key的hashCode值计算key的hash值

spread 方法代码如下：

3.5、putTreeVal(int h, K k, V v) 方法

该方法功能是将key-value添加到红黑树中，是内部类TreeBin的方法

putTreeVal 代码如下：

/*** 向红黑树中添加数据* 整体判断就是判断当前数据是插入到红黑树的左子树还是右子树、或者是覆盖操作* 确定插入左子树或右子树后，直接维护双向链表和红黑树结构，并再判断是否需要自平衡* 参数：*     h: 当前k 的hash值*     k: 新增数据的key*     v: 新增数据的值*/  final TreeNode<K,V> putTreeVal(int h, K k, V v) {// 保存comparableClassFor判断的结果Class<?> kc = null;//搜索节点boolean searched = false;//死循环，p是根节点的临时引用for (TreeNode<K,V> p = root;;) {//dir: 用来表示节点是插入左子树还是右子树的标记，-1:左子树，1：右子树//ph: 当前p节点的hash值（注意：最终是父节点的Hash值）int dir, ph;//K pk;if (p == null) {first = root = new TreeNode<K,V>(h, k, v, null, null);break;}//根据当前节点的hash值与要插入节点的hash值h比较，判断新节点是插入到当前节点p的左子树还是右子树else if ((ph = p.hash) > h)dir = -1;else if (ph < h)dir = 1;else if ((pk = p.key) == k || (pk != null && k.equals(pk)))//如果当前节点p的key与新节点的key相等，则直接返回，由putVal去修改数据return p;//hash值一致，但key 的“==” 与“equals” 都不一样，则需要基于Compare去判断else if ((kc == null &&(kc = comparableClassFor(k)) == null) ||//dir等于0，表示基于Compare 判断也是一样的(dir = compareComparables(kc, k, pk)) == 0) {//开启搜索操作，查看是否有相同的key，如果找到了key一致的数据，就表示可以覆盖数据，就直接返回//注意：这个搜索操作只有第一次会执行，因为执行一次后 searched 就被设置为trueif (!searched) {TreeNode<K,V> q, ch;searched = true;if (((ch = p.left) != null &&(q = ch.findTreeNode(h, k, kc)) != null) ||((ch = p.right) != null &&(q = ch.findTreeNode(h, k, kc)) != null))return q;}//再次判断hash大小，若小于等于，就返回-1dir = tieBreakOrder(k, pk);}//xp: 父节点的临时节点TreeNode<K,V> xp = p;//基于dir判断是插入左子树还是右子树，并重新给p赋值if ((p = (dir <= 0) ? p.left : p.right) == null) {//first引用TreeNode<K,V> x, f = first;//构建新节点first = x = new TreeNode<K,V>(h, k, v, f, xp);/*** 因为当前TreeBinc除了红黑树，还维护了一个双向链表* 这里是维护双向链表的操作，将新节点放入链表的首位*/if (f != null)f.prev = x;//新节点插入到父节点的左侧if (dir <= 0)xp.left = x;else//新节点插入到父节点的右侧xp.right = x;if (!xp.red)//若父节点是黑色的，则设置当前节点x为红色x.red = true;else {//说明新插入节点x是黑色的//加锁lockRoot();try {//红黑树自平衡调节root = balanceInsertion(root, x);} finally {//释放锁unlockRoot();}}break;}}//检查红黑树结构assert checkInvariants(root);//返回null表示插入了新节点return null;}

3.6、treeifyBin(Node<K,V>[] tab, int index) 方法

该方法的功能是数组table扩容，或将链表转换为红黑树

private final void treeifyBin(Node<K,V>[] tab, int index) {Node<K,V> b;int n, sc;if (tab != null) {//判断数组table的长度是达到“最小转换为红黑树的长度”，若没达到，则尝试扩容数组tableif ((n = tab.length) < MIN_TREEIFY_CAPACITY)//尝试扩容数组table，将数组长度扩容为原来的2倍// n << 1 n 左移一位，相当于*2tryPresize(n << 1);//将链表转换为红黑树else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {synchronized (b) {if (tabAt(tab, index) == b) {//将单项链表转换为一个双向链表，聊表中每个节点都是 TreeNodeTreeNode<K,V> hd = null, tl = null;for (Node<K,V> e = b; e != null; e = e.next) {TreeNode<K,V> p =new TreeNode<K,V>(e.hash, e.key, e.val,null, null);if ((p.prev = tl) == null)hd = p;elsetl.next = p;tl = p;}//在new TreeBin时将双向链表转换为红黑树setTabAt(tab, index, new TreeBin<K,V>(hd));}}}}}

3.7、tryPresize(int size) 方法

该方法功能是将数组table长度扩容到size大小。

private final void tryPresize(int size) {/*** 获取数组table新的长度 c* 首先判断当前size 是否大于等于 最大数组长度MAXIMUM_CAPACITY的一半，若大于，则把数组长度扩容到 MAXIMUM_CAPACITY；* 若不大于，则需要保证数组的长度是2的n次方** 这段操作是为了table初始化操作准备的，当ConcurrentHashMap 实例化时，若构造方法参数是一个集合，则* 会调用 putAll 方法，putAll方法也会触发 tryPresize 方法进行初始化*/int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :tableSizeFor(size + (size >>> 1) + 1);int sc;//将 sizeCtl 赋值给sc，并判断是否大于0//若sc大于0，表示此时没有初始化操作，也没有扩容操作while ((sc = sizeCtl) >= 0) {Node<K,V>[] tab = table;//声明数组长度nint n;if (tab == null || (n = tab.length) == 0) {//表示数组table 没有初始化，putAll 方法的触发的初始化才会执行到这里//直行到这里，表示需要初始化数组table//比较sc与c的大小，以较大的数作为数组table的长度n = (sc > c) ? sc : c;//基于CAS将sizeCtl的值修改为-1，表示数组table正在初始化if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {//再次判断table是否被其他线程初始化if (table == tab) {//初始化数组table@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = nt;//计算下一次扩容阈值sc = n - (n >>> 2);}} finally {//最后更新 sizeCtl 的值sizeCtl = sc;}}}//如果计算的数组table长度c小于sc，表示扩容已经完成，则直接退出循环，结束//或者 table当前长度已经大于 MAXIMUM_CAPACITY，表示table无法扩容，则也退出循环结束else if (c <= sc || n >= MAXIMUM_CAPACITY)break;//若 table != tab，表示并发下有其他线程修改了数组table，则进入下一次循环else if (tab == table) {//n是数组table的长度//计算扩容标识戳，根据当前数组的长度，计算一个16位的扩容标识戳/*** rs作用：*   1）保证后边 SIZECTL 赋值值，保证SIZECTL是小于-1的小数*   2）用来记录当前是从什么长度开始扩容的*/int rs = resizeStamp(n);//执行到这里，若sc小于0，表示当前有线程正在扩容//todo 注意：此时sc一定不是-1，因为前边 tab == table 已经判断table已经完成了初始化，sc=-1表示table正在初始化if (sc < 0) {//sc 小于0永远进不去，是一个bug/*** todo 注意：*    执行到这里说明有其他线程正在对table进行扩容（rs<-1），来协助扩容的操作*/Node<K,V>[] nt;if ((sc >>> RESIZE_STAMP_SHIFT) != rs //当前线程扩容时，老数组长度是否与当前线程扩容时的老数组长度一致//下边2个判断都有问题，核心应该是先把rs左移16位，然后再追加当前值|| sc == rs + 1  //判断当前扩容是否已经即将结束，正确写法：rs << 16 + 1|| sc == rs + MAX_RESIZERS//判断当前扩容线程是否达到了最大限度，正确写法 rs << 16 + MAX_RESIZERS|| (nt = nextTable) == null //若新数组等于null，表示扩容已经操作完成|| transferIndex <= 0) //transferIndex 用来记录迁移的索引位置，他会从高位往低位迁移，transferIndex<=0表示数据已经迁移完成break;//如果线程需要协助扩容，首先先把sc加1，即表示增加了一个线程来帮助扩容if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))//nt 表示新数组transfer(tab, nt);}/*** 执行到这里说明，当前线程是第一个来扩容的线程** 扩容戳rs左移16位后，首位（符号位）是1，表示这个数是负数，rs 高16位表示是扩容戳，低16位表示正在扩容的线程个数* 如这里：rs左移16位，然后加2，之所以加2，是为了表示当前有 （2-1）1个线程在扩容（若加10，则表示有9个线程在扩容）* todo 问题：为什么rs低16位是2时，表示只有一个线程正在扩容操作？*           每一个线程扩容完成后，都会对rs低16位进行减1操作，当最后一个线程完成扩容操作后，减1后的结果还是-1；*           当值为-1时，要对老数组进行一次扫描，查看是否有数据没有迁移到新数组** 基于CAS 将 sizeCtl 的值由sc修改成rs，若修改成功则调用 transfer 执行扩容*/else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))//调用 transfer，并且将第二个参数设置为null，表示是第一次扩容transfer(tab, null);}}}

3.8、transfer(Node<K,V>[] tab, Node<K,V>[] nextTab)

transfer 是真正的扩容方法

参数： tb: 老数组

nextTab: 新数组，当多个线程同时扩容时，nextTab可能不为null

transfer 方法代码如下：

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {//n:老数组的长度//stride：每个扩容线程一次性迁移多少数据到新数组，迁移数据的步长或间隔int n = tab.length, stride;/*** 根据CPU内核数来设置一次性迁移多少长度的数据最合理* 如果计算结果大于16，则使用计算结果，若小于16，则使用最小迁移长度16*/if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // subdivide range//第一个进来扩容的线程需要把新数组初始化if (nextTab == null) {            // initiatingtry {//将原数组长度左移1位构建新数组，若原数组长度n太大，左移1位后超过了Integer.MAX_VALUE，则抛出异常@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];nextTab = nt;} catch (Throwable ex) {      // try to cope with OOMEsizeCtl = Integer.MAX_VALUE; //达到数组长度最大的取值范围//设置 sizeCtl 后直接结束，表示当前数组长度是不需要扩容的return;}//将成员变量的新数组赋值nextTable = nextTab;//设置新数组数据迁移下标，默认值是老数组长度transferIndex = n;}//新数组长度int nextn = nextTab.length;/*** 在老数组迁移完数据后做的标识*/ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);/***advance 默认为true,*        true=表示当前线程需要接收任务，然后再执行迁移*        false=表示已经接收完成任务*/boolean advance = true;// finishing 用来标识是否迁移结束，true=数据迁移结束，false=未结束boolean finishing = false; // to ensure sweep before committing nextTab/*** i: 当前数据迁移数据的索引值* bound: 需要迁移的数据范围*/for (int i = 0, bound = 0;;) {Node<K,V> f;int fh;//当前线程需要接收任务while (advance) {int nextIndex, nextBound;//对变量i进行了修改，i -= 1，并判断当前任务是否处里完毕//第一次进来，这2个判断肯定进不去if (--i >= bound || finishing)advance = false;else if ((nextIndex = transferIndex) <= 0) {//判断transferIndex 是否小于等于0，若transferIndex<=0 表示数据已经迁移完毕，没有任务可以领取//在线程领取任务后会对 transferIndex 进行修改，修改为 transferIndex -= stridei = -1;advance = false;}//当前线程开始读取任务//采用CAS方式修改 nextIndex（值transferIndex）的值，修改为减去步长strideelse if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ? //nextIndex - stride : 0))) {//对bound赋值bound = nextBound;//修改ii = nextIndex - 1;//advance 设置为false,表示当前线程领取到了任务advance = false;}}/*** todo 执行到这里，表示开始迁移数据，并且在数据迁移完毕后，会将 advance 设置为true** i < 0 ：当前没有现成接收任务* i >= n：迁移索引为止不可能大于等于老数组长度n，这个判断永不成立* i + n >= nextn：i最大值就是旧数组长度，再加上旧数组长度也不会大于新数组长度nextn，这个也不会成立*/if (i < 0 || i >= n || i + n >= nextn) {//如果执行到这里，表示当前线程没有接收到任务int sc;//判断扩容是否已经结束，若扩容已经结束，则将 nextTable 设置为null，将table执行新的数组//并将sizeCtlif (finishing) {//扩容完成，将新数组nextTable设置为nullnextTable = null;table = nextTab;//重新计算扩容阈值，扩容戳 0.75sizeCtl = (n << 1) - (n >>> 1);//真正结束扩容return;}/*** 若当前线程没接收到任务，则让当前简称结束扩容操作* 采用CAS方式，将sizeCtl-1，表示当前执行扩容操作的线程数减少了一个* CAS操作成功，当前线程退出*/if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {/*** (sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT 表示当前线程并不是最后一个退出扩容的线程，则直接结束*/if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)return;//执行到这里，表示当前线程是最后一个退出扩容的线程，则将 finishing 和 advance 设置为true,表示扩容结束finishing = advance = true;//将i设置为老数组长度，目的是为了让最后一个线程再从老数组从尾到头检查一遍，检查数据是否全部迁移i = n; // recheck before commit}}/*** 真正数据迁移，并且在数据迁移完毕后，会将 advance 设置为true*/else if ((f = tabAt(tab, i)) == null) //通过tabAt 方法拿到老数组tab位置i的数据，若i位置的数据为null，表示i位置的数据已经迁移，并将i位置修改成ForwardingNodeadvance = casTabAt(tab, i, null, fwd);else if ((fh = f.hash) == MOVED)//判断老数组i位置的Node的hash是否等于MOVED（-1），这一步一般是在数据迁移完成后，最后一个迁移线程再次扫描检查老数组数据是否迁移时执行的//i位置的数据完成迁移advance = true; // already processedelse {/*** 执行到这里说明 老数组当前i位置（桶）有数据*/synchronized (f) {//锁定当前位置桶/*** 判断之前取出的i位置的数据f是否与当前i位置的数据相等* 这一步是为了避免并发下，其他线程对i位置的数据做了修改*/if (tabAt(tab, i) == f) {/*** ln: lowNode* hn: highNode*/Node<K,V> ln, hn;//fh是当前桶位置的数据node的hsh值//fh >= 0 表示当前老数组i位置（桶）的数据正常，且不是红黑树，当做链表处里if (fh >= 0) {//fh & n ：运算结果只有2种，要么是0要么是n，数组长度n是2的m次幂，说明n的最低位是0int runBit = fh & n;Node<K,V> lastRun = f;/*** 循环的目的是为了得到链表下经过 hash&n 计算结果一致的最后一些数据，* 在迁移数据时，只需要迁移到lastNode即可，剩下的node指针不需要改变*/for (Node<K,V> p = f.next; p != null; p = p.next) {int b = p.hash & n;if (b != runBit) {runBit = b;lastRun = p;}}//如果 runBit==0。则把lastRun 赋值给ln，否则把值赋值给hnif (runBit == 0) {ln = lastRun;hn = null;}else {hn = lastRun;ln = null;}//循环得到lastRun指向的数据即可，后续不需要再遍历for (Node<K,V> p = f; p != lastRun; p = p.next) {//获取当前node的key值、hash值、val值int ph = p.hash; K pk = p.key; V pv = p.val;//如果 当前节点p的hash值与数组长度n做与运算，若结果为0，就把当前节点p挂载到lowNode//否则，把p挂载到hignNodeif ((ph & n) == 0)ln = new Node<K,V>(ph, pk, pv, ln);elsehn = new Node<K,V>(ph, pk, pv, hn);}//采用CAS将ln挂载到新数组的i位置，即原位置setTabAt(nextTab, i, ln);//采用CAS将hn挂载老新数组的i+n位置（n是老数组的长度）setTabAt(nextTab, i + n, hn);//采用CAS将老数组中的数据设置成fwdsetTabAt(tab, i, fwd);//迁移成功将advance 设置为trueadvance = true;}//红黑树的数据迁移//todo: 红黑树迁移过程中是使用 TreeBin 维护的双向链表来完成数据迁移的else if (f instanceof TreeBin) {TreeBin<K,V> t = (TreeBin<K,V>)f;//lo、hi是扩容后需要存放到新数组的低、高位的链表TreeNode<K,V> lo = null, loTail = null;TreeNode<K,V> hi = null, hiTail = null;//lc:记录低位数据的长度//hc:记录高位数据的长度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);//与老数组长度n做与运算，基于运算结果判断当前节点p是放低位还是高位if ((h & n) == 0) {//放低位if ((p.prev = loTail) == null)lo = p;elseloTail.next = p;loTail = p;++lc;}else {//放高位if ((p.prev = hiTail) == null)hi = p;elsehiTail.next = p;hiTail = p;++hc;}}//封装低位节点，如果红黑树节点个数小于等于6，则把红黑树转为链表//如果树节点个数大于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;//新数组的i位置，放低位节点setTabAt(nextTab, i, ln);//新数组的i+n位置放高位节点setTabAt(nextTab, i + n, hn);//设置老数组i位置的数据已迁移setTabAt(tab, i, fwd);//advance = true;}}}}}}

3.9、helpTransfer(Node<K,V>[] tab, Node<K,V> f) 方法

该方法功能是协助扩容

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {//新数组Node<K,V>[] nextTab;//sc: sizeCtl的临时变量int sc;/*** 1、判断老数组不为null* 2、判断新数组不为null，并将新数组赋值给nextTab*/if (tab != null && (f instanceof ForwardingNode) &&(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {//ConcurrentHashMap 正在扩容//基于老数组长度计算扩容戳int rs = resizeStamp(tab.length);/*** nextTab == nextTable：判断fwd中的新数组是否等于当前的新数组，若相等，则表示当前线程可以协助扩容；*                       若不相等，则有2种情况：1）扩容已经结束，2）开启了新的扩容* table == tab：判断并发下老数组有没有被其他线程修改，若相等，则可以协助扩容，若不相等，则表示扩容结束了* (sc = sizeCtl) < 0：将sizeCtl的值赋值给sc，并判断 sc是否小于0，sc<0 此时sc只能是小于-1，表示数组正在扩容*/while (nextTab == nextTable && table == tab &&(sc = sizeCtl) < 0) {//将sc 右移16位，然后判断sc是否与扩容戳一致，若不一致，则说明扩容长度不一样，当前线程则退出协助扩容/*** sc == rs + 1：是一个bug，应该先把rs左移16位再加1，即 sc == rs<<16 + 1，表示如果 rs<<16 + 1与sc相等，说明扩容已经到了最后检查的阶段* sc == rs + MAX_RESIZERS ：是一个bug，应该先把rs左移16位再加1，即 sc == rs<<16 + MAX_RESIZERS*          rs<<16 + MAX_RESIZERS 若等于sc，表示协助扩容的线程已经达到最大值* transferIndex <= 0：扩容时是从老数组table的高索引位置开始迁移数据的，若 transferIndex 不大于0，表示迁移数据的任务都被领完了，所以*                    当前线程也不需要协助扩容*/if ((sc >>> RESIZE_STAMP_SHIFT) != rs|| sc == rs + 1 //|| sc == rs + MAX_RESIZERS //是一个bug，|| transferIndex <= 0)//break;//执行到这里说明当前线程能协助扩容//基于CAS将sc值加1，表示协助扩容的线程增加一个if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {//协助扩容//todo: 注意这里肯定不是第一个参与扩容的线程，因为第一个扩容的线程需要创建新数组，调用transfer 时//      第二个参数传入的是nulltransfer(tab, nextTab);break;}}return nextTab;}return table;}

3.10、addCount(long x, int check) 方法

addCount 方法本身就是为了记录 ConcurrentHashMap 中元素的个数
addCount主要由2方面功能组成
1）计数器：若添加数据成功，计数器需要自增新添加的数据个数
2）检查当前 ConcurrentHashMap 是否需要扩容

addCount方法代码如下：

 /*** addCount 方法本身就是为了记录 ConcurrentHashMap 中元素的个数* addCount主要由2方面功能组成*    1）计数器：若添加数据成功，计数器需要自增 新添加的数据个数*    2）检查当前 ConcurrentHashMap 是否需要扩容** 参数：*    x:*    check: 链表的长度*           注意：若添加数据时没有出现hash冲突，check 就等于0*                若桶中的数据是红黑树，则 check 的值固定是2*/private final void addCount(long x, int check) {//声明一个 CounterCell 数组，CounterCell[] as;//s: 自增后的元素个数//b: 原来的baseCount值long b, s;//判断 counterCells 不为null，表示有并发，之前有冲突问题//如果 counterCells 等于null，则直接通过CAS操作，修改 baseCount 的值// baseCount 记录当前元素个数//如果对 baseCount 操作成功，则直接结束，如果CAS失败，则进入if中if ((as = counterCells) != null ||!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {//todo :执行到这，则说明可能有并发问题CounterCell a; long v; int m;//是否有冲突，默认是true，表示没有冲突boolean uncontended = true;if (as == null || (m = as.length - 1) < 0 || //判断 CounterCell[] 是否已经被初始化，若没初始化，则执行fullAddCount初始化CounterCell[]数组//ThreadLocalRandom.getProbe() 返回一个随机数(a = as[ThreadLocalRandom.getProbe() & m]) == null ||//这里 CounterCell[]已经初始化了，基于随机数拿到一个数组上的 CounterCell 对象，并判断是否为空，若为空表示数组中的值还没初始化，则执行fullAddCount初始化CounterCell//到这里说明数组指定位置上的 CounterCell 初始化完成，则执行CAS修改指定位置的CounterCell的value值//若CAS操作成功，则直接结束，若CAS操作失败，则说明有冲突，uncontended会被设置为false，然后执行 fullAddCount 方法!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {fullAddCount(x, uncontended);return;}//如果链表长度小于等于1，表示数组table当前位置只有一个数据，则就没必要执行扩容了if (check <= 1)return;//将CounterCell中的数据value追加到 baseCounts = sumCount();}//=============================== 上边，记录ConcurrentHashMap 元素个数 ==============//=============================== 下边，检查是否需要扩容 ============================//check大于等于0，注意REMOVE 操作时 check小于0if (check >= 0) {Node<K,V>[] tab, nt; int n, sc;//s >= (long)(sc = sizeCtl): 判断当前元素个数是否大于扩容阈值while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&(n = tab.length) < MAXIMUM_CAPACITY) {//数组长度最大值判断//扩容标识戳int rs = resizeStamp(n);//sc小于0，表示可能正在扩容if (sc < 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();}}}

4、ConcurrentHashMap 存储数据流程

4.1、get(Object key) 方法

该方法功能是根据key从ConcurrentHashMap 获取数据，若key不存在，则返回null；

get 方法代码如下：

public V get(Object key) {// tab: 数组，桶// e: 数组中指定位置的节点数据，// p: 根据key查询到的节点Node<K,V>[] tab; Node<K,V> e, p;//n: 数组长度//eh: 数组中指定位置的节点的hash值//ek: 数组中指定位置的节的key值int n, eh; K ek;//计算key的hash值int h = spread(key.hashCode());if ((tab = table) != null && (n = tab.length) > 0 && //数组不为空，且数组中有数据(e = tabAt(tab, (n - 1) & h)) != null) {//获取到数组中指定位置的节点数据e//判断e的hash值与key的hash值是否相等if ((eh = e.hash) == h) {//判断e的key是否与参数key相等，若相等则表示e就是要查询的节点数据if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}else if (eh < 0)///*** 表示 e.hash<0 比较特殊，有三种情况* 1、数据已经被迁移走了* 2、位置被占用* 3、是一个红黑树（以红黑树作为重点）*/return (p = e.find(h, key)) != null ? p.val : null;//这里表示e是一个链表的头节点，遍历链表查询key相等的数据while ((e = e.next) != null) {if (e.hash == h &&((ek = e.key) == key || (ek != null && key.equals(ek))))return e.val;}}//key不存在return null;}

4.2、find(int h, Object k) 方法

该方法是红黑树TreeBin 的方法，功能是根据key的hash值和key从红黑树中获取数据

final Node<K,V> find(int h, Object k) {//判断键值k不为nullif (k != null) {/*** e: 红黑树中所维护的链表的头元素;*    遍历红黑树中的链表，从该链表中查询数据*/for (Node<K,V> e = first; e != null; ) {//TreeBin 的锁状态int s;//当前链表节点的keyK ek;//if (((s = lockState) & (WAITER|WRITER)) != 0) {//如果if进来，说明有写线程正在等待获取写锁 或者 有写线程正在持有写锁，此时会从双向链表中查询数据if (e.hash == h &&((ek = e.key) == k || (ek != null && k.equals(ek))))return e;e = e.next;}//执行到这里，说明没有现成在等待写锁或持有写锁，则从红黑树中查询数据//CAS修改 lockState，lockState+4：表示当前线程可以去红黑树中查询数据else if (U.compareAndSwapInt(this, LOCKSTATE, s,s + READER)) {TreeNode<K,V> r, p;try {p = ((r = root) == null ? null :r.findTreeNode(h, k, null));//遍历红黑树，从红黑树中查找数据} finally {Thread w;//最终，释放读锁//如果 lockState-1 等于 WAITER，说明可能有写线程在等待// waiter若不等于null，表示当前线程是左后一个在红黑树中完成检索的线程，同时有线程在等待写锁//，则当前读线程需要唤醒等待中的写线程if (U.getAndAddInt(this, LOCKSTATE, -READER) ==(READER|WAITER) && (w = waiter) != null)LockSupport.unpark(w);}return p;}}}return null;}