ConcurrentHashMap需要解决的线程安全问题
CAS操作
ConcurrentHashMap为了保证线程安全，在多处使用的CAS乐观锁来保证线程是安全的，CAS的实现：
Cas(this,address,except,update)。其中this指对象，address内存地址（在数组的扩容时候，这个内存地址表示的是table对象属性中sizectl的内存地址，），except指的是期望值，update需要更新的值。当线程在进行更新操作时会检查当前对象的属性值是不是和except值相等，相等就更新。不等就返回false.
一 ConcurrentHashMap初始化问题：多个线程同时进行初始化怎么办。
首先多个线程都会进行判断动态数组是未为空或长度是否为0，然后判断sizCtrl是否小于0.都是刚进来的线程肯定不小于0，接线来多个线程会进行锁竞争使用乐观锁（U.compareAndSwapInt(this, SIZECTL, sc, -1)）来保证只有一个线程，能进行后面的初始化数组的操作，其他线程直接break 返回的是一个空数组。如果后面再有线程进来初始化时 发现sizCtl小于0，Thread.yield() 让出CPU的时间片。
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {//2
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {// 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 - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
流程图：
微信截图_20190727233331.png
二 如何保证多个线程在操作数组里的元素时都是最新值。
从ConcurrentHashMap的源码中可以看到其中的动态数组是有volatile关键字修饰的.但是这只能保证的数组都是最新的数组不能去保证数组中的元素都是最新的元素。这里ConcurrentHashMap是采用UnSafe.getObjectVolatile()来保证数组中元素的一致性。
transient volatile Node<K,V>[] table; // volatile关键字修饰的节点数组保证了多线程下,table永远都是最新值
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {//Unsafe 类去内存中取最新值
return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
三 Put数据时,如何保证只有一个线程能够在坑位下put一个元素成功
Cas乐观锁来控制。
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);
}
四 在判断当前数组坑位有元素的时候，如何去保证链表或树节点是线程安全的？
synchronized 上锁(互斥锁)锁住当前数组的这份下标，doublecheck一下当前下标的数组的值就是判断数组小标是否有值的值。然后进入自旋查到当前节点的下一个节点的一直查下一个节点，直到查到的这个节点的下一个节点为空时，将需要将新的节点放到节点的下一个节点为空节点的位置。
五 扩容之后数组的复制问题
ConcurrentHashMap在扩容之后进，进行数组复制的时候。如何保证多个线程同时参与到数组复制的过程中，同时这个时候如果还有新的线程来put元素的怎么办？
5.1 首先解决多个线程同时参与复制的问题
每一次put元素后都会addCount，addCount的过程也是CAS来操作，baseCount就是计数器，x代表每个线程put数据时+1操作。
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
//没有锁失败的线程会到这个里面来执行相关操
}
竞争锁成功的线程开始进行transfer(tab,null)操作，
分解来看transfer的代码：
1 首先为了能够保证过个线程都能参与到复制数组的操作，需要根据CPU的核数确定步长（每个线程需要复制数组中的哪一些部分到新数组中。）
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
}
2 初始化新的数组，初始化新数组的时候虽然没有CAS上锁操作，但是前面的进入transfer方法之前已经有了CAS了，而且在这里还进行了判断nextTab是否为空。原因是之前没有竞争到transfer的线程在自旋。
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
、、、、、、、、、、、、、、、、、、、、、、、、
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;
}
}
再来看addCount中的一部分代码,这个时候会进入到sc<0的代码块中，有两个单独的if判断第一if判断表示如果这个时候还没有初始完数组的时候，直接break。表示该线程不会参与transfer中了。第二个if表示其他线程进行CAS操作这个时候nextTable已经不为空了，参与到了tranfer中。
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 0) {
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);
s = sumCount();
}
为了验证我之前说的是正确的特地自己做了个实验，看都在什么时候会break，什么时候会transfer(tab,nt);
创建了20个线程并发操作完ConcurrentHashMap中放元素。
package com.chen.springboot.map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
public class ConcurrentHashMapTest {
private static ConcurrentHashMap<Integer, String> map = new ConcurrentHashMap<>();
private static CountDownLatch countDownLatch = new CountDownLatch(1);
public static void main(String[] args) {
for (int i = 0; i < 20; i++) {
Thread t = new Thread(new MapPutRunnable(i, String.valueOf(i)));
t.start();
}
countDownLatch.countDown();
}
static class MapPutRunnable implements Runnable {
private Integer key;
private String value;
public MapPutRunnable(Integer key, String value) {
this.key = key;
this.value = value;
}
@Override
public void run() {
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
for (int i = key * 100; i < 100 * (++key); i++) {
map.put(i, value);
}
}
}
}
addCount的流程图
image.png
在这两个地方打个断点就能看到效果啦。
image.png
5.2 解决数组在复制的过程中增加了put进来了新的元素
这时还需要分两种情况： 1 put元素的坑位这个时候还没有进行复制，这个就简单了就直接往这个坑位放元素就可以了，2 如果这个坑位已经都被迁移过了该怎么办？下面就来解决第二个问题。
迁移流程图：
Untitled Diagram(6).jpg
首先确定线程确定哪些坑位是我需要迁移的，然后从后往前进行迁移
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;
}
}
下面开始正式迁移，迁移的过程就是从后往前遍历数组的下标。从迁移首先会进入到tabAt(tab,i)==null，表示当前坑位没有元素直接将这个节点设置而为一个叫做 ForwardingNode<K,V>的对象父类node节点，只不过这个节点的hash是一个MOVED=-1。接下在这个过程中如果有一个线程put元素时正好是这个坑位的节点时，就会发现当前坑位中已经有一个节点而且这个节点的hash值是MOVED=-1。put元素的线程就会参与到transfer中。如果坑位不为空的节点，这时就开始进行迁移操作迁移过程跟HashMap迁移类似。只不过这个会对当前有节点的坑位进行上锁迁移。迁移完成之后会把旧数组在当前坑位设置成 ForwardingNode<K,V>节点表示已经迁移完成了。
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
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);
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
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;
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);
setTabAt(tab, i, fwd);
advance = true;
}
helpTransfer的代码：
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;
}
从代码中可以看出，helpTransfer还是需要做很多判断。大if判断table是否为空，当前坑位的节点是否为ForwardingNode，ForwardingNode节点的nextTable是否为空，条件都满足之后，while循环判断ForwardingNode的nextTable是否和当前Map中的nextTable相等，旧的table是否也想定，sizCtl是否还是小于0的三个条件都满足时，进入循环，再来判断是否以前迁移完成就直接break，外面也是一层自旋会将需要put的元素最终put进去。没有完成迁移时，就进入transfer操作。这个时候map的属性sizeCtl就成了记录线程参与transfer的一个凭证。所有的线程都迁移完成之后sizCtl回复原来的作用。
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);//数组扩容的条件
return;
}
网友评论