走近concurrentHashMap（JDK1.8）
前言
前面我们学习了HashMap的数据结构，分析了其源码 在本篇文章中与HashMap相同的部分就不在赘述，但是HashMap是线程不安全的容器，在多线程环境下会有线程完全问题，虽然也有线程安全容器Hashtable，但是其通过synchronized修饰方法，通过独占锁的方式锁定类对象，效率不高，所以Java 又提供了线程安全容器ConcurrentHashMap，与HashMap的底层的数据结构相同,ConcurrentHashMap也是采用的“散列表+链表+红黑树”,不过红黑树中存储的不是TreeNode，而是TreeBin。在JDK1.8中 ConcurrentHashMap 大量采用CAS算法,unsafe.compareAndSwapInt(this, valueOffset, expect, update); CAS（compareAndSwap）比较并交换，就是比较valueOffset位置上的值是否等于expect，如果等于的话则返回true,并更新值。（PS：在JDK1.7中采用的是分段锁的方式）。在扩容，设值的过程中大量采用CAS无锁不阻塞的方式支持并发操作，但是是不是就不需要加锁了呢？答案是否定的。
环境
本源码基于JDK1.8
源码分析
首先，我们来看看ConcurrentHashMap中三个重要的原子操作。这三个方法的作用分别的 1.获得在i位置上的Node节点 2.利用CAS算法设置i位置上的Node节点 3.设置节点位置的值，仅在上锁区被调用
ConcurrentHashMap定义的三个原子操作
//获得在i位置上的Node节点
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);
}
//利用CAS算法设置i位置上的Node节点，之所以能实现并发是因为他指定了原来这个节点的值是多少。
//在CAS算法中，会比较内存中的值与指定的值是是否相等，如果相等则更新，并返回true，如果不相等则不更新，直接返回false。
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
//设置节点位置的值，仅在上锁区被调用
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
接下来我们来看看容器初始化的过程，调用构造函数只是设置了相关的参数，并没有实际的创建容器，分配内存，initTable 是在调用put方法时才会调用的,也就是说只有设置了第一个元素，才会真正的初始化容器。
初始化initTable方法
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//sizeCtl 表示有其他线程正在进行初始化操作，把线程挂起，对于table的初始化操作，只能有一个线程进行
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { //利用CAS方法把sizeCtl的值设置为-1，表示本线程正在进行初始化
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2); //相当于0.75*n 设置一个扩容的阈值
}
} finally {
//将sc赋值给sizeCtl
sizeCtl = sc;
}
break;
}
}
return tab;
}如上代码：我们初始化容器主要就是创建一个大小为n的Node数组，可以看出与HashMap中初始化容器最大的不同就是加了并发控制，通过共享变量sizeCtl来控制的，如果sizeCtl小于0，表示表示有其他线程正在进行初始化操作，把线程挂起，对于table的初始化操作，只能有一个线程进行。否则，通过CAS的方式将sizeCtl的值设置为-1，表示本线程正在进行初始化。最后在finally代码块中将sizeCtl设置为0.75*n。
sizeCtl 控制符标识
这里我们特别需要注意的一个变量是sizeCtl。
private transient volatile int sizeCtl; //控制标识符sizeCtl是一个用于同步多个线程的共享变量，是控制符标识符，扩容控制标识符，负数表示正在进行初始化或者扩容，操作，-1表示正在初始化，-N表示有N-1个线程正在进行扩容操作。正数或者0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小，类似于扩容阈值，它的值始终是当前ConcurrentHashMap容量的0.75倍，这与loadfactor是对应的。实际容量>=sizeCtl,则扩容。
ForwardingNode
ForwardingNode是一个用于连接两个table的节点类，它包含一个nextTable指针，用于指向下一张表，而这个节点的key,value next指针全部为null，它的hash值为-1，这里面定义的find的方法从nextTable里进行查询节点，而不是以自身为头节点进行查找。
static final class ForwardingNode<K,V> extends Node<K,V> {
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
super(MOVED, null, null, null);
this.nextTable = tab;
}
Node<K,V> find(int h, Object k) {
// loop to avoid arbitrarily deep recursion on forwarding nodes
outer: for (Node<K,V>[] tab = nextTable;;) {
Node<K,V> e; int n;
if (k == null || tab == null || (n = tab.length) == 0 ||
(e = tabAt(tab, (n - 1) & h)) == null)
return null;
for (;;) {
int eh; K ek;
if ((eh = e.hash) == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
if (eh < 0) {
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode<K,V>)e).nextTable;
continue outer;
}
else
return e.find(h, k);
}
if ((e = e.next) == null)
return null;
}
}
}扩容方法tryPresize和transfer
首先我们来看看tryPresize方法，与HashMap类似的是ConcurrentHashMap扩容也是分为两步，创建两倍大小的新数组，复制原数组的元素到新新数组，但是因为支持并发，所以ConcurrentHashMap的扩容过程相对复杂不少。接下来我们从源码可以看到，在复制元素时并不是对整个容器进行加锁，其只是锁住table[i]链表的头结点位置，其余线程可以向后继续遍历该容器。大大提高了并发度。
/**
扩容相关
tryPresize在putAll以及treeifyBin中调用
*/
private final void tryPresize(int size) {
//给定的容量若>=MAXIMUM_CAPACITY的一半，直接扩容到允许的最大值，否则调用tableSizeFor函数计算
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
//只有大于等于0才表示该线程可以扩容
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
if (tab == null || (n = tab.length) == 0) { //没有被初始化
n = (sc > c) ? sc : c;
// 期间没有其他线程对表操作，则CAS将SIZECTL状态置为-1，表示正在进行初始化
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) { //再一次检查
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2); //无符号右移2位，此即0.75*n
}
} finally {
sizeCtl = sc; //更新扩容阈值
}
}
}
//若欲扩容值不大于原阈值，或现有容量>=最值，什么都不用做了
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
else if (tab == table) { //table不为空，且在此期间其他线程未修改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;
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);
}
}
}如上代码，当前线程获取到sizeCtl>=0时则进入循环，在循环内，首先判断当前是否为空，如果为空在对容器进行初始化，与initTable()方法相同，否则，如果欲扩容值不大于原阈值，或现有容量>=最值，什么都不用做了。如果table不为空，且其他线程在此期间未修改table，进入扩容的分支。transfer扩容操作的两部分
1.第一部分是构建一个newTable，它的容量是原来的两倍。这个过程是单线程操作的，通过RESIZE_STAMP_SHIFT通过一次原子运算保证。
2.第二部分是将node移动到newTable上去。这个是支持多线程并发执行的。我们来看看第二步 在单线程环境下 首先是遍历原table,将位置i上的元素复制到newTable上去,将位置i上的元素的头结点赋值个f，fh表示结点的hash值。
3.如果这个位置为空，就在原table中的i位置放入forwardNode节点；4.如果fh>=0,表示这是一个链表的头结点。然后构建一个反序链表，然后遍历反序链表，将链表上的元素复制到newTable的i和i+n的位置上5.如果fh<0,表示这是一个ForwardingNode结点。6.如果f 是TreeBin，则说明这是一个树结点，也做一个反序处理，并且判断是否需要untreefi，把处理的结果分别放在nextTable的i和i+n的位置上 元素复制完成之后，让newTable成为一个新的table。在多线程下 在代码的第69行，有一个判断如果遍历到ForwardingNode节点，说明这个点已经被处理过了，直接跳过，继续向后遍历，同时对位置i上的链表的头结点上锁，保证了多线程的安全控制，当一个线程将位置i处的结点处理完成之后则将该节点置为forward结点，就这样交叉复制，就完成了复制工作。
static final int MOVED = -1; // hash for forwarding nodes
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
//构造一个连节点指针，用于标志位
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true; //并发扩容的关键属性，如果等于true,说明这个节点已经处理过
boolean finishing = false; // 确保在提交newTab之前完成清理
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//这个while循环体的作用就是在控制i--,通过i--可以依次遍历原hash表中的节点
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;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
//如果所有的节点都已经完成复制工作，就把nextTable赋值给table，清空临时对象nextTable
nextTable = null;
table = nextTab;
//扩容阈值设置为原来容量的1.5倍，依然相当于现在容量的0.75倍
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//利用CAS方法更新这个扩容阈值，在这里sizeCtl值减一，说明新加入一个线程参与扩容操作
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
}
}
//如果遍历到的节点为空，则放入ForwardingNode操作
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;
//如果fh>=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);
}
//在nextTable的i位置插入一个链表
setTabAt(nextTab, i, ln);
//在nextTable的i+n的位置插入另一个链表
setTabAt(nextTab, i + n, hn);
//在table的i位置插入forwardNode节点，表示已经处理过该节点
setTabAt(tab, i, fwd);
//设置advance为true，返回到上面的while循环中，就可以执行i--操作
advance = true;
}
//对TreeBin对象进行处理，与上面的过程类似
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;
}
}
//如果扩容后已经不再需要tree的结构，反向转换为链表结构
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;
//在nextTable的i位置插入一个链表
setTabAt(nextTab, i, ln);
//在nextTable的i+n的位置插入另一个链表
setTabAt(nextTab, i + n, hn);
//在table的i位置上插入forwardNode节点，表示已经处理过该节点
setTabAt(tab, i, fwd);
//设置advance为true，返回到上面的while循环中，就可以执行i--操作
advance = true;
}
}
}
}
}
}put方法
put 方法是ConcurrentHashMap的最核心方法，前面做的一系列铺垫也是为了解释put方法做准备。与HashMap类似，put操作采用CAS+synchronized实现并发插入或更新操作。其主要的过程也是如下几步：
1.计算hash值，然后根据hash值计算出key的存储位置i。2.如果位置i没有值，则通过CAS算法直接放进去，不需要加锁3.如果table[i]的节点的hash等于MOVED，则检测到正在扩容，则帮助其扩容4.如果不等于MOVED,则给链表的头结点上锁，然后遍历链表，通过尾插法的方式将元素插入到链表的尾部。
说明：
5.获取index位置上的元素是通过 U.getObjectVolatile 来获取的，而不是直接通过table[index] 这是为什么呢？在java内存模型中，我们知道每个线程都有一个工作内存，里面存储的是table的副本，虽然table是volatile修饰的，但是不能保证线程每次都拿到table的最新元素。U.getObjectVolatile 可以直接获取指定内存的数据，保证了每次取到的数据都是最新的。6.ConcurrentHashMap和HashMap的区别还有一点，就是HashMap允许一个key和value为null，而ConcurrentHashMap则不允许key和value为null，如果发现key或者value为null，则会抛出NPE，这一点需要特别注意，而这也说明，在ConcurrentHashMap中可以通过使用get操作来测试是否具有某个记录，因为只要get方法返回null，就说明table中必然不存在一个记录和当前查询的匹配，而在HashMap中，get操作返回null有可能是我们查询的记录的value就是null，所以不能使用get方法来测试某个记录是否存在于table中。
put方法流程图
final V putVal(K key, V value, boolean onlyIfAbsent) {
//不允许key或者value为null？
if (key == null || value == null) throw new NullPointerException();
//计算hash值
int hash = spread(key.hashCode());
int binCount = 0;
//死循环，何时插入成功，何时跳出
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//如果table为空的话，初始化table
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//根据hash值计算出在table里面的位置
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//如果这个位置没有值，则通过CAS算法直接放进去，不需要加锁
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//检查table[i]的节点的hash是否等于MOVED,如果等于，则检测到正在扩容，则帮助其扩容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f); //帮助其扩容
else { //运行到这里，说明table[i]的节点的hash值不等于MOVED
V oldVal = null;
//结点上锁，这里的结点可以理解为hash值相同组成的链表的头结点
synchronized (f) {
if (tabAt(tab, i) == f) {
//fh>=0 说明这个节点是一个链表的节点，不是树的节点
if (fh >= 0) {
binCount = 1;
//在这里遍历链表所有的结点
for (Node<K,V> e = f;; ++binCount) {
K ek;
//如果hash值和key值相同，则修改对应结点的value值
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
//如果遍历到最后一个结点，那么就证明新的节点需要插入（尾插法）
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;
}
}
}
}
if (binCount != 0) {
//如果链表长度已经达到临界值8，就需要把链表转换成树结构
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
//将当前ConcurrentHashMap的元素数量+1
addCount(1L, binCount);
return null;
}helpTransfer方法
helpTransfer方法是辅助进行扩容的方法，进入该方法说明nextTable 已经产生，只需要进行元素的复制工作。
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
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) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}treeifyBin方法
该方法是将过长的链表转换为TreeBin对象，但它并不是直接进行转换的，而是进行一次容量判断如果容量未达到则先去扩容并返回，否则，给当前节点的头结点上锁，然后，遍历，将所有的Node节点包装成TreeNode节点放进TreeBin对象中。
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
//如果table.length<64 就扩大一倍 返回
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
//构造了一个TreeBin对象，把所有Node节点包装成TreeNode放进去
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); //这里只是利用了TreeNode封装，而没有利用TreeNode的next域和parent域
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
//在原来index的位置，用TreeBin替换掉原来的Node对象
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}get方法
最后我们来看看get方法的源码，get方法获取元素比较简单，与HashMap不同的是，获取元素的方式是是通过CAS的方式获取。
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
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;
}
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;
}总结
ConcurrentHashMap是一个并发散列映射表的实现，它允许完全并发的读取，并且支持给定数量的并发更新，比Hashtable的性能更高，在JDK1.8中的并发控制是通过CAS+synchroinzed 来实现的，摒弃了JDK1.8之前的分段锁实现方式。
参考文章
《Java源码分析》：ConcurrentHashMap JDK1.8[1] ConcurrentHashMap源码分析（JDK8版本）[2] Java 8 ConcurrentHashMap源码分析[3] 深入分析ConcurrentHashMap1.8的扩容实现[4]
References
[1] 《Java源码分析》：ConcurrentHashMap JDK1.8: https://blog.csdn.net/u010412719/article/details/52145145
[2] ConcurrentHashMap源码分析（JDK8版本）: https://blog.csdn.net/u010723709/article/details/48007881
[3] Java 8 ConcurrentHashMap源码分析: https://juejin.im/entry/59fc786d518825297f3fa968
[4] 深入分析ConcurrentHashMap1.8的扩容实现: https://www.jianshu.com/p/f6730d5784ad
腾讯云开发者
