1 ConcurrentHashMap 简介

Map 一种存储键值对 (key-value) 的数据结构, 可以通过 key 快速地定位到需要的 value, 在 Java 中是一个使用频率很高的一个数据结构。一般情况下, 我们都是可以直接使用它的实现类 HashMap 就能满足需求了。
但是 HashMap 在多线程情况, 并不是一个线程安全的类。
解决的方式有很多, 例如:

1. 使用在 Java 体系中古老的 Hashtable 作为替代, 该类基本上所有的方法都采用 synchronized 进行线程安全的控制, 虽然保证了线程安全, 但是牺牲了很大的性能。 在高并发的情况下, 每次只有一个线程能够获取对象监视器锁, 这样的并发性能的确不令人满意。
2. 通过 Collections 的 synchronizedMap(Map<K,V> m) 方法返回一个线程安全的 Map。但是这个的效果实际上和 Hashtable 的一样, 依然是采用 synchronized 独占式锁进行线程安全的并发控制的, 所以这种方案的性能也是令人不太满意的。

那么有没有一种既保证了线程安全性, 性能也不错的 Map 呢? ConcurrentHashMap 就是我们的选择了, 内部利用了锁分段的思想提高了并发度。当然的, 随着 JDK 的升级, ConcurrentHashMap 的实现也有了不同的方式。
大体了分为 2 个版本: 1.6 版本的 和 1.8 版本。

ConcurrentHashMap 在 JDK 1.6 的版本网上资料很多, 有兴趣的可以去看看。 JDK 1.6 版本关键要素:

1. segment 继承了 ReentrantLock 充当锁的角色, 为每一个 segment 提供了线程安全的保障
2. segment 维护了哈希散列表的若干个桶, 每个桶由 HashEntry 构成的链表

而到了 JDK 1.8 的 ConcurrentHashMap 就有了很大的变化, 光是代码量就足足增加了很多。1.8 版本舍弃了 segment, 并且大量使用了 synchronized, 以及 CAS 无锁操作以保证 ConcurrentHashMap 操作的线程安全性。
至于为什么不用 ReentrantLock 而是 Synchronzied 呢? 实际上, synchronzied 在 JDK 1.6 做了很多的优化, 包括偏向锁, 轻量级锁, 重量级锁, 可以依次向上升级锁状态。
因此, 使用 synchronized 相较于 ReentrantLock 的性能会持平甚至在某些情况更优, 具体的性能测试可以去网上查阅一些资料。另外, 底层数据结构改变为采用数组 + 链表 + 红黑树的数据形式。

2 ConcurrentHashMap 的关键属性, 类和 CAS 方法

在了解 ConcurrentHashMap 的具体方法实现前, 我们需要系统的来看一下几个关键的地方, 从这几个地方, 从大体上了解 ConcurrentHashMap 的一些特性。

2.1 常量设置 (这些设置基本和 HashMap 的类似)

1. ConcurrentHashMap 内部是使用一个数组存放数据的, 这个数组的长度必须是 2 的 n 次方, 默认的容量是 16, 最大是 1 << 30, 既 2 的 30 次方
2. 默认负载因子为 0.75, 当数组的的存数据的项 >= 数组的长度 * 负载因子, 就会进行数组长度扩容
3. 数组的每一项, 在 JDK 1.8 中, 可以是链表, 也可以是红黑树。
4. 当数组的长度大于等于 64, 数组的某一项的长度大于等于 8, 会转为红黑树, 小于等于 6, 会重新转为链表
5. key 和 value 都不允许为 null (HashMap 允许一个 key 为 null 和 不限制的 value 为 null)
6. 在 ConcurrentHash 中数组中的每个节点的 hash 都有特殊作用

ConcurrentHash 中数组中的节点的 hash 值不同的含义

1. >= 0, 表明这个节点为 Node, 既链表节点
2. -1, 表明这个节点为 ForwardingNode, 说明这个位置正在数据迁移
3. -2, 表明这个节点为 TreeBin, 树节点 TreeNode 的进一步封装
4. -3, 表明这个节点为 ReservationNode, 通过 JDK 1.8 新增的几个方法给 ConcurrentHashMap 设值时, 会先对应位置的节点变为 ReservationNode, 再做处理

比如 JDK 1.8 中 Map 新提供了 computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) 方法, 这个方法的作用, 就是向 Map 获取 key 对应的值不存在时, 将后面的 lambda 表达式的返回值设置到 Map 中, 然后返回。
这样就能保证从 Map 不会获取到 null 值。

Map<Integer, String> map = new ConcurrentHashMap<>();
map.put(1, "1");// 输出: 2123
System.out.println(map.computeIfAbsent(2, k -> k + "123"));// 输出: null
System.out.println(map.get(10));

computeIfAbsent 定义在 Map 接口中声明的方法, 同时基于 JDK 1.8 的特性, 用 default 关键字进行了修饰, 提供了默认的实现。

Map 提供的这些 default 方法, 可以发现都是线程不安全的, 所以 ConcurrentHashMap 对他们进行了重写, 在这些方法中, 需要对向已有的数组里面的添加新值,
那么就会向把对应位置的那一项, 设置为 ReservationNode 节点。用于通知其他的线程, 这个位置的数据在修改中。

2.2 关键属性

