早期ConcurrentHashMap:
通过分段锁Segment实现,将锁一段一段存储,默认会分配16个segment,当一个线程占用一把锁segment访问其中一段数据的时候,位于其它segment的数据也能被其它线程同时访问,每个segment分配子数组,将hashmap数组逻辑上拆分成多个子数组,每个子数组配置一把锁,线程在获取到某把分段锁的时候才能操作该子数组,其它线程想要操作该子数组时只能被阻塞。
ConcurrentHashMap比起Segment,锁拆的更细;
CAS + synchronized使锁更细化;
synchronized只锁当前树或者链表的首节点,这样哈希不冲突就不会产生并发;
- 使用无锁操作CAS插入头节点,失败则循环重试
- 如果头节点已经存在,则尝试获取头节点的同步锁,再进行操作
对比HashMap,HashTable: - HashMap线程不安全,数组+链表+红黑树
- HashTable线程安全,锁住整个对象,数组+链表,key和value都不能为null
- ConcurrentHashMap线程安全,将锁细粒度化到table的每个元素来提示并发性能,CAS+同步锁,数组+链表+红黑树
final V putVal(K key, V value, boolean onlyIfAbsent) {
//key和value都不能为null
if (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[]数组是否初始化,没有则进行初始化操作
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//通过hash定位数组的索引坐标,看看是否有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
}
//检查到内部正在扩容就帮助它一块扩容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//如果f不为null,就锁住f元素(链表/红黑二叉树的头元素)
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) {
//如果是TreeNode则执行树添加操作
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;
}