ConcurrentHashMap源码分析
1、它实现了concurrentmap
接口,该接口定义了一些原子操作约定
2、线程安全
- 完全的并发读和高并发写
- 读操作完全无锁,牺牲了一致性;写操作部分有锁
- 它与
hashtable
、collections.synchronizedmap
-
hashmap
支持null
,concurrenthashmap
、hashtable
不支持null
3、java7
- 分段锁
- 哈希表/链表
4、java8
-
cas
+unsafe
- 哈希表/链表 + 红黑树
java7
的实现
一、相关概念
1、分段锁
concurrenthashmap
底层采用多个分段segment
,每段下面都是一个哈希表,这就是分段。每当需要对每段数据上锁操作时,只需要对segment
上锁即可,这就是分段锁。通常称segment
的数量叫做并发度concurrency
。
优点:
- 在未上锁的情况下,提高了并发度;
2、并发度concurrency
/** * the default concurrency level for this table, used when not * otherwise specified in a constructor. */ static final int default_concurrency_level = 16;
这表示默认情况下,会有16个段
segment
3、每个segment
的哈希表长度都是2的幂次方
在concurrenthashmap
构造方法中
二、源码分析
1、get
方法
- 计算
segment
的位置 - 找到这个段下面的哈希表
- 遍历链表,看是否存在
public v get(object key) { segment<k,v> s; // manually integrate access methods to reduce overhead hashentry<k,v>[] tab; int h = hash(key); // 获取到key所在segment数组的下标 long u = (((h >>> segmentshift) & segmentmask) << sshift) + sbase; // 判断这个下标是否存在,以及segment下面的哈希表是否存在 if ((s = (segment<k,v>)unsafe.getobjectvolatile(segments, u)) != null && (tab = s.table) != null) { // 熟悉的:(tab.length - 1) & h操作 for (hashentry<k,v> e = (hashentry<k,v>) unsafe.getobjectvolatile (tab, ((long)(((tab.length - 1) & h)) << tshift) + tbase); e != null; e = e.next) { k k; if ((k = e.key) == key || (e.hash == h && key.equals(k))) return e.value; } } return null; }
(1)为什么要使用unsafe.getobjectvolatile(segments, u)
这种方式来读取数组下标的某个元素?
提高性能。使用常用segments[i]
这种语法,在编译字节码的时候,是会检查数组是否越界;而使用上面的代码,会节省这一步。
(2)如何保证线程安全性?
即如何保证在多线程环境下,当线程在做更新操作时,如果其他线程在同步读的话,是可能出现脏数据、空指针情况。那么concurrenthashmap
是如何保证的?concurrenthashmap
为了提高高并发,而牺牲了一致性,但这种一致性是弱一致性,不会对程序造成大的过错。所以脏数据是无法避免的,因此在java8
的类注释写到不建议使用size
、isempty
、containsvalue
来进行判断语句。
* bear in mind that the results of aggregate status methods including * {@code size}, {@code isempty}, and {@code containsvalue} are typically * useful only when a map is not undergoing concurrent updates in other threads. * otherwise the results of these methods reflect transient states * that may be adequate for monitoring or estimation purposes, but not * for program control.
2、put
方法
- 找到
segment
,必要时新建; -
segment
执行put
操作,必要时扩容;
public v put(k key, v value) { segment<k,v> s; if (value == null) throw new nullpointerexception(); int hash = hash(key); int j = (hash >>> segmentshift) & segmentmask; if ((s = (segment<k,v>)unsafe.getobject // nonvolatile; recheck (segments, (j << sshift) + sbase)) == null) // in ensuresegment s = ensuresegment(j); return s.put(key, hash, value, false); }
(1)扩容时如何保证线程安全性?
- 在创建
segment
时,采用cas
保证线程安全性; - 在创建
entry
时,因为segment
本身就是reentrantlock
,在其segment.put()
方法是一定保证在获取到锁的情况下才执行操作的;
(2)unsafe.getobject()
的作用?
java8
的实现
一、与java7
的改进
使用哈希表 + 链表/红黑树 的数据结构
放弃使用分段锁,改用cas
、volatile
、unsafe
java7
的分段锁很好,但锁毕竟还是很慢的,所以java8
实现了尽可能地无锁环境。
这里所说地无锁也仅仅大多数情况下,在某些特殊场景还是需要锁地。
锁的粒度更细
java7
锁地粒度是segment
,而在java8
中锁地粒度是每个entry
二、源码分析
1、get
方法
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) { // 如果第一个就找到,直接返回 if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } // 如果元素地hash值小于0,就往红黑树查找 else if (eh < 0) return (p = e.find(h, key)) != null ? p.val : null; // 链表下地查找 while ((e = e.next) != null) { if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; }
(1)查找没有锁,如何有人在写入怎么办?
- 在红黑树状态下,查找是有读写锁;
- 在链表状态下,跟
java7
相似,牺牲了弱一致性;
2、put
方法
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; // 如果哈希表为空,就新建 if (tab == null || (n = tab.length) == 0) tab = inittable(); // 找到对应下标entry,如果为空,就新建 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); // 在对应entry下,进行put操作 else { v oldval = null; // synchronized锁定entry,进行put synchronized (f) { if (tabat(tab, i) == f) { // 链表地put操作 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; } } } // 红黑树地put操作 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; }
下一篇: Spring框架入门(一)