public class ConcurrentHashMap<K,V> {// 装载 Node 的数组, 作为 ConcurrentHashMap 的数据容器, 采用懒加载的方式, 直到第一次插入数据的时候才会进行初始化操作, 数组的大小总是为 2 的幂次方transient volatile Node<K,V>[] table;// 扩容时使用, 平时为 null, 只有在扩容的时候才为非 null, 扩容时先把数据放在这个对象, 扩容完成后, 才把 table 指向这个对象transient volatile Node<K,V>[] nextTable;// 在没有发生争用时的元素个数的统计, 即当前的元素个数transient volatile long baseCount;// 数组大小控制// 这个属性非常重要, 取值不同有不同的情况// 1. 当 sizeCtl > 0 时// 在 ConcurrentHashMap 声明, 但是内部存储数据的数组是延迟加载的, 这个数组的大小临时存放到 sizeCtl 中// 初始化后表示扩容的阈值, 默认为当前数组的长度 * 0.75// 2. 当 sizeCtl = 0 时// 在声明 ConcurrentHashMap, 没有指定容量大小, 这时 sizeCtl 为 0, 也就是表明 数组 table 的长度去默认值: 16// 3. 当 sizeCtl = -1 时// 表明当前 table 正在初始中// 4. 当 sizeCtl < -1 时// sizeCtl 的数据类型为 int, 占 32 位。// 那么 sizeCtl 的高 16 位存放的是扩容时标识符(具体的解释可以看后面, 现在知道有这个设定就行了), 低 16 位存放的是参与扩容的线程数目 (CocurrentHashMap 在扩容的时候, 有别的线程发现正在扩容中, 会一起参与进来, 一起扩容)transient volatile int sizeCtl;// 扩容索引值, 表示已经分配给扩容线程的 table 数组索引位置, 主要用来协调多个线程间迁移任务的并发安全transient volatile int transferIndex;}

2.3 关键内部类

public class ConcurrentHashMap<K,V> {static class Node<K,V> implements Map.Entry<K,V> {final int hash;final K key;// 很多属性都是用 volatile 进行修饰的, 也就是为了保证内存可见性volatile V val;volatile Node<K,V> next;......}static final class TreeNode<K,V> extends Node<K,V> {TreeNode<K,V> parent;  // red-black tree linksTreeNode<K,V> left;TreeNode<K,V> right;TreeNode<K,V> prev;    // needed to unlink next upon deletionboolean red;......}static final class TreeBin<K,V> extends Node<K,V> {// 这个类并不负责包装用户的 key、value 信息, 而是包装的很多 TreeNode 节点。 // 实际的 ConcurrentHashMap "数组" 中, 存放的是 TreeBin 对象, 而不是 TreeNode 对象TreeNode<K,V> root;volatile TreeNode<K,V> first;volatile Thread waiter;volatile int lockState;// values for lockStatestatic final int WRITER = 1; // 持有写锁时的标志static final int WAITER = 2; // 等待写锁的标志static final int READER = 4; // 读锁的增加值......}static final class ForwardingNode<K,V> extends Node<K,V> {// 在扩容时才会出现的特殊节点, 作为一个标记节点放在桶的首位, // 其 key, value, next 全部为 null, hash 为 MOVED (-1), 并拥有 nextTable 指针指向新的 table 数组final Node<K,V>[] nextTable;ForwardingNode(Node<K,V>[] tab) {// hash, key, value, nextsuper(MOVED, null, null, null);this.nextTable = tab;}}static final class ReservationNode<K,V> extends Node<K,V> {// 调用 Map default 方法时, 会把节点修改为这个特殊节点// 其 key, value, next 全部为 null, hash 为 RESERVED (-3)ReservationNode() {super(RESERVED, null, null);}Node<K,V> find(int h, Object k) {return null;}}}

结合上面几个内部类, 基本可以推理出 ConcurrentHashMap 的存储数据的方式和 HashMap 一样, 都是数组, 数组的数据类型可以是链表, 也可以是红黑树。

大体的结构如下:

2.4 一些高频的 CAS 方法

public class ConcurrentHashMap<K,V> {/*** 该方法用来获取 tab 数组中索引为 i 的 Node 元素*/static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {return (Node<K,V>)U.getObjectAcquire(tab, ((long)i << ASHIFT) + ABASE);}/*** 利用 cas 操作将 tab 数组中索引为 i 的元素从 c 替换为 v*/static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) {return U.compareAndSetObject(tab, ((long)i << ASHIFT) + ABASE, c, v);}/*** 将 tab 数组中索引为 i 的元素设置为 v*/static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {U.putObjectRelease(tab, ((long)i << ASHIFT) + ABASE, v);}
}

3 ConcurrentHashMap 的源码实现

3.1 实例构造器方法

public class ConcurrentHashMap<K,V> {// 1. 构造一个空的 map, 即 table 数组还未初始化, 初始化放在第一次插入数据时, 默认大小为 16public ConcurrentHashMap(){}// 2. 给定 map 的大小public ConcurrentHashMap(int initialCapacity){}// 3. 给定一个 mappublic ConcurrentHashMap(Map<? extends K, ? extends V> m){}// 4. 给定 map 的大小以及负载因子public ConcurrentHashMap(int initialCapacity, float loadFactor){}// 5. 给定 map 大小, 加载因子以及最少多少个桶(数组的最小长度)public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel){}}

ConcurrentHashMap 一共给我们提供了 5 种构造器方法, 具体使用请看注释, 我们来看看第 2 种构造器, 传入指定大小时的情况 (其他的构造方式都是类似的), 该构造器源码为:

public class ConcurrentHashMap<K,V> {// 存储数据的数组的最大容量 private static final int MAXIMUM_CAPACITY = 1 << 30;public ConcurrentHashMap(int initialCapacity) {//1. 小于 0 直接抛异常if (initialCapacity < 0)throw new IllegalArgumentException();//2. 判断指定的容量是否超过了允许的最大值, 超过了话则取最大值, 否则再对该值进一步处理, 使其为 2 的 n 次方// 在 initialCapacity 的值大于等于 MAXIMUM_CAPACITY / 2 的情况下, 直接取最大值 MAXIMUM_CAPACITY, 否则重试计算出第一个大于 initialCapacity 的 2 的 n 次方的数// initialCapacity >>> 1 相当于除以2 取整, 经过这样处理可以得到第一个大于 initialCapacity 的 2 的 n 次方的数int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?  MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));    //3. 赋值给 sizeCtl, 这时候 sizeCtl 作用是存放初始时数组的容量this.sizeCtl = cap;}// 经过这个方法的处理后, 可以得到第一个大于等于 c 的 2 的 n 次方的数private static final int tableSizeFor(int c) {int n = c - 1;n |= n >>> 1;n |= n >>> 2;n |= n >>> 4;n |= n >>> 8;n |= n >>> 16;// MAXIMUM_CAPACITY == 1 << 30return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;}
}

这段代码的逻辑请看注释, 很容易理解。
如果小于 0 就直接抛出异常, 如果指定值大于了所允许的最大值的话就取最大值。
否则, 在对指定值做进一步处理。最后将计算出来的容量 cap 赋值给 sizeCtl, 关于 sizeCtl 的说明请看上面的说明, 当调用构造器方法之后, sizeCtl 的大小就代表了 ConcurrentHashMap 的大小, 即 table 数组长度。

在指定容量的处理方法 tableSizeFor 中, 根据入参的值计算出第一个大于当前入参的 2 的 n 次方数, 这个值就是 ConcurrentHashMap 中数组进行声明时的容量了。
比如, 当指定大小为 18 时, 经过这个方法处理, 会得到一个 32 的值。

需要注意的是, 调用构造器方法的时候并未构造出 table 数组 (可以理解为 ConcurrentHashMap 的数据容器) , 只是算出 table 数组的长度, 当第一次向 ConcurrentHashMap 插入数据的时候才真正的完成初始化创建 table 数组的工作。

3.2 ConcurrentHashMap 中数组的初始方法 initTable()

在上面的指定容量的构造函数中, 可以看到, 只是对入参的容量参数进行了处理, 然后赋值给 sizeCtl, 就结束了, 而真正存储数据的 table 数组还是为空的。
这是一种懒加载的方式, 而只要第一次向里面放数据时, 就会进行数组的初始化, initTable 就是初始化的方法。

public class ConcurrentHashMap<K,V> {private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;// table 为空 或者长度为 0, 进入循环, 否则进入后面的逻辑while ((tab = table) == null || tab.length == 0) {// 当前的 sizeCtl 小于 0, 表示 ConcurrentHashMap 正在扩容中if ((sc = sizeCtl) < 0)// 让当前线程让出行时间段, 使正在运行中的线程重新变成就绪状态, 后面重新竞争 CPU 的调度权// 确保当前只有一个线程在初始化Thread.yield();else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {// 通过 CAS 将 sizeCtl 设置为 -1, 成功了, 进行初始化// 在初始时, 会先将 sizeCtl CAS 为 -1, 也就是其他的线程进入到 while 里面, 也会在上面的判断时, 判断为 true 然后让出执行权try {// 再次判断 table 为 null 或者长度为 0 if ((tab = table) == null || tab.length == 0) {// 如果指定的容量小于 0, 使用默认值 16, 进行声明, 否则就用指定的大小进行声明int n = (sc > 0) ? sc : DEFAULT_CAPACITY;// 初始数组@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = tab = nt;// 计算数组中可用的大小: 实际大小 n * 0.75 (加载因子) // n >>> 2 相等于 n /2 /2 = n/4 = 0.25// sc = n - 0.25 * n = 0.75 * nsc = n - (n >>> 2);}} finally {// 这时候 sizeCtl 存放的是阈值, 不是数组的长度, 也不是 -1 了// 如果上面异常了, 可以把 sc 重新赋值过来, 也就是还是维护的是数组的长度sizeCtl = sc;}}    }}
}

代码的逻辑请见注释, 有可能存在一个情况是多个线程同时走到这个方法中, 为了保证能够正确初始化, 会在数组初始时做了一个判断 sizeCtl < 0 的判断,
若当前已经有一个线程正在初始化即 sizeCtl 值变为 -1, 这个时候其他线程在 if 判断为 true 从而调用 Thread.yield() 让出 CPU 时间片。
正在进行初始化的线程会调用 U.compareAndSwapInt 方法将 sizeCtl 改为 -1 即正在初始化的状态。

构造方法 + initTable 基本就构成了 ConcurrentHashMap 的初始的全部逻辑了。

3.3 ConcurrentHashMap 新增数据 - put() 方法

public class ConcurrentHashMap<K,V> {static final int spread(int h) {// ConcurrentHashMap 中 hash 值的计算方法// h 异或 (h 无符号右移 16 位) 后, 再与上 HASH_BITS// HASH_BITS = 0x7fffffff = 2147483647, 二进制表示为  01111111 11111111 11111111 11111111return (h ^ (h >>> 16)) & HASH_BITS;}public V put(K key, V value) {return putVal(key, value, false);}final V putVal(K key, V value, boolean onlyIfAbsent) {// 注意: 这里和 HashMap 的区别, HashMap 允许一个 key 为 0 和 多个 value 为 null// ConcurrentHashMap 则 key 和 value 都不允许, 直接抛出异常if (key == null || value == null) throw new NullPointerException();// 第一步, 计算 key 对应的 hash 值int hash = spread(key.hashCode());   // 用来计算对应位置的链表的长度int binCount = 0;// 注意: 这里做了一个死循环, 自旋操作for (Node<K,V>[] tab = table;;) {Node<K,V> f; int n, i, fh;// 第二步, 当前的数据结构为 null 或者 长度为 0, 进行初始化if (tab == null || (n = tab.length) == 0)// 继续数组的声明, 数组创建成功后, 回到循环的头部, 重新循环tab = initTable();// 第三步, 将当前 key 对应的 hash 值 & 上 数组的长度 - 1 (这一步的操作, 在数组的长度为 2 的 n 次方的情况下《等同于 hash % 数组的长度) // 得到当前 value 存放在数组的哪个位置, 并赋值给 i    // 通过 CAS 得到 i 位置的值, 如果为空, 进入判断里面的操作else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 把当前的数据封装为节点, 通过 cas 把这个节点放到 i 位置if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))// cas 操作成功, 跳出循环, 但是还是会执行到下面的 addCount 方法, 这时候 binCount = 0break;// 第四步, MOVED = -1; 对应位置的节点的 hash 为 -1, 既当前节点为 ForwardingNode 节点, 表示当前正在扩容, 当前线程进行协助扩容        } else if ((fh = f.hash) == MOVED) {// 协助扩容// 协助扩容后, 又会回到循环的头部, 重新处理tab = helpTransfer(tab, f);} else {// 当前 value 应该存放到位置已经有节点了V oldVal = null;// 如果指定的位置不为 null, 同时不是 ForwardingNode, 对这个对象上锁// 因为这里对这个节点上锁了, 所以扩容和添加新数据, 只有一个线程进入synchronized (f) {// 做多一次判断, 当前位置的节点还是指定的节点if (tabAt(tab, i) == f) {// 第五步, 当前为链表, 在链表中插入新的键值对, 或者存在 key 相同的, 进行值替换// fh = 定位到的位置的 hash, > 0 为 链表, -2 表示的是树节点if (fh >= 0) {// 链表当前有节点个数binCount = 1;// 循环遍历 链表for (Node<K,V> e = f;; ++binCount) {K ek;// 找到 hash 和 key 完成相同的节点, 覆盖旧值即可if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) {// 旧值oldVal = e.val;// 是否替换旧值, 这里是 !falseif (!onlyIfAbsent)e.val = value;// 这里旧值不为 null 并且 binCount >= 1, 走到下面的判断后, 不会走到 addCount 方法break;}Node<K,V> pred = e;// 如果当前节点的下一个节点为 null, 把当前数据封装为新节点放到链表的尾部, 跳出循环if ((e = e.next) == null) {pred.next = new Node<K,V>(hash, key, value, null);// 这时候旧值为 null, binCount = 链表的长度, 会走到 addCount 方法break;}}}// 第六步, 当前为红黑树节点的话, 将新的键值对插入到红黑树中else if (f instanceof TreeBin) {Node<K,V> p;// 直接把 binCount 设置为 2, 用于后面判断, 跳出循环binCount = 2;// 将节点转为 TreeBin 红黑树, 调用 TreeBin 的 putTreeVal 方法, 尝试将当前节点放到树内if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) {oldVal = p.val;if (!onlyIfAbsent)p.val = value;}}}                     }// 第七步, 插入完键值对后再根据实际大小看是否需要转换成红黑树if (binCount != 0) {// 链表的长度大于等于 8 了, 红黑树化if (binCount >= TREEIFY_THRESHOLD)// 在 treeifyBin 中会判断当前的数组长度, 如果长度小于 MIN_TREEIFY_CAPACITY = 64// 不会红黑树化, 而是直接扩容treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}    }// 第八步, 新增元素节点 1 个, 当前新增元素所在位置的节点个数 binCount 元素计数, // 在 addCount 中检查是否需要扩容addCount(1L, binCount);return null;}
}

从整体而言, 为了解决线程安全的问题, ConcurrentHashMap 使用了 synchronized 和 CAS 的方式。

ConcurrentHashMap 的数据存储方式和 HashMap 没有多大的区别。ConcurrentHashMap 是一个哈希桶数组, 如果不出现哈希冲突的时候, 每个元素均匀的分布在哈希桶数组中。
当出现哈希冲突的时候, 是标准的链地址的解决方式, 将 hash 值相同的节点构成链表的形式, 称为 “拉链法”, 另外, 在 1.8 版本中为了防止拉链过长, 当链表的长度大于
8 (同时还需要满足数组的长度达到了 64, 否则只会进行数组的扩容, 进行重新调整) 的时候会将链表转换成红黑树。table 数组中的每个元素实际上是单链表的头结点或者红黑树的根节点。

上面的新增数据的流程大体如下:

第一步: 计算 key 的 hash 值

当插入键值对时, 首先应该定位到要插入的桶, 即插入 table 数组的索引 i 处。那么, 怎样计算得出索引 i 呢? 在 ConcurrentHash 通过 spread 方法就能得到 key 对应的 hash 值。

spread() 方法 主要是将 key 的 hashCode 和 key 的高 16 位进行异或运算, 这样不仅能够使得 hash 值能够分散, 均匀减小 hash 冲突的概率, 另外只用到了异或运算,
在性能开销上也能兼顾, 做到权衡。

第二步: table 的初始化
判断当前的存放数据的 table 是否为空或者长度为 0, 如果是的话, 调用 initTable() 方法进行初始化, 方法的介绍可以看上面的。

第三步: 能否直接将新值插入到 table 数组中

通过公式 通过当前 key 计算出来的 hash 值 & (当前存储数据的数组的长度 - 1), 得到当前的键值对应该存储在数组的哪个位置, 这个计算公式在数组长度确保为 2 的 n 次方下,
相当于 hash 对数组的长度取模, 既 hash % n。

得到当前键值对待插入的位置, 再通过 tabAt() 方法获取该位置的节点, 如果节点为 null 的话, 就可以直接用 casTabAt() 方法将新值插入即可。

第四步: 当前是否正在扩容, 协助扩容
如果键值对待插入的位置的节点不为 null, 且该节点为特殊节点 (ForwardingNode) 的话, 就说明当前 ConcurrentHashMap 正在进行扩容操作。
怎样确定当前的这个 Node 是不是特殊的节点了? 是通过判断该节点的 hash 值是不是等于 -1(MOVED)。如果是说明当前正在扩容中, 则会协助扩容。

helpTransfer 协助扩容这里涉及到扩容的一些知识, 所以将其放到后面的扩容一起讲解。

第五步: 当前为链表, 在链表中插入新的键值对, 或者存在 key 相同的, 进行值替换
在 table[i] 不为 null 并且不为 ForwardingNode 时, 并且当前 Node f 的 hash 值大于 0 (fh >= 0) 的话说明当前节点 f 为当前桶的所有的节点组成的链表的头结点。
那么接下来, 要想向 ConcurrentHashMap 插入新值的话就是向这个链表插入新值就行了。通过 synchronized(f) 的方式进行加锁以实现线程安全性。
只对数组中这个位置的链表进行加锁, 而不是整个数组加锁, 提高了并发性。

第六步: 当 table[i] 为红黑树的根节点, 在红黑树中插入新值
按照之前的数组 + 链表的设计方案, 这里存在一个问题, 即使负载因子和 Hash 算法设计的再合理, 也免不了会出现拉链过长的情况, 一旦出现拉链过长, 甚至在极端情况下,
查找一个节点会出现时间复杂度为 O(n) 的情况, 则会严重影响 ConcurrentHashMap 的性能, 于是, 在 JDK 1.8 版本中, 对数据结构做了进一步的优化, 引入了红黑树。
而当链表长度太长 (默认超过 8) 时, 链表就转换为红黑树, 利用红黑树快速增删改查的特点提高 ConcurrentHashMap 的性能。

通过 f instanceof TreeBin 判断当前 table[i] 是否是树节点, 这下也正好验证了我们在最上面介绍时说的 TreeBin 会对 TreeNode 做进一步封装,
对红黑树进行操作的时候针对的是 TreeBin 而不是 TreeNode。
如果在红黑树中存在于待插入键值对的 Key 相同 (hash 值相等并且 equals 方法判断为 true) 的节点的话, 就覆盖旧值, 否则就向红黑树追加新节点

第七步: 插入完键值对后再根据实际大小看是否需要转换成红黑树
如果当前链表节点个数大于等于 8 (TREEIFY_THRESHOLD) 的时候, 就会调用 treeifyBin 方法将 tabel[i] (第 i 个散列桶) 拉链转换成红黑树。

第八步 添加元素计数, 并在 binCount 大于 0 时检查是否需要扩容
每次新增的时候都会对当前的数组的长度做一个检查, 超过了临界值, 就会进行扩容, 相关的方法 addCount。

同样的, addCount 涉及到了扩容方面的知识, 将其也放到扩容的时候一起讲。

整体的流程是这样的:

1. 首先对于每一个放入的值, 首先利用 spread 方法对 key 的 hashcode 进行一次 hash 计算, 由此来确定这个 key 应该存放在 table 中的位置
2. 如果当前 table 数组还未初始化, 先将 table 数组进行初始化操作
3. 如果要存放的位置是 null 的, 那么使用 cas 操作直接放入
4. 如果这个位置存在结点, 说明发生了 hash 碰撞, 首先判断这个节点的类型。如果该节点 fh == MOVED (代表 forwardingNode, 数组正在进行扩容) 的话, 说明正在进行扩容, 协助扩容, 扩容结束后, 重新判断
5. 如果是链表节点 (fh > 0), 则得到的结点就是 hash 值相同的节点组成的链表的头节点。需要依次向后遍历确定这个新加入的值所在位置。如果遇到 key 相同的节点, 则只需要覆盖该结点的 value 值即可。否则依次向后遍历, 直到链表尾插入这个结点。
6. 如果这个节点的类型是 TreeBin 的话, 直接调用红黑树的插入方法进行插入新的节点
7. 插入完节点之后再次检查链表长度, 如果长度大于 8, 就把这个链表转换成红黑树
8. 对当前容量大小进行检查, 判断是否需要扩容

3.4 ConcurrentHashMap 获取数据 - get() 方法

看完了 put 方法再来看 get 方法就很容易了, 用逆向思维去看就好, 怎么存, 反过来这么取就好了, get 方法的源码:

public class ConcurrentHashMap<K,V> {public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;// 得到 key 的 hash 值int h = spread(key.hashCode());// table 数组不为空同时长度大于 0, 计算得到的位置不为 nullif ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) {// 指定位置的 hash key 都相同, 直接在头节点找到了, 直接返回这个节点值if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}        // 节点的 hash 为 0 说明为树节点, 在红黑树中查找即可else if (eh < 0)return (p = e.find(h, key)) != null ? p.val : null;// 依次遍历链表的每一个节点while ((e = e.next) != null) {// hash 和 key 都相同, 返回这个节点 value if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek))))// 找到了直接返回return e.val;}    }return null;}
}

整理后的流程大体如下:

1. 计算出 key 的 hash
2. 通过 hash 得到对应位置的节点, 如果为 null, 直接返回
3. 判断对应位置的节点的 hash 和 key 是否一致, 如果是的话, 返回这个节点的值
4. 判断这个节点的 hash 小于 0, 说明这个为树节点, 在树中进行查找
5. 到了这一步, 说明这个节点为链表, 依次遍历这个链表的节点, 进行查找
6. 如果都没有的的, 直接返回 null

3.5 ConcurrentHashMap 的扩容

当 ConcurrentHashMap 容量不足的时候, 需要对 table 进行扩容。这个方法的基本思想跟 HashMap 是很像的, 但是由于它是支持并发扩容的, 所以要复杂的多。
原因是它支持多线程进行扩容操作, 而并没有加锁。这样做的目的不仅仅是为了满足 concurrent 的要求, 而是希望利用并发处理去减少扩容带来的时间影响。

3.5.1 扩容标识的获取 - resizeStamp() 方法

public class ConcurrentHashMap<K,V> {/*** 返回一个标志位用于调整 table 的大小* 这个数左移 RESIZE_STAMP_SHIFT, 也就是 16 位, 必须后能得到一个负数*/static final int resizeStamp(int n) {// Interger.numberOfLeadingZero 的作用: 返回整型 n 非零最高位前面 0 的个数// 比如: 13 的二进制为: 00000000 00000000 00000000 00001101 , 从右往左, 最后一个非零的前面有 28 个 0, 所以这里等于 28// RESIZE_STAMP_BITS = 16 , 所以 1 << (RESIZE_STAMP_BITS - 1) 固定为 32768, 二进制表达为 00000000 00000000 10000000 00000000// 这里的 |, 或操作, 2 位只要有一个是 1 就是 1 了, 在这里可以简单的看出是 2 个数相加return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));}
}

这个方法看起来就一行代码, 但是拆成三部分来看。

Integer.numberOfLeadingZeros(n)
这个方法的作用: 返回一个整形, 从左往右到第一个非 0 位的位数, 比如 13 的二进制为 00000000 00000000 00000000 00001101, 从左往右算到第一个 1 时, 总共有 28 位, 所以这个返回 28。

因为 int 为 32 位, 那么经过这个方法计算最终得到的结果最大为 32 (输入了 0), 最小当然是 0 (输入了 -1), 所以这个方法返回值在 0 - 32 之间。返回值的二进制为 00000000 00000000 00000000 00XXXXXX

1 << (RESIZE_STAMP_BITS - 1)

RESIZE_STAMP_BITS 是一个常量, 等于 16, 1 << (16 - 1) = 32768, 二进制为: 00000000 00000000 10000000 00000000, 刚好得到了一个第 16 位为 1 的数

| (或操作)
或运算, 这里可以看做是 2 个数相加, 将上面 2 步的结果进行或运算后, 可以得到一个二进制: 00000000 00000000 10000000 00XXXXXX

这个方法的作用是返回一个扩容标志符, 这个标志符经过左移 16 位后, 必须是一个负数。这个标志位主要用于后面的扩容。

通过这个方法得到的数字, 左移 16 位后, 为 10000000 00XXXXXX 00000000 00000000, 将这个数分成 2 部分。

高 16 位表示当前在扩容中, 而且保存了当前 table 的长度。
调用 resizeStamp() 方法, 传过来的 n = 当前 table 的长度, 而 table 的长度是 2^n 次方, 2^n 在二进制的形式为从 01 {n 个 0} 的形式, 结合 Interge.numberOfLeadingZeros 就可以知道当前 table 的长度了。
而低 16 位用在后面的记录当前参与扩容的线程数。每有一个线程参加了扩容就加 1, 所以最多参与扩容的线程个数就是 00000000 00000000 11111111 11111111 既 2^17 - 1 个线程。

先记住他的含义, 后面扩容的时候, 就能了解他的神奇之处。

3.5.2 扩容的判断 - addCount() 方法

在每次 put 新值的时候, 存放成功后, 都会执行一次是否需要扩容的检测, 如果需要就会调用扩容, 扩容的检测的方法就是 addCount() 方法

public class ConcurrentHashMap<K,V> {/*** 这个方法的作用: 更新当前的键值对的个数, 同时根据个数判断是否需要扩容* 不存在并发的情况下, 键值对的个数存放在 baseCount 下* 但是当出现了并发的时候, 键值对的个数的统计是通过数组 counterCells 每一项的 value 的累加** @param x  增加的个数* @param check 检查模式,   check < 0 不进行扩容的的检测, check >= 0 进行扩容检测, 但是出现了并发时,  0<= check <= 1, 不进行扩容检测*/private final void addCount(long x, int check) {// CounterCell 是 ConcurrentHashMap 中的一个内部类// 整个类只有一个 long value 的属性// 主要用于并发时, ConcurrentHashMap 的数据个数的计算CounterCell[] as; long b, s;if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {// 1. 用于统计数组个数的 counterCells 数组已经不为 null 了// 2. 通过 cas 将当前直接存储 ConcurrentHashMap 元素个数的 bseCount + x, 失败 (baseCount 存放的是当前 ConcurrentHashMap 中存放的数据个数)// 这 2 个情景都可以确定当前的增加元素个数进入了并发情景了CounterCell a; long v; int m;// 线程争用的状态标记, 默认为无竞争boolean uncontended = true;// ThreadLocalRandom.getProbe() 随机产生一个 int 的数字, // 下面的 m = ConcurrentHashMap 的 CounterCell[] counterCells 数组的长度, 默认声明是为 2, 后面扩容也是在原来的基础上 * 2, 所以 counterCell 的长度也是 2 的 n 次方// 所以通过 ThreadLocalRandom.getProbe() & m, 就是从  CounterCell[] counterCells 中随机去一个位置if (as == null || (m = as.length - 1) < 0 || (a = cs[ThreadLocalRandom.getProbe() & m]) == null || !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)))) {// 1. ConcurrentHashMap 的 CounterCell[] counterCells 为空或者长度小于 0// 2. 从 CounterCell[] counterCells 随机取出一个位置为 null// 3. 从 CounterCell[] counterCells 随机取出一个位置不为 null, 通过 cas 将这个位置的 CounterCell 的 value + x 失败了// 这里只有是给当前的 ConcurrentHashMap 的元素个数加上 x, 通过 fullAddCount() 方法实现的// 这个方法的复杂度很高, 就不展开了, 说一下大体的功能// 先通过 ThreadLocalRandom.getProbe() 当前线程的用于产生随机数的随机种子// 同一个线程一直调用这个方法会返回通过一个值, 调用 advanceProbe() 方法进行刷新, 再次调用 getProbe() 可以得到另一个新的随机种子// 下面的整个操作的过程都是自旋的// 1. 当前的 counterCells 有数据// 1.1 通过获取到随机种子, 算出在 counterCells 中的一个位置// 1.2 获取到的位置不为 null, 通过 cas 给这个位置的 CounterCell 的 value + x// 1.2.1 cas 设置成功了, 结束整个方法// 1.2.2 cas 设置失败了, 给 counterCells 进行扩容, 变为原来的 2 倍, 调用 ThreadLocalRandom.advanceProbe() 获取新的随机种子, 回到自旋循环的开头// 1.3 获取到的位置为 null, 创建一个新的 CounterCell, value 为 x// 1.3.1 在判断这个位置为 null, 将新的 CounterCell 放到这个位置, 结束// 1.3.3 在判断这个位置不为 null, 回到自旋循环的开头// 2. counterCells 没有数据, 通过 cas 将 baseCount 加 x// 在 fullAddCount 中实际的实现比上面的复杂, 会通过 cas cellsBusy 从 0 设置为 1, 设置成功, 才能进行上面的操作, 设置失败都只能自旋fullAddCount(x, uncontended);// 注意这里的 return 有个问题, 通过 fullAddCount() 给当前的 ConcurrentHashMap 的元素个数 + 1, 为什么不检测是否需要扩容呢?// 这里其实需要从多线程的角度考虑了, 能够进到这里, 有哪些情况? 结合上面 2 个 if// 1. counterCells 不为 null,  并且通过 cas 更新 counterCells 的计算出来的位置的值失败, 那么就是有另一个线程更新了这个值, 扩容的检测交给那个线程就行了// 2. counterCells 为 null, 通过 cas 更新 baseCount 失败了, 进到这里很大概率 as 为空, 同样的是交给另一个线程进行扩容的检测return;}// 增加元素个数进入了并发情景了// 不需要进行扩容的检测if (check <= 1)return;// 统计总个数// baseCount + CounterCell[] counterCells 中每个不为 null 的元素的 value  s = sumCount();}// 进行扩容的检测if (check >= 0) {Node<K,V>[] tab, nt; int n, sc;// 当前的元素个数 > 阈值 并且 table 不为空, 同时 table 的长度小于最大容量 2^30, 还有扩容空间while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) {//根据数组长度获取一个扩容标志, 格式 16 个 0 + 1 + 15 位 (表示当前 table 容量的二进制表示从左往右 0 的个数)int rs = resizeStamp(n);// 在上面我们说过了 sizeCtl 的不同取值有不同的含义, // 小于 0, 初始中或者有线程在扩容中, 但是在 addCount 方法可以断定为扩容中if (sc < 0) {// 1. sc 无符号右移 16 位和扩容标识 rs 不相同, 将不做处理, // 出现不同的原因就是 table 的长度改变了, 导致计算出来的 rs 变了 也就是上次的扩容完成了, 不进行处理// 当第一个线程进行扩容的时候, 会把 sizeCtl 设置为计算出来的 rs 右移 16 位后 + 2, // 所以当改变后的 sizeCtl 左移 16 位和当前的 rs 不一致了, 说明 table 的长度已经变了// 如果 rs 从 + 1, 是无法区分 sizeCtl 低 16 位都是 0 时, 是扩容为开始还是扩容已经结束// 如果直接 + 2, 后续减少时 -1, 那么低 16 位都是 0 表示扩容未开始, 低 16 位为 1, 扩容结束// 2. sc == rs + 1 这里是一个 bug, 应该改为 sc == (rs <<< RESIZE_STAMP_SHIFT) + 1// sc 在经过条件 1 后, 值是不会变的, 那么 sc 依旧是小于 0, 而 rs 在我们分析后必定是一个大于 0 的数, 所以 sc 一定不会等于 rs + 1, 应该是 sc == (rs <<< RESIZE_STAMP_SHIFT) + 1// 这个 bug 可以查看 这里 https://bugs.java.com/bugdatabase/view_bug.do?bug_id=JDK-8214427, 预计在 JDK 12 修复, 在 14 的时候已经看到修复了// 如果修改为上面的, 我们知道 sc 的低位存放的是当前线程参与扩容的线程数, 默认是从 2 开始累积的, 在扩容 transfer 中,// 每有一个线程扩容完成时会 - 1, 当 sc 的低位减到了 1 时, 说明这时候扩容已经完成了, 而 rs 无符号左移 16 位后, 低 16 位都是 0, // 如果 sc 的低位 = 右移后的 rs 的低位 + 1, 说明扩容完成了, 正在进行末尾的处理了, 不进行处理// 3. 同上面的条件 2 一样, 应该修改为 sc == (rs <<< RESIZE_STAMP_SHIFT) + MAX_RESIZERS, 这里的 MAX_RESIZERS = 1 << 16 - 1; // 也就是 00000000 00000000 11111111 11111111, 也就是我们扩容能支持的最大线程数, 这里也就是我们当前扩容的线程数达到了上限, 不进行处理// 4. 在扩容的时候, 我们会先声明 nextTable, 逐渐将数据放到这个对象, 如果这个对象为 null, 说明没有线程在进行扩容, 不进行处理, 在上面的 size < 0, 我们可以得知线程进来是有线程在扩容, 现在 nextTable 为空, 说明扩容完成了// 5. 转移索引标识 transferIndex <= 0, 不进行处理, 这个变量用于扩容时控制, 可以查看下面的扩容方法, <= 0 说明我们的扩容已经完成了if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0)break;// 在 sizeCtl 添加多一个线程, 开始协助扩容, 这时候扩容的新数组就是 nextTableif (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))transfer(tab, nt);    } // 尝试将 sizeCtl 设置为 扩容标志 rs 左移 16 位 + 2, 2 表示线程数默认是从 2 开始累积的else if (U.compareAndSetInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2))// cas 将 sizeCtl 设置成功, 开始扩容//  进行扩容, 存放数据的新数组为 null,transfer(tab, null);// 重新计算元素个数s = sumCount();}}}
}

总的来说: addCount 就做了 2 件事

1. 更新当前的元素个数
2. 检查是否需要扩容或者协助扩容

3.5.3 扩容 - transfer() 方法

public class ConcurrentHashMap<K,V> {/*** 扩容* 大体的步骤：* 通过当前机器的 cpu 和 当前数组的容量算出一个迁移跨度  stride* 每次一个线程进来扩容的话，就会分配当前数组的 stride 个位置给它, 另一线程进来或者当前线程 stride 个位置已经迁移完了, * 就再分配另外 stride 个位置给他, 直到当前数组所有的位置都分配完成* ConcurrentHashMap 的  transferIndex 扩容时默认为当前数组的容量, 类似于一个游标, 每次分配是从这个游标开始往前 stride 个位置就是分配个* 这个线程的区间了。* 里面声明的 2 个变量 i 和 boud, 就是当前线程可以处理的区间的位置 [bound, i], [transferIndex - 1 - stride, transferIndex - 1]* 同时把 transferIndex 更新为 transferIndex - stride* * 遍历中的操作* 1. 获取到的位置为空, 填充为 ForwardingNode 节点， 处理下一个节点* 2. 获取到的位置的节点为 ForwardingNode, 跳过处理下一个节点* 3. 判断当前位置的节点是否为链表或者树* 3.1 链表, 新找到尾节点, 尾节点的 hash & 当前数组的容量 == 0 ? 放到低纬链表的头部, 放到高纬链表的头部, 使用同 HashMap 的高低位链表的方式进行迁移* 3.2 遍历当前位置的链表直到倒数第二个, 处理的节点的 hash & 数组的长度 = 0? 应该放到低纬, 应该放到高纬度* 3.3 把当前的节点封装为新的节点, 这个节点的下一个节点为当前的高纬/低纬节点, 然后把当前节点放到低纬/高纬的位置* 3.4 最后把临时数组 nextTab 的 i 位置设置为当前的低纬链表， i + 当前数组的容量的位置设置为当期的高纬链表* 3.5 把当前数组的 i 位置设置为 ForwardingNode 节点, 下一个遍历** 迁移完成* 把临时数据 nextTable 赋值给真正的存储数组 table* 把 nextTable 设置为 null * 把 sizeCtl 设置为 旧数组容量的 * 1.5 * * 没法迁移了, 就给当前的 sizeCtl 减  1** @param tab 扩容前的数组* @param nextTab 新的数组, 扩容的话, 这里为 null, 协助扩容的为, 为新的数组*/private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {// 这里的 tab 就是成员变量 tableint n = tab.length, stride;// MIN_TRANSFER_STRIDE = 16   NCPU = 当前服务器的 CPU 核数// n 为数组的长度, 默认为 2 的 n 次方, 所以这里的 >>> 3, 相当于除以 8// 当前为多核服务器的话 stride = 当前的容量 / 8 / CPU 的数量// 如果为单核的话的话, stride = 当前数组容量// 最终计算出来的 stride 如果小于 16, 将其变为 16if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE;// 如果用于存数据的下一个数组为 null, 进行初始, 然后把这个数组赋值给变量 nextTable,if (nextTab == null) {try {// 扩容为原来的 2 倍@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];nextTab = nt;} catch (Throwable ex) {// 扩容失败, 直接将阈值变为 int 的最大值, 然后返回sizeCtl = Integer.MAX_VALUE;return;}// 将创建出来的扩容数组赋给给 nextTable, 可以让其他线程看见nextTable = nextTab;// 扩容索引变为扩容前数组的长度transferIndex = n;}   // 记录过渡数组的长度int nextn = nextTab.length; // 创建一个 ForwardingNode 节点, 用于占位。当别的线程发现这个槽位中是 ForwardingNode 类型的节点, 则跳过这个节点ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);// 该变量控制迁移的进行 true 说明可以再次迁移一个下标 (i--) , // 反之, 如果是 false, 那么就不能推进下标, 需要将当前的下标处理完毕才能继续推进boolean advance = true;// 完成状态, 如果是 true, 说明迁移完成, 可以结束此方法boolean finishing = false;// 下面的循环主要是// 1. 为了计算出当前线程需要处理的区间 [transferIndex - stride, transferIndex], 即 [bound (transferIndex - stride), i(transferIndex - 1)], // 同时更新 transferIndex = transferIndex - stride, 方便后续的线程或者线程重新获取的迁移的区间// 2. 遍历中, 负责进行 i - 1 操作, 然后决定是继续遍历还是进行下一次分配// i 表示处理的当前桶区间最大下标, bound 表示当前线程可以处理的当前桶区间最小下标for (int i = 0, bound = 0;;) {Node<K,V> f; int fh;// 当前线程是否可以向后推进, 这个循环就是控制 i 递减, 同时, 每个线程都会进入这里取得自己需要转移的桶的区间while (advance) {int nextIndex, nextBound;// 对 i 减一, 判断是否大于等于 bound (正常情况下, 如果大于等于 bound 不成立, 说明该线程上次领取的任务已经完成了那么, 需要在下面继续领取任务)// 如果对 i 减一, 大于等于 bound (还需要继续做任务) , 或者迁移完成了, 修改推进状态为 false, 不能推进了。 每迁移完成一个位置, 推进状态会被修改为 true// 通常, 第一次进入循环, i-- 这个判断会无法通过, 从而走下面的 nextIndex 赋值操作 (获取最新的转移下标) //其余情况都是: 如果可以推进, 将 i 减一, 然后修改成不可推进。如果 i 对应的位置处理成功了, 又把推进状态修改成可以推进。if (--i >= bound || finishing)advance = false;else if ((nextIndex = transferIndex) <= 0) {// 如果 transferIndex 小于等于 0, 说明没有区间了, i 改成 -1, 推进状态变成 false, 不再推进, 表示, 扩容结束了, 当前线程可以退出了i = -1;advance = false;} else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { // 通过 cas 减少 transferIndex (transferIndex - stride 跨度) 的值成功了, 为当前线程重新分配新的扩容区间// 这里的目的是: // 1. 当一个线程进入时, 使其可以获取最新的转移下标// 2. 当一个线程处理完自己的区间时, 如果还有剩余区间的没有别的线程处理, 再次获取区间// 这个值就是当前线程可以处理的最小当前区间最小下标bound = nextBound;// 初次对 i 赋值, 这个就是当前线程可以处理的当前区间的最大下标i = nextIndex - 1;advance = false;}}// i < 0 (当 transferIndex <= 0 时, 数组已经分配完成了, 线程进入到上面的 while, 使得 i = -1, 然后走到这个方法, 让线程结束扩容 )// i >= 旧数组的长度// i + n >= 新数组的长度if (i < 0 || i >= n || i + n >= nextn) {int sc;// 如果完成了扩容if (finishing) {// 成员变量设置为 nullnextTable = null;// 将迁移完成的数组赋值给 tabletable = nextTab;// sizeCtl 变为 1.5 倍 (2n - 0.5n)sizeCtl = (n << 1) - (n >>> 1);return;}// 通过 cas 设置 sizeCtl 的扩容线程数减 1if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// 一开始进入扩容时 sc + 2, 这里 - 2 等于扩容标识, 说明当前扩容完成了, 只剩一个线程了// 让这个线程进行扩容完成后的处理if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)return;// 更新结束标志为 true, 推进标志为 truefinishing = advance = true;// 将 i 重新设置为 n, 让上面的 --i >= bound 为 true, 可以进入后面 finish 的判断i = n;}  }                  // 扩容时发现负责的区域有空的桶直接使用 ForwardingNode 填充else if ((f = tabAt(tab, i)) == null)advance = casTabAt(tab, i, null, fwd);// 对应位置的节点为 -1 (MOVED) 已经处理了, 继续下一个位置的处理   else if ((fh = f.hash) == MOVED)advance = true;else {    // 对 table 的 i 位置的节点上锁, 防止 putVal 的时候向链表插入新的数据synchronized (f) {// 做一次判断, 确保对应位置的对象为需要处理的对象if (tabAt(tab, i) == f) {// 此处扩容和 HashMap 有点像, 在原数组的 pos 位置的节点, 在扩容后, 会在新数组的 pos 位置 或者 pos + 原数组长度的位置// 前者为低纬  lowNode, 后者为高纬 highNodeNode<K,V> ln, hn;// 如果 f 的 hash 值大于 0, 表明这个节点为链表if (fh >= 0) {// 节点的 hash  & 旧数组的长度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;// hash 和 runBit 不一样了, 进行替换if (b != runBit) {runBit = b;lastRun = p;}}// 尾结点的 hash 处理后为 0, 将尾节点放在低纬链表if (runBit == 0) {ln = lastRun;hn = null;}// 尾结点的 hash 处理后不为 0, 将尾结点放在高纬链表else {hn = lastRun;ln = null;}// 没有上面的操作, 直接遍历链表也完成功能, 只是上面的都是做了一些优化// 比如 原来数组的链表为 1(低纬) - 2(高纬) - 3(高纬) - 4(低纬) - 5(低纬) - 6(低纬) - 7(低纬)// 那么经过上面的处理, lastRun = 4(低纬)// ln = 4(低纬) - 5(低纬) - 6(低纬) - 7(低纬)// 因为 lastRun 后面的节点必定和 lastRun 在同一个位置, 那么我们在迁移时, 只需要遍历到 lastRun 这个位置就行了// 后面的都是位置和 lastRun 一样, 直接把 lastRun 移过来就行了, 后面的节点都在 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;// 可以看出这里是头插法if ((ph & n) == 0)ln = new Node<K,V>(ph, pk, pv, ln);elsehn = new Node<K,V>(ph, pk, pv, hn);}// 在 nextTable 的 i 位置上插入一个链表setTabAt(nextTab, i, ln);// 在 nextTable 的 i + 旧数组长度的位置上插入一个链表setTabAt(nextTab, i + n, hn);// 在 table 的 i 位置上插入 forwardNode 节点setTabAt(tab, i, fwd);// 设置推进标志为 trueadvance = 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);// 和链表相同的判断, 与运算 == 0 的放在低位if ((h & n) == 0) {if ((p.prev = loTail) == null)lo = p;elseloTail.next = p;loTail = p;++lc;} else {// 不是 0 的放在高位if ((p.prev = hiTail) == null)hi = p;elsehiTail.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;}}}}}} 
}

第一部分 构建一个 nextTable, 它的容量是原来的两倍, 这个操作是单线程完成的。新建 table 数组的代码为: Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1], 在原容量大小的基础上右移一位

第二部分就是将原来 table 中的元素复制到 nextTable 中, 主要是遍历复制的过程。 根据运算得到当前遍历的数组的位置 i, 然后利用 tabAt 方法获得i位置的元素再进行判断

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

3.5.4 协助扩容 - helpTransfer() 方法

当 ConcurrentHashMap 正在扩容中, 有其他的线程计划向里面添加数据时, 因为正在扩容中, 无法添加。
但是 ConcurrentHashMap 的设计是不会让这个线程阻塞, 而是让这个线程帮忙扩容, 既执行 helpTransfer() 方法。

public class ConcurrentHashMap<K,V> {/*** @param tab 扩容的数组, 这里看成存放数据的 table 就对了* @param f 当前处于扩容中的节点*/final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {Node<K,V>[] nextTab; int sc;// 1. 当前的 tab(也就是 table) 不为 null,// 2. 当前的节点为 ForwardingNode// 3. 当前 ForwardingNode 中用于临时存放扩容时, 数据的 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) {// 这里的逻辑和 addCount 里面的差不多, 就不做解释了if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0)break;// 通过 CAS 操作, 设置当前的扩容线程 + 1, if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {// 设置成功了, 进入协助扩容transfer(tab, nextTab);break;}       }return nextTab;}return table;}
}

4 ConcurrentHashMap 与 size 相关的一些方法

对于 ConcurrentHashMap 来说, 这个 table 里到底装了多少东西其实是个不确定的数量, 因为不可能在调用 size() 方法的时候像 GC 的 “stop the world” 一样让其他线程都停下来让你去统计,
因此只能说这个数量是个估计值。对于这个估计值, ConcurrentHashMap 也是大费周章才计算出来的。

为了统计元素个数, ConcurrentHashMap 定义了一些变量和一个内部类

public class ConcurrentHashMap<K,V> {/***  Contended 注解可以用来解决伪共享*/@sun.misc.Contended static final class CounterCell {volatile long value;CounterCell(long x) { value = x; }}/** 当出现了并发的时候, 整个 ConcurrentHashMap 的元素个数, 是分散存在这个数组内*/private transient volatile CounterCell[] counterCells;/** 创建 CounterCells 是充当自旋锁使用 */private transient volatile int cellsBusy;/** 实际保存的是 Map 中的元素个数  利用 CAS 锁进行更新, 但是当出现了并发情况, 元素的个数就不从这个进行获取了*/private transient volatile long baseCount;
}

size() 方法

public class ConcurrentHashMap<K,V> {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;/** counterCells 不为 null, 就是出现了并发情况, 这个对象被声明了 */if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += a.value;//所有counter的值求和}}return sum;}/*** mappingCount 与 size 方法的类似 * 官方的给出的注释来看, 应该使用 mappingCount 代替 size 方法*/public long mappingCount() {long n = sumCount();return (n < 0L) ? 0L : n; }
}

5 总结

JDK 1.6, 1.7 中的 ConcurrentHashmap 主要使用 Segment 来实现减小锁粒度。 分割成若干个 Segment, 在 put 的时候需要锁住 Segment, get 时候不加锁, 使用 volatile 来保证可见性。
当要统计全局时 (比如 size) , 首先会尝试多次计算 modcount 来确定, 这几次尝试中, 是否有其他线程进行了修改操作, 如果没有, 则直接返回 size。如果有, 则需要依次锁住所有的 Segment 来计算。

1.8 之前 put 定位节点时要先定位到具体的 segment, 然后再在 segment 中定位到具体的桶。而在 1.8 的时候摒弃了 segment 臃肿的设计, 直接针对的是 Node[] table 数组中的每一个桶, 进一步减小了锁粒度。
并且防止拉链过长导致性能下降, 当链表长度大于 8 的时候采用红黑树的设计。

主要设计上的变化有以下几点:

1. 放弃采用 segment 而采用 node, 锁住 node 来实现减小锁粒度
2. 设计了 MOVED 状态, 当 resize 的过程中, 线程 2 在 put 数据, 线程 2 会帮助 resize
3. 使用 3 个 CAS 操作来确保 node 的一些操作的原子性, 这种方式代替了锁。
4. sizeCtl 的不同值来代表不同含义, 起到了控制的作用
5. 采用 synchronized 而不是 ReentrantLock

更多关于 1.7 版本与 1.8 版本的 ConcurrentHashMap 的实现对比, 可以参考这篇文章。

6 参考

ConcurrentHashmap简介
ConcurrentHashMap源码阅读
How does ConcurrentHashMap resizeStamp method work