【Java并发】- ConcurrentHashMap原理解析(JDK1.8)
程序员文章站
2022-06-07 13:36:27
...
ConcurrentHashMap概述
在JDK1.8中ConcurrentHashMap的实现方式或者说实现思想相比之前的版本已经相差了很大了，在JDK1.8之前ConcurrentHashMap是采用的是Segment（分段锁）的方式，而JDK1.8中的ConcurrentHashMap则摒弃了原来的做法，同时沿用了JDK1.8中的HashMap的数组+链表+红黑树的方式，同时大量的使用了java并发包作者非常多用的Unsafe里面的方法，也就是利用CAS的思想来实现，虽然已网上已经有了很多优秀的分析，但是为了加深印象还是自己写一下。
数据构成
ConcurrentHashMap与HashMap的内部数据结构与HashMap的思想基本一致，都是使用的数组+链表+红黑树的，但是相比之下ConcurrentHashMap也有一些改动，比如ConcurrentHashMap的树结构并不是像HashMap那样直接链上TreeNode节点而是在内部定义了一个TreeBin的类来作为容器去包含了这些树节点，也就是说实际存放在数组上的对象是TreeBin而不是TreeNode，同时TreeNode也承担了红黑树的构造的工作。
并发
ConcurrentHashMap的数据的插入与删除等的基本流程和HashMap基本一致，在线程安全方面ConcurrentHashMap使用了CAS方式加上synchronized关键字而synchronizez使用的锁则是当前数组中需要加入的位置对应的数组元素，也就是“动哪儿锁哪儿”。所以并发上面性能也很好。
在HashMap中有一个很头疼的resize扩容很慢的问题也得到了解决，ConcurrentHashMap允许多个线程并发的去进行resize操作，这样大大的提高了效率。
源码解析
由于许多内容是和HashMap相同的，说明的会适当的简略一些。
重要的属性
//多线程扩容的时候单个线程负责的table中的元素个数
private static final int MIN_TRANSFER_STRIDE = 16;
private static int RESIZE_STAMP_BITS = 16;
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
static final int MOVED = -1; // hash for forwarding nodes ForwardNode的hash值
static final int TREEBIN = -2; // hash for roots of trees //TreeBin节点的的hash值
static final int RESERVED = -3; // hash for transient reservations //ReservationNode节点的的hash值
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash //用于计算hash值
//获取可用的CPU数
static final int NCPU = Runtime.getRuntime().availableProcessors();
//数组
transient volatile Node<K,V>[] table;
//只有在进行扩容的时候使用
private transient volatile Node<K,V>[] nextTable;
/* 当是负数的时候表示正在初始化或者resize
* 当是-1的时候表示正在初始化
* -N 表示有N-1个线程正在进行resize
*/
private transient volatile int sizeCtl;
//用于多线程一起扩容的时候记录位置
private transient volatile int transferIndex;其中的sizeCtl变量的出镜率很高。用于控制并发resize的时候的线程。
节点
Node节点
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
//这里与HashMap不同的是val和next被定义为了volatile
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;
}
public final K getKey() { return key; }
public final V getValue() { return val; }
public final int hashCode() { return key.hashCode() ^ val.hashCode(); }
public final String toString(){ return key + "=" + val; }
//不允许setValue（与HashMap不同）
public final V setValue(V value) {
throw new UnsupportedOperationException();
}
public final boolean equals(Object o) {
Object k, v, u; Map.Entry<?,?> e;
return ((o instanceof Map.Entry) &&
(k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
(v = e.getValue()) != null &&
(k == key || k.equals(key)) &&
(v == (u = val) || v.equals(u)));
}
//定义了find方法用于辅助map.get()方法
Node<K,V> find(int h, Object k) {
Node<K,V> e = this;
if (k != null) {
do {
K ek;
if (e.hash == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
} while ((e = e.next) != null);
}
return null;
}
}这里与HashMap不同的是使用volatile标识了val和next，是的val和next的读/写具备了原子性，可以配合CAS一起使用。
TreeNode
TreeNode和HashMap基本差不多。
TreeBin
static final class TreeBin<K,V> extends Node<K,V> {
TreeNode<K,V> root;
volatile TreeNode<K,V> first;
volatile Thread waiter;
volatile int lockState;
// values for lockState
static final int WRITER = 1; // set while holding write lock
static final int WAITER = 2; // set when waiting for write lock
static final int READER = 4; // increment value for setting read lock
//构造红黑树
TreeBin(TreeNode<K,V> b) {
super(TREEBIN, null, null, null);
this.first = b;
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;
if (r == null) {
x.parent = null;
x.red = false;
r = x;
}
else {
K k = x.key;
int h = x.hash;
Class<?> kc = null;
for (TreeNode<K,V> p = r;;) {
int dir, ph;
K pk = p.key;
if ((ph = p.hash) > h)
dir = -1;
else if (ph < h)
dir = 1;
else if ((kc == null &&
(kc = comparableClassFor(k)) == null) ||
(dir = compareComparables(kc, k, pk)) == 0)
dir = tieBreakOrder(k, pk);
TreeNode<K,V> xp = p;
if ((p = (dir <= 0) ? p.left : p.right) == null) {
x.parent = xp;
if (dir <= 0)
xp.left = x;
else
xp.right = x;
r = balanceInsertion(r, x);
break;
}
}
}
}
this.root = r;
assert checkInvariants(root);
}
//。。。。TreeBin在构造函数直接根据传入的TreeNode节点（一个TreeNode链表的头节点）来构造出红黑树，并且大量属性都用的volatile标识。
ForwardingNode
ForwardingNode是仅仅在resize的时候线程对某个位置替换成ForwardingNode节点表示已经有某个线程已经在对这个位置进行了操作，从而新线程将跳过这个节点往后面查找可以操作的节点。
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;
}
}
}
}可以看到ForwardingNode继承自Node但是实际上只有hash值一个属性，同时hash值使用的是MOVED(-1),同时还包含了一个属性nextTble指向下一个数组，同时还定义了一个find方法从nextTble总查询。
重要方法
三个核心操作
//获取i位置上的节点
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位置上的节点为v
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);
}
//利用volatile方法设置i位置的值
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}这三个方法都利用了JAVA中的Unsafe的方法，其中第二个CAS方式设置值的方法更是能够实现“无锁并发”的核心。
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();
//获取hash值
int hash = spread(key.hashCode());
int binCount = 0;
//自旋循环
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//如果table为空则初始化数组
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//如果通过hash算法计算的位置的元素为null，
//那么直接尝试构造一个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
}
//如果对应位置已经有元素存在，
//判断是hash值是否为MOVED(-1)，也就是是否是ForwardingNode
//如果是ForwardingNode说明现在正在进行resize操作，那么调用helpTransfer去协助其他线程(如果有的话)并发resize
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
//正常的插入
else {
V oldVal = null;
//synchronized加锁，加锁对象用的是当前位置的节点对象
synchronized (f) {
//稳定性判断
if (tabAt(tab, i) == f) {
//如果f节点的hash值大于等于0(说明是普通的Node节点)
//插入操作和HashMap基本一致
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;
}
}
}
//如果f节点的hash值小于0说明是TreeBin节点
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;
}
}
}
}
//binCount代表链表的长度
if (binCount != 0) {
//链表的长度大于8的话转化为红黑树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
//对size做+1的操作并判断是否需要resize
addCount(1L, binCount);
return null;
}上面的代码应该很清晰的能看出ConcurrentHashMap的插入流程：
- 如果数组还未初始化或者为空则先进行初始化操作。
- 如果当前插入位置为没有元素那么尝试CAS+自旋的方式插入
- 如果当前插入位置不为空先判断是否为ForwardingNode节点，是则说明正在进行resize那么尝试去协助其他线程一起并发的resize（此时并没有退出循环），resize完成后将再次循环的进行下一步的判断。
- 如果上面情况都不满足那么将当前插入位置的Node节点对象作为锁用synchronized同步进行插入。插入过程基本和HashMao一致。
数组的初始化
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
//自旋
while ((tab = table) == null || tab.length == 0) {
//如果sizeCtl小于0，说明已经有别的线程在初始化了，那么让出CPU执行权
//Thread.yield()会让当前线程放弃当前好不容易抢到的CPU执行权（并不进入等待）。
//相当于又回到了起点和其他的线程又开始竞争CPU的执行权
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) {
//确定容量
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//sc = n * 0.75
sc = n - (n >>> 2);
}
} finally {
//设置sizeCtrl
sizeCtl = sc;
}
break;
}
}
return tab;
}可以看到容器的初始化操作其实是单线程执行的，那么控制并发时候线程安全的核心还是利用CAS+自旋的方式并通过一个sizeCtrl变量的值来控制。
- 如果sizeCtrl>=0说明还没有别的线程开始初始化操作，那么CAS方式尝试将sizeCtrl置为-1，如果成功则说明获取到了初始化的权利。
- 如果sizeCtrl<0说明已经有别的线程抢先开始了初始化。
数组的resize
正常的扩容发生一般是：一个线程调用put方法，成功插入之后在addCount方法中发现需要resize所以开始去resize并初始化nextTable变量同时将原数组中的对应节点替换为ForwardingNode以表示这个节点已经有线程在操作了。这时候别的线程陆续调用put操作同时由于发现自己正在resize，所以调用helpTransfer去协助一起并发的resize。
先看addCount方法：
private final void addCount(long x, int check) {
//执行size+1的操作
CounterCell[] as; long b, s;
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 = sumCount();
}
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
//如果当前size超过了sizeCtl
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
//返回一个数，并且这个数的第16位为1
//也就是说rs<<RESIZE_STAMP_SHIFT必定是一个负数
int rs = resizeStamp(n);
//
if (sc < 0) {
//别的线程还没开始resize或者超过了最大数量限制的话退出
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
//CAS设置sizeCtrl为sizeCtrl+1，成功的话表示可以开始进行resize的操作
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
//CAS设置sizeCtrl为rs << RESIZE_STAMP_SHIFT) + 2(负数)，
//当前线程位第一个开始resize的线程尝试开始resize并且会初始化nextTable
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}这里的resize操作和初始化比较相似，也是通过一个sizeCtrl变量（复数）表示有多少个线程正在执行resize操作
transfer方法：
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//通过可用的CPU个数确定stride
//stride为一个线程处理的节点的个数
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;
}
//赋值给新的数组，并且
nextTable = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
//构造一个ForwardingNode节点并指向了nextTab
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
//控制并发处理的标识，true的话说明这个节点已经处理过了
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
//这里的i将遍历tab中的每个位置
//一个线程只能处理从index为bound到index为nextBound之间的节点
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
//从nextIndex - 1位置开始依次递减，直到nextBound位置
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//当前线程如果第一次进来则先通过transferIndex去确定自己应该在哪一段开始作业
//一般为transferIndex-stride到transferIndex-1的位置
//CAS设置transferIndex为transferIndex - stride
//如果transferIndex = stride那么transferIndex = 0
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
//bound = nextIndex
//i = nextIndex - stride
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
//resize完成
if (finishing) {
nextTable = null;
table = nextTab;
//sizeCtl为原来的长度的1.75倍
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//CAS尝试更新sizeCtl为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
}
}
//如果i位置节点为null将i位置节点替换为ForwardingNode
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//如果i位置节点已经被处理了，那么advance = true说明已经处理了
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
//resize部分
else {
//用当前节点加锁
synchronized (f) {
if (tabAt(tab, i) == f) {
//这里的ln.hn代表lowNode，highNode
Node<K,V> ln, hn;
//hash值大于0说明是Node节点，链表结构
if (fh >= 0) {
//获取到链表中的最后一个节点f以及高位是0还是1
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;
}
}
//说明在新的table中的位置不变
if (runBit == 0) {
ln = lastRun;
hn = null;
}
//说明在新的table中的位置应该位i + n
else {
hn = lastRun;
ln = null;
}
//因为只会在i和i+n两个位置，
//所以从当前链表的最后一个节点开始根据节点的hash值的高位是否为0构造两条新的链
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);
setTabAt(tab, i, fwd);
//节点处理完成
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;
}
}
}
}
}
}这里比较关键的就是stride，它代表了一个线程能处理的节点数。假设当前table的长度为64，stride为最小值16：
此时的transferIndex为64
- 进入for循环i=0,bound=0
- 进入while循环CAS设置transferIndex为transferIndex - stride = 48，此时i =63,bound =48,跳出while循环
- 替换table中i位置为ForwardingNode并做resize处理或者跳过当前节点。
-
i–处理第63个节点的位置。。直到i = 48或者已经结束
这个过程中如果有别的线程一起在并行操作，因为关键的transferIndex和ForwardingNode以及sizeCtrl都是CAS操作控制，失败则继续循环尝试即可。
helpTransfer方法：
helpTransfer方法的并发控制基本和transfer中差不多，都是考的是sizeCtrl变量的CAS操作来控制。
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;
}get方法
get方法方法的逻辑相对比较简单，利用hash值去做了一下判断。
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) {
//当前位置的hash值刚好相同
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//如果是TreeBin节点则调用find方法去查找
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
//是Node节点的链表则循环去查找
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}小结
JDK1.8中ConcurrentHashMap还有很多值得研究的方法，这里没有详细研究，同时对于ConcurrentHashMap的hash算法等设计部分因为是沿用HashMap的所以没有作说明。最后ConcurrentHashMap在并发上相比HashTable有很高的效率，减少了锁的竞争，同时也能看出CAS算法真的是被Doug Lea运用的炉火纯青。。。
上一篇： 浅析java.util.logging