JDK1.8 concurrentHashMap 同步机制难点解析
Posted kobebyrant
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了JDK1.8 concurrentHashMap 同步机制难点解析相关的知识，希望对你有一定的参考价值。
jdk1.8我认为有几个主要的难点：
- 同步机制
- 红黑树的操作
- 数学原理（重要是基于统计值的算法选取和变量设定）
其中这里只分析同步机制中比较重要的部分。
这篇东西和上一篇文章LongAdder的原理关联性比较大，如果懂LongAdder的则忽略。
全文主要从以下几方面来讲：
- 为什么1.8 concurrentHashMap要重构
- 1.8 concurrentHashMap的基本设计
- 1.8concurrentHashMap的难点方法解析
- 个人存在的疑惑点
这个类是写博客至今研究过的难度最大的类，前前后后看了很久才大致有点感觉。所以在写的时候只能通过先写第二第三点来温习知识才能写第一点（所以一些语言的顺序性可能会很奇怪）。
为什么1.8 concurrentHashMap要重构
我们记得jdk1.7中concurrentHashMap是用了分段锁的设计原理，segment的数量取决于concurrencyLevel；
然后rehash的时候是在segment（分段锁）里面操作的，虽然说是没有阻塞整个哈希表，但是是会阻塞某个分段锁。
1.8 concurrentHashMap中抛弃了分段锁，并且把put流程的控制粒度更加细化，细化的粒度降低到当前哈希表中的一个HashEntry。
而且整个表需要resize的时候不会阻塞任何一个线程，当容量大并且需要resize的时候，可能并发量越高resize的速度还会越快。
因此可以说1.8的concurrentHashMap提升了并发吞吐量和减少了resize的时候的阻塞时间，究竟他是怎么做到的呢？
1.8 concurrentHashMap的基本设计
类继承体系和之前的1.7 concurrentHashMap类似。
但是这里没有通过分段锁来保证并发的吞吐量。
先看看基本的类成员变量
/**
* Minimum number of rebinnings per transfer step. Ranges are
* subdivided to allow multiple resizer threads. This value
* serves as a lower bound to avoid resizers encountering
* excessive memory contention. The value should be at least
* DEFAULT_CAPACITY.
*/
private static final int MIN_TRANSFER_STRIDE = 16;
/**
* The number of bits used for generation stamp in sizeCtl.
* Must be at least 6 for 32bit arrays.
*/
private static int RESIZE_STAMP_BITS = 16;
/**
* The maximum number of threads that can help resize.
* Must fit in 32 - RESIZE_STAMP_BITS bits.
*/
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
/**
* The bit shift for recording size stamp in sizeCtl.
*/
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;MIN_TRANSFER_STRIDE： 意思是resize时候每个线程每次最小的搬运量（搬运多少个entry），范围会继续细分以便可以允许多个resize线程。
RESIZE_STAMP_BITS ：位的个数来用来生成戳，用在sizeCtrl里面；
MAX_RESIZERS：最大参与resize的线程数量;
RESIZE_STAMP_SHIFT：用来记录在sizeCtrl中size戳的位偏移数。
/*
* Encodings for Node hash fields. See above for explanation.
*/
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
/** Number of CPUS, to place bounds on some sizings */
static final int NCPU = Runtime.getRuntime().availableProcessors();上面定义了一些特别的哈希值来表征节点的状态：
-1：forwarding nodes 主要作用是表征一个节点已经被处理干净（resize的时候被转移到新表了）
-2：表示树的根节点
-3：表示transient
HASH_BITS：31个1用来计算普通的node的哈希码
NCPU：当前可用的处理器的核数量
//也是用节点数组来构成哈希表，在第一次插入的时候会懒初始化，size是2的整数次幂，直接通过iterators来访问
transient volatile Node<K,V>[] table;
//下一个要用的数组，当在进行resizing的时候非空
private transient volatile Node<K,V>[] nextTable;
//用来计算整个表的size，和longadder里面的base一致
private transient volatile long baseCount;
//用来控制表初始化，当为负数值。各种状态下这个int的值如下:
//初始化的：-1
//resize：-(1 + 被激活的参与resize的数量)
//表为空：表的initial size，如果没有这个值则为0
//初始化后：该表下次应该resize的值，等于当前表的size的0.75倍
private transient volatile int sizeCtl;
//当resizing的时候下一个tab下标索引值（当前值+1）
private transient volatile int transferIndex;
//当resize和创建counterCells的时候的自选锁，和longadder一致
private transient volatile int cellsBusy;
//和longadder的cells一致
private transient volatile CounterCell[] counterCells;接下来我们看看构造方法
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel)
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
//concurrencyLevel表示估计的参与并发更新的线程数量，必须比初始化容量的要大
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
//整个构造方法过程就是为了能到这一步。
//初始化只有这一个实际的赋值方法，因此是懒初始化的，当前的map是null的，sizeCtl存储的值是当前要初始化的map的size值
this.sizeCtl = cap;
1.8concurrentHashMap的难点方法解析
其中最难也是最核心的方法肯定是put方法。
注：以下的模拟值全部按照默认初始化值进行计算
public V put(K key, V value)
return putVal(key, value, false);
putVal很长，解释直接注释在代码内；
建议自己跟读代码的时候先不要杀进去C和E的部分，看大概扫一眼正常的流程。建议先看A,B和D。
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent)
//concurrentHashMap中 key和value都不能为空
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
//自旋操作，每次都把当前的table赋给tab
for (Node<K,V>[] tab = table;;)
Node<K,V> f; int n, i, fh;
//A:concurrentHashMap懒初始化，初始化表
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//B:找到应该put的index对应的节点，并赋值给f
//如果node数组下标对应的node为空，则cas新建，由于自旋失败了也无所谓
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null)
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
//C:这种情况是在resize
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
//D：正常熟悉的哈希表的put流程
else
V oldVal = null;
//针对单个node节点加锁
synchronized (f)
//双重检测
if (tabAt(tab, i) == f)
//正常的节点hash值>0
if (fh >= 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))))
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;
//树节点hash值为-2
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;
//如果不是onlyIfAbsent为true，且已经找到了key对应的node的话都会来到这一步，因为都会插入成功
if (binCount != 0)
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
//E:size+1，里面会触发resize操作
addCount(1L, binCount);
return null;
A：initTable
/**
* Initializes table, using the size recorded in sizeCtl.
*/
private final Node<K,V>[] initTable()
Node<K,V>[] tab; int sc;
//自旋操作，每次都对tab赋值而且判断tab的tab数组的长度
while ((tab = table) == null || tab.length == 0)
//如果抢锁失败（sizeCtl是作为自选锁使用），则告诉cpu可以让出时间片
//其他线程如果初始化表成功则自旋结束退出方法
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1))
//此时抢锁成功
try
//双重检测
if ((tab = table) == null || tab.length == 0)
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//sizeCtl置为n的0.75倍
sc = n - (n >>> 2);
finally
//赋值顺便解锁
sizeCtl = sc;
//自旋结束
break;
return tab;
先看E：这样会更好理解。
方法头的注释意思大概是：
增加表容量的计数，如果这个表需要resize但还没开始resize，则初始化transfer（这个东西用来进行resize）。如果已经开始resize了，则看看需不需要加入帮助一起resize。然后重新检查表的计数看看需不需要再次resize。
以下的代码片段第一次看的时候最好先忽略A部分
/**
* Adds to count, and if table is too small and not already
* resizing, initiates transfer. If already resizing, helps
* perform transfer if work is available. Rechecks occupancy
* after a transfer to see if another resize is already needed
* because resizings are lagging additions.
*
* @param x the count to add
* @param check if <0, don't check resize, if <= 1 only check if uncontended
*/
//这个方法既可以做加法（put的时候调用）又可以做减法（remove或者clear的时候）
//只有对size做加法的时候才用检查resize
private final void addCount(long x, int check)
CounterCell[] as; long b, s;
//A:以下代码和longadder的add操作思路完全一致
//这个方法等同于longAdder的Add
//然后fullAddCount等同于Striped64的longAccumulate，可以参考我之前的博客
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x))
CounterCell a; long v; int m;
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)))
fullAddCount(x, uncontended);
return;
if (check <= 1)
return;
//给s赋值为当前的哈希表的size
s = sumCount();
//这里要开始判断到底需不需要resize
if (check >= 0)
Node<K,V>[] tab, nt; int n, sc;
//短路与的话看如何能够进入代码块，这里代码风格和以往一致，判断的同时赋值
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY)
//来到这里相当于就一定要resize了，至于这个戳有个印象就好，每次resize每个线程都确认一个戳！标识当前的size（当前的2的多次整数次幂）
//生成一个戳 算法是Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
//如果n=16，则生成的值为0000000000000000010000000000011011（27和2的15次方相或）
int rs = resizeStamp(n);
if (sc < 0)
//size<0表征当前正在resize
//A :这里应该算是自旋的停止条件: 能够到达下一个if需要经过5个条件:
//1. 可以对比一下上面和C的两个值，发现应该是要相等的
// 2.sc!=rs+1;这个暂时不知道是什么意思
//resizer线程不能超过最大允许数量;nextTable非空;transferIndex>=0（resize没结束）
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
//B : 每次将这个值SIZECTL cas+1 成功的话参与transfer
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
//C: 第一个参与resize的线程：SIZECTL置为1000000000001101100000000000000010,这个值后面+2是为了计算上面的停止条件的，不能让resizer线程无限制增加
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
//到了这里就是resize完成了，然后通过下一个while自旋判断是否需要再次resize
s = sumCount();
接下来看transfer方法（最难方法），方法头说了这个方法就是将每个bin从旧的table转移到新的table；
以下方法建议先看主线方法E，再看ABCD
/**
* 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表征的是每一个thread进来的时候要搬运的量
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
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;
//第一次进来resize的时候初始化下一次需要的table，
//transferIndex赋值为n，意味着从后往前循环进行表转移
nextTable = nextTab;
transferIndex = n;
int nextn = nextTab.length;
//这个节点第一次看会很懵，在这个方法中是用来表征当前tab[i]的节点不需要被搬运的意思
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
//自旋操作并初始化i和bound，i为操作索引，bound为下限
for (int i = 0, bound = 0;;)
Node<K,V> f; int fh;
//A advance表征是否继续走下去：1.是否继续--i来转移表 2.是否继续cas TRANSFERINDEX来获取当前的i值
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)))
//第一次进入的时候如果到达了这里则接到了当前应该进行的任务：负责搬运[nextBound:i]之间的内容
bound = nextBound;
i = nextIndex - 1;
advance = false;
//B 当线程到达这里则证明家搬运结束，或者异常停止
if (i < 0 || i >= n || i + n >= nextn)
int sc;
//如果其他人已经设置了这个标志位，则正式完成并return
if (finishing)
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1))
//这个判定条件在addCount的A段方法解释过
//意思就是双重校验失败的意思，其他线程已经做过了这个操作了
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit
//C 如果领取到了的任务i对应的tab[i] 为空，那恭喜了把fwd贴上去然后继续领活吧，cas失败也不要紧~下次来看看tab[0]是不是被打标签了，是的话恭喜了，继续加班吧，不是的话就再试试cas总会有加班的一天的
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//D 意思是就是已经被别人打上了fwd的表情，可以重新领取任务了
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
//E搬运的主线流程
else
synchronized (f)
//双重检测
if (tabAt(tab, i) == f)
Node<K,V> ln, hn;
//如果不是树节点
if (fh >= 0)
int runBit = fh & n;
Node<K,V> lastRun = f;
//这里的高低位的分法其实和1.7 concurrentHashMap一样的逻辑，这里不解释了
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);
//注意这里! 将第tab的第i个项的node设为fwd，这样就意味着这个节点不需要再次被搬运了，同时他的hash值=-1
setTabAt(tab, i, fwd);
//搬运了一次，会重置i和bound继续自旋领取任务作为码农的奖励，这里别以为帮大家搬砖一次就完事了，就像你进了单位以为干完活就能早下班，实际上领导看你有空会给你继续派活直到整个组都没活干为止
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);
advance = true;
这个方法就是神一样的方法，解决了1.7concurrentHashMap中有可能resize影响性能的问题，原理就是让每一个put完的线程，或者想要put的线程到了整个表需要resize的时候都过来进入resize流水线工作，直到有一个线程说自己已经全部弄完了，方法才能返回。
以前的正常设计是resize的时候整个表就阻塞住了，但是现在resize的时候，想要操作的线程都会来参与一起resize，这么一来其他的线程就不用干等着了。
相当于想完成一个项目，大家伙都先做完了一个活，然后有一个人分配的活最多，而且都是阻塞的活，那这个时候大家干等着还不如一起帮他来完成这个活，这个被派到搬砖脏活的人也很聪明，把该搬的砖头分成一份份，让其他同事可以有序的完成，这个方法展现了团结友爱的精神也展现了被派了脏活的人的机智和有条不紊。
现在回过头来看putVal方法的C部分：
现在来看代码就很好懂了，和addCount的下半部分大致类似，只要记住当前的状态是当前线程还没有put的状态，而且已经有线程开始进行流水线搬砖resize的工作了。
/**
* 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;
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;
至此put方法讲解完毕，如果这个方法能够看懂，其他方法难度应该都不大，写不动了，下次有空再看看其他方法。
个人疑惑
还有一一些疑问请教大神们：
- addCount方法里面的一个判定条件sc!=rs+1是什么意思？
请知道答案的大腿不吝赐教，谢谢！
以上是关于JDK1.8 concurrentHashMap 同步机制难点解析的主要内容，如果未能解决你的问题，请参考以下文章
ConcurrentHashMap 源码详细分析(JDK1.8)
ConcurrentHashMap(JDK1.8)中红黑树的实现
JUC系列并发容器之ConcurrentHashMap(JDK1.8版)