ConcurrentHashMap线程安全：分段锁 到 synchronized + CAS

专栏系列文章地址：https://blog.csdn.net/qq_26437925/article/details/145290162

本文目标：

1. 理解ConcurrentHashMap为什么线程安全；ConcurrentHashMap的具体细节还需要进一步研究

ConcurrentHashMap介绍

百度AI介绍如下：

JDK7的分段锁实现

在 JDK7 中，ConcurrentHashMap 使用“分段锁”机制实现线程安全，数据结构可以看成是”Segment数组+HashEntry数组+链表”，一个 ConcurrentHashMap 实例中包含若干个 Segment 实例组成的数组，每个 Segment 实例又包含由若干个桶，每个桶中都是由若干个 HashEntry 对象链接起来的链表。

Segment 类继承 ReentrantLock 类，锁的粒度为其中一个Segment，而不是整体。

JDK8的synchronized + CAS实现

每个桶可能是链表结构或者红黑树结构，锁针对桶的头节点加，锁粒度小

put方法

定位Node数组位置使用CAS操作定位，真正进行插入操作的时候会使用synchronized关键字加锁头部

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {// key, value 都不能为nullif (key == null || value == null) throw new NullPointerException();int hash = spread(key.hashCode());int binCount = 0;for (Node<K,V>[] tab = table;;) {Node<K,V> f; int n, i, fh;// 如果Node数组是空，则进行初始化；初始化是CAS操作if (tab == null || (n = tab.length) == 0)tab = initTable();else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 数组位置节点为null，则CAS方式进行添加Node到数组位置if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))break;                   // no lock when adding to empty bin}else if ((fh = f.hash) == MOVED)// 如果数组位置节点正在迁移，则帮助迁移tab = helpTransfer(tab, f);else {// 没有迁移，且数组位置不是空，则进行聊表或者红黑树的插入操作，可能涉及到链表转红黑树V oldVal = null;// 直接用 synchronized 锁住 链表或者红黑树的头部synchronized (f) {if (tabAt(tab, i) == f) {// 链表遍历判断，替换老值，或者进行尾插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;}}}// 红黑树替换老值，或者进行红黑树插入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) {if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}addCount(1L, binCount);return null;
}

put完成后addCount(1L, binCount);会进行数量统计和扩容判断操作，也是CAS操作

/*** 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*/
private final void addCount(long x, int check) {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;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();}}
}