Java 并发编程（十一）： Java 8 基于CAS 的ConcurrentHashMap 的实现原理
1、ConcurrentHashMap 简介
ConcurrentHashMap 是一个支持完全并发检索和更新的哈希表。该类遵循了与 java.util.Hashtable 相同的功能规范，与 Hashtable 的每个方法都有相对应的版本。它的所有操作都是线程安全的，检索操作也不需要锁定，但是不支持锁定整个表。在依赖于线程安全但不依赖于其同步细节的程序中，此类可与Hashtable完全互操作。检索操作（包括get）通常不会阻塞，因此可能与更新操作（包括put和remove）重叠。
2、ConcurrentHashMap 出现的原因
在并发编程中使用HashMap可能导致程序死循环。而使用线程安全的HashTable效率又非常低下，基于以上两个原因，ConcurrentHashMap 也就诞生了。
（1）非线程安全的HashMap在多线程环境下，使用HashMap进行put操作会引起死循环，导致CPU利用率接近100%，所以在并发情况下不能使用HashMap。例如，以下代码就会引起死循环。
public static void main(String[] args) throws InterruptedException {
final HashMap<String, String> map = new HashMap<String, String>(2);
Thread t = new Thread(() -> {
for (int i = 0; i < 10000; i++) {
new Thread(() -> map.put(UUID.randomUUID().toString(), ""), "child_thread_" + i).start();
}
}, "parent_thread");
t.start();
t.join();
}
HashMap在并发执行put操作时会引起死循环，是因为多线程会导致HashMap的Entry链表形成环形数据结构，一旦形成环形数据结构，Entry的next节点永远不为空，就会产生死循环获取Entry。
（2）效率低下的HashTable
HashTable 容器使用synchronized来保证线程安全，在线程竞争激烈的情况下 HashTable 的效率非常低下。因为当一个线程访问 HashTable 的同步方法，其他线程也访问HashTable的同步方法时，会进入阻塞或轮询状态。如线程1使用put进行元素添加，线程2不但不能使用put方法添加元素，也不能使用get方法来获取元素，所以竞争越激烈效率越低。
（3）ConcurrentHashMap 用分离锁实现多个线程间的并发写操作
在 ConcurrentHashMap 中，线程对映射表做读操作时，一般情况下不需要加锁就可以完成，对容器做结构性修改的操作才需要加锁。
（4）用 Volatile 变量协调读写线程间的内存可见性
由于内存可见性问题，未正确同步的情况下，写线程写入的值可能并不为后续的读线程可见。
根据 Java 内存模型，对 同一个 volatile 变量的写 / 读操作可以确保：写线程写入的值，能够被之后未加锁的读线程“看到”。
3、ConcurrentHashMap 数据结构
Java 8为进一步提高并发性，摒弃了分段锁（Segment）的方案，而是启用了一种全新的方式实现,利用CAS算法。它沿用了 HashMap 的思想，底层由 数组+链表+红黑树 的方式实现，但是为了做到并发，又增加了很多辅助的类。
同时为了提高哈希碰撞下的寻址性能，Java 8在链表长度超过一定阈值（8）时将链表（寻址时间复杂度为O(N)）转换为红黑树（寻址时间复杂度为O(log(N))）
将原先table数组＋单向链表的数据结构，变更为table数组＋单向链表＋红黑树的结构。对于hash表来说，最核心的能力在于将key hash之后能均匀的分布在数组中。如果hash之后散列的很均匀，那么table数组中的每个队列长度主要为0或者1。但实际情况并非总是如此理想，虽然ConcurrentHashMap类默认的加载因子为0.75，但是在数据量过大或者运气不佳的情况下，还是会存在一些队列长度过长的情况，如果还是采用单向列表方式，那么查询某个节点的时间复杂度为O(n)；因此，对于个数超过8(默认值)的列表，jdk1.8中采用了红黑树的结构，那么查询的时间复杂度可以降低到O(logN)，大大提高了检索效率。
4、ConcurrentHashMap 元素定位方式
Java 8 的 ConcurrentHashMap 同样是通过Key的哈希值与数组长度取模确定该Key在数组中的索引。为了避免不太好的Key的hashCode设计，它通过如下方法计算得到Key的最终哈希值。不同的是，Java 8的ConcurrentHashMap作者认为引入红黑树后，即使哈希冲突比较严重，寻址效率也足够高，所以作者并未在哈希值的计算上做过多设计，只是将Key的hashCode值与其高16位作异或并保证最高位为0。
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
5、ConcurrentHashMap 同步方式
对于put操作，如果Key对应的数组元素为null，则通过CAS操作将其设置为当前值。如果Key对应的数组元素（链表表头或树的根元素）不为null，则对该元素使用synchronized关键字申请锁，然后进行操作。如果该put操作使得当前链表长度超过一定阈值，则将该链表转换为树，从而提高寻址效率。
对于读操作，由于数组被volatile关键字修饰，因此不用担心数组的可见性问题。同时每个元素是一个Node实例（Java 7中每个元素是一个HashEntry），它的Key值和hash值都由final修饰，不可修改，无需关心它们被修改后的可见性问题。而其Value及对下一个元素的引用由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(int hash, K key, V val, Node<K,V> next) {
this.hash = hash;
this.key = key;
this.val = val;
this.next = next;
}
--------------------省略---------------------
}
6、ConcurrentHashMap 操作方法
6.1、Get 操作
总的来说 get 操作是最简单的，除非正好遇到了扩容 ForwardingNode.find(…)
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//计算hash值
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//判断头节点是否是我们需要的节点
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) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
1、根据 key 计算 hash 值
2、根据 hash 值找到数组对应位置: (n – 1) & h
3、判断该位置是链表还是红黑树
3.1、如果该位置为 null，那么直接返回 null
3.2、如果该位置的节点刚好就是我们需要的，返回该节点的值
3.3、如果该位置节点的 hash 值小于 0，说明正在扩容，或者是红黑树
3.4、如果以上都不满足，那就是链表，进行遍历查找
6.2、PUT 操作
public V put(K key, V value) {
return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
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;
//数据 tab 为空，初始化数组
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//根据 hash 值，找到对应数组下标，并获取对应位置 的第一个节点 f
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 如果数组该位置为空，用 CAS 操作将这个新值放入这个位置
// 如果 CAS 失败，那就是有并发操作，继续下一个循环
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//f 的 hash 值等于 MOVED，说明在扩容
else if ((fh = f.hash) == MOVED)
//进行数据迁移
tab = helpTransfer(tab, f);
else {
//f 是该位置的首节点，而且不为空
V oldVal = null;
// 获取数组该位置的首节点的监视器锁
synchronized (f) {
if (tabAt(tab, i) == f) {
// 首节点 hash 值大于 0，说明是链表
if (fh >= 0) {
//记录链表长度
binCount = 1;
//链表遍历
for (Node<K,V> e = f;; ++binCount) {
K ek;
//// 如果找到了相等的 key，判断是否要进行值覆盖
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;
//到了链表的尾部，将新值放到链表的尾部
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
//如果当前位置是红黑树
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
//红黑树插入新节点
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
//如果链表长度不为0
if (binCount != 0) {
//如果链表长度大于等于临界值【8】，将链表转换为红黑树
if (binCount >= TREEIFY_THRESHOLD)
//将链表转换成红黑树前判断，如果当前数组的长度小于 64，那么会选择进行数组扩容，而不是转换为红黑树
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
put 操作流程梳理：
1、根据 key 计算 hash 值
2、遍历数组 tab
2.1、如果数组为空，进行数组初始化 initTable()
2.2、如果数组不为空，根据 hash 值找到在数组 tab 中的位置，并获取该位置的第一个节点元素对象
2.2.1、如果数组该位置为空，使用 CAS 操作将值放入该位置
2.2.2、如果数组该位置的首个元素的hash 值等于 MOVED ，说明在扩容，帮助数据迁移
helpTransfer
2.3、如果不满足以上（2.1、2.2）条件，执行以下逻辑：
2.3.1、获取数组该位置的首节点的监视器锁
2.3.2、首节点 hash 值大于 0，说明是链表，链表遍历插入值
2.3.3、否则为红黑树，使用红黑树的 putTreeVal 方法，插入新值
2.3.4、如果以上执行的是链表操作，判断链表长度是否达到限定值，是否需要转换为红黑树
6.2.1、initTable 初始化数组
主要就是初始化一个合适大小的数组，然后设置 sizeCtl，初始化方法中的并发问题是通过对 sizeCtl 进行一个 CAS 操作来控制的。
/**
* Initializes table, using the size recorded in sizeCtl.
*/
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//sizeCtl 的值小于 0 ，说明正在初始化
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
//CAS 操作，并将 sizeCtl 设置为 -1，说明抢到了锁
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
// DEFAULT_CAPACITY 默认初始容量是 16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
// 初始化数组，长度为 16 或指定的长度
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
//将数组赋值给 table，它是 volatile 的
table = tab = nt;
// 如果 n 为 16 的话，那么这里 sc = 12，其实就是 0.75 * n
// 根据负载因子计算负载值（0.75*n）,n = 16，这里 sc = 12
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
6.2.2、treeifyBin 链表转红黑树
treeifyBin 不一定就会进行红黑树转换，也可能是仅仅做数组扩容，转换之前会首先判断链表的长度，如果链表长度小于限定值 64 则进行数组扩容，否则进行链表转换红黑树操作
/**
* Replaces all linked nodes in bin at given index unless table is
* too small, in which case resizes instead.
*/
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
// MIN_TREEIFY_CAPACITY 为 64
// 如果数组长度小于 64，进行数组扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1);
//否则链表转换为红黑树，b 是链表首元素
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
//对链表首元素加锁
synchronized (b) {
//遍历链表建立一颗红黑树
if (tabAt(tab, index) == b) {
TreeNode<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;
else
tl.next = p;
tl = p;
}
//将红黑树放在数组的相应位置
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
6.2.3、tryPresize 数组扩容
数组扩容后是原始大小的两倍
/**
* Tries to presize table to accommodate the given number of elements.
*
* @param size number of elements (doesn't need to be perfectly accurate)
*/
private final void tryPresize(int size) {
// size 的 1.5 倍，再加 1，再往上取最近的 2 的 n 次方。
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
else if (tab == table) {
int rs = resizeStamp(n);
if (sc < 0) {
Node<K,V>[] nt;
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
//使用 CAS 将 sizeCtl 加 1，然后执行 transfer 方法
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);
}
}
}
6.2.4、transfer 数据迁移
这个方法很点长，它是将原来的 tab 数组的元素迁移到新的 nextTab 数组中。
此方法支持多线程执行，调用此方法的时候，会保证第一个发起数据迁移的线程，nextTab 参数为 null，之后再调用此方法的时候，nextTab 不会为 null。
原数组长度为 n，所以我们有 n 个迁移任务，让每个线程每次负责一个小任务是最简单的，每做完一个任务再检测是否有其他没做完的任务，帮助迁移就可以了，而 Doug Lea 使用了一个 stride，简单理解就是步长，每个线程每次负责迁移其中的一部分，比如每次迁移 16 个小任务。所以，我们就需要一个全局的调度者来安排哪个线程执行哪几个任务，这个就是属性 transferIndex 的作用。
第一个发起数据迁移的线程会将 transferIndex 指向原数组最后的位置，然后从后往前的 stride 个任务属于第一个线程，然后将 transferIndex 指向新的位置，再往前的 stride 个任务属于第二个线程，依此类推。
/**
* Moves and/or copies the nodes in each bin to new table. See
* above for explanation.
*/
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// stride 在单核下直接等于 n，多核模式下为 (n>>>3)/NCPU，最小值是 16
// stride 可以理解为 步长，有 n 个位置是需要进行迁移的，
// 将这 n 个任务分为多个任务组，每个任务组有 stride 个任务
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
// 如果 nextTab 为 null，先进行初始化
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 是 ConcurrentHashMap 中的属性
nextTable = nextTab;
// transferIndex 是 ConcurrentHashMap 的属性，用于控制迁移的位置
transferIndex = n;
}
int nextn = nextTab.length;
// 这个构造方法会生成一个Node，key、value 和 next 都为 null，关键是 hash 为 MOVED
// 后面我们会看到，原数组中位置 i 处的节点完成迁移工作后，就会将位置 i 处设置为这个 ForwardingNode，用来告诉其他线程该位置已经处理过了
// 它其实相当于是一个标识。
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
// i 是位置索引，bound 是边界
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// advance 为 true 表示可以进行下一个位置的迁移了
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
// 将 transferIndex 值赋给 nextIndex
// 这里 transferIndex 一旦小于等于 0，说明原数组的所有位置都有相应的线程去处理了
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
// nextBound 是这次迁移任务的边界，是从后往前
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
// 所有的迁移操作已经完成
nextTable = null;
// 将新的 nextTab 赋值给 table 属性，完成迁移
table = nextTab;
// 重新计算 sizeCtl：n 是原数组长度，所以 sizeCtl 得出的值将是新数组长度的 0.75 倍
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// sizeCtl 在迁移前会设置为 (rs << RESIZE_STAMP_SHIFT) + 2
// 每个线程参与迁移就会将 sizeCtl 加 1，
// 这里使用 CAS 操作对 sizeCtl 进行减 1，代表做完了属于自己的任务
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
}
}
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) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
// 头结点的 hash 大于 0，说明是链表的 Node 节点
if (fh >= 0) {
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);
}
// 其中的一个链表放在新数组的位置 i
setTabAt(nextTab, i, ln);
// 另一个链表放在新数组的位置 i+n
setTabAt(nextTab, i + n, hn);
// 将原数组该位置处设置为 fwd，代表该位置已经处理完毕，
// 其他线程一旦看到该位置的 hash 值为 MOVED，就不会进行迁移了
setTabAt(tab, i, fwd);
// advance 设置为 true，代表该位置已经迁移完毕
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;
}
}
// 如果一分为二后，节点数少于 8，那么将红黑树转换回链表
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;
// 将 ln 放置在新数组的位置 i
setTabAt(nextTab, i, ln);
// 将 hn 放置在新数组的位置 i+n
setTabAt(nextTab, i + n, hn);
// 将原数组该位置处设置为 fwd，代表该位置已经处理完毕，
// 其他线程一旦看到该位置的 hash 值为 MOVED，就不会进行迁移了
setTabAt(tab, i, fwd);
// advance 设置为 true，代表该位置已经迁移完毕
advance = true;
}
}
}
}
}
}
6.3、Size 方法
由源码可知该方法的最大返回值为 Integer.MAX_VALUE，但是这个 Map 的长度是有可能超过 int 最大值的，所以 JDK 8 增了 mappingCount 方法。
mappingCount 方法的返回值是 long 类型。所以不必限制最大值必须是 Integer.MAX_VALUE
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
/**
* Returns the number of mappings. This method should be used
* instead of {@link #size} because a ConcurrentHashMap may
* contain more mappings than can be represented as an int. The
* value returned is an estimate; the actual count may differ if
* there are concurrent insertions or removals.
*
* @return the number of mappings
* @since 1.8
*/
public long mappingCount() {
long n = sumCount();
return (n < 0L) ? 0L : n; // ignore transient negative values
}