深入学习java并发包ConcurrentHashMap源码
正文
以前写过介绍hashmap的文章,文中提到过hashmap在put的时候,插入的元素超过了容量(由负载因子决定)的范围就会触发扩容操作,就是rehash,这个会重新将原数组的内容重新hash到新的扩容数组中,在多线程的环境下,存在同时其他的元素也在进行put操作,如果hash值相同,可能出现同时在同一数组下用链表表示,造成闭环,导致在get时会出现死循环,所以hashmap是线程不安全的。
jdk1.7的实现
整个 concurrenthashmap 由一个个 segment 组成,segment 代表”部分“或”一段“的意思,所以很多地方都会将其描述为分段锁。注意,行文中,我很多地方用了“槽”来代表一个 segment。
简单理解就是,concurrenthashmap 是一个 segment 数组,segment 通过继承 reentrantlock 来进行加锁,所以每次需要加锁的操作锁住的是一个 segment,这样只要保证每个 segment 是线程安全的,也就实现了全局的线程安全。
concurrencylevel:并行级别、并发数、segment 数。默认是 16,也就是说 concurrenthashmap 有 16 个 segments,所以理论上,这个时候,最多可以同时支持 16 个线程并发写,只要它们的操作分别分布在不同的 segment 上。这个值可以在初始化的时候设置为其他值,但是一旦初始化以后,它是不可以扩容的。
再具体到每个 segment 内部,其实每个 segment 很像之前介绍的 hashmap,不过它要保证线程安全,所以处理起来要麻烦些。
初始化
initialcapacity:初始容量,这个值指的是整个 concurrenthashmap 的初始容量,实际操作的时候需要平均分给每个 segment。
loadfactor:负载因子,之前我们说了,segment 数组不可以扩容,所以这个负载因子是给每个 segment 内部使用的。
public concurrenthashmap(int initialcapacity, float loadfactor, int concurrencylevel) { if (!(loadfactor > 0) || initialcapacity < 0 || concurrencylevel <= 0) throw new illegalargumentexception(); if (concurrencylevel > max_segments) concurrencylevel = max_segments; // find power-of-two sizes best matching arguments int sshift = 0; int ssize = 1; // 计算并行级别 ssize,因为要保持并行级别是 2 的 n 次方 while (ssize < concurrencylevel) { ++sshift; ssize <<= 1; } // 我们这里先不要那么烧脑,用默认值,concurrencylevel 为 16,sshift 为 4 // 那么计算出 segmentshift 为 28,segmentmask 为 15,后面会用到这两个值 this.segmentshift = 32 - sshift; this.segmentmask = ssize - 1; if (initialcapacity > maximum_capacity) initialcapacity = maximum_capacity; // initialcapacity 是设置整个 map 初始的大小, // 这里根据 initialcapacity 计算 segment 数组中每个位置可以分到的大小 // 如 initialcapacity 为 64,那么每个 segment 或称之为"槽"可以分到 4 个 int c = initialcapacity / ssize; if (c * ssize < initialcapacity) ++c; // 默认 min_segment_table_capacity 是 2,这个值也是有讲究的,因为这样的话,对于具体的槽上, // 插入一个元素不至于扩容,插入第二个的时候才会扩容 int cap = min_segment_table_capacity; while (cap < c) cap <<= 1; // 创建 segment 数组, // 并创建数组的第一个元素 segment[0] segment<k,v> s0 = new segment<k,v>(loadfactor, (int)(cap * loadfactor), (hashentry<k,v>[])new hashentry[cap]); segment<k,v>[] ss = (segment<k,v>[])new segment[ssize]; // 往数组写入 segment[0] unsafe.putorderedobject(ss, sbase, s0); // ordered write of segments[0] this.segments = ss; }
初始化完成,我们得到了一个 segment 数组。
我们就当是用 new concurrenthashmap() 无参构造函数进行初始化的,那么初始化完成后:
- segment 数组长度为 16,不可以扩容
- segment[i] 的默认大小为 2,负载因子是 0.75,得出初始阈值为 1.5,也就是以后插入第一个元素不会触发扩容,插入第二个会进行第一次扩容
- 这里初始化了 segment[0],其他位置还是 null,至于为什么要初始化 segment[0],后面的代码会介绍
- 当前 segmentshift 的值为 32 - 4 = 28,segmentmask 为 16 - 1 = 15,姑且把它们简单翻译为移位数和掩码,这两个值马上就会用到
segment
static class segment<k,v> extends reentrantlock implements serializable { transient volatile hashentry<k,v>[] table; transient int count; transient int modcount; }
从上segment的继承体系可以看出,segment实现了reentrantlock,也就带有锁的功能,table使用volatile修饰,保证了内存可见性。
put 过程分析
我们先看 put 的主流程,对于其中的一些关键细节操作,后面会进行详细介绍。
public v put(k key, v value) { segment<k,v> s; if (value == null) throw new nullpointerexception(); // 1. 计算 key 的 hash 值 int hash = hash(key); // 2. 根据 hash 值找到 segment 数组中的位置 j // hash 是 32 位,无符号右移 segmentshift(28) 位,剩下高 4 位, // 然后和 segmentmask(15) 做一次与操作,也就是说 j 是 hash 值的高 4 位,也就是槽的数组下标 int j = (hash >>> segmentshift) & segmentmask; // 刚刚说了,初始化的时候初始化了 segment[0],但是其他位置还是 null, // ensuresegment(j) 对 segment[j] 进行初始化 if ((s = (segment<k,v>)unsafe.getobject // nonvolatile; recheck (segments, (j << sshift) + sbase)) == null) // in ensuresegment s = ensuresegment(j); // 3. 插入新值到 槽 s 中 return s.put(key, hash, value, false); }
初始化槽: ensuresegment
concurrenthashmap 初始化的时候会初始化第一个槽 segment[0],对于其他槽来说,在插入第一个值的时候进行初始化。
这里需要考虑并发,因为很可能会有多个线程同时进来初始化同一个槽 segment[k],不过只要有一个成功了就可以。
private segment<k,v> ensuresegment(int k) { final segment<k,v>[] ss = this.segments; long u = (k << sshift) + sbase; // raw offset segment<k,v> seg; if ((seg = (segment<k,v>)unsafe.getobjectvolatile(ss, u)) == null) { // 这里看到为什么之前要初始化 segment[0] 了, // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k] // 为什么要用“当前”,因为 segment[0] 可能早就扩容过了 segment<k,v> proto = ss[0]; int cap = proto.table.length; float lf = proto.loadfactor; int threshold = (int)(cap * lf); // 初始化 segment[k] 内部的数组 hashentry<k,v>[] tab = (hashentry<k,v>[])new hashentry[cap]; if ((seg = (segment<k,v>)unsafe.getobjectvolatile(ss, u)) == null) { // 再次检查一遍该槽是否被其他线程初始化了。 segment<k,v> s = new segment<k,v>(lf, threshold, tab); // 使用 while 循环,内部用 cas,当前线程成功设值或其他线程成功设值后,退出,如果其他线程成功设置后,这里获取到直接返回 while ((seg = (segment<k,v>)unsafe.getobjectvolatile(ss, u)) == null) { if (unsafe.compareandswapobject(ss, u, null, seg = s)) break; } } } return seg; }
总的来说,ensuresegment(int k) 比较简单,对于并发操作使用 cas 进行控制。
第一层很简单,根据 hash 值很快就能找到相应的 segment,之后就是 segment 内部的 put 操作了。
segment 内部是由 数组+链表 组成的
final v put(k key, int hash, v value, boolean onlyifabsent) { // 在往该 segment 写入前,需要先获取该 segment 的独占锁 // 先看主流程,后面还会具体介绍这部分内容 hashentry<k,v> node = trylock() ? null : scanandlockforput(key, hash, value); v oldvalue; try { // 这个是 segment 内部的数组 hashentry<k,v>[] tab = table; // 再利用 hash 值,求应该放置的数组下标 int index = (tab.length - 1) & hash; // first 是数组该位置处的链表的表头 hashentry<k,v> first = entryat(tab, index); // 下面这串 for 循环虽然很长,不过也很好理解,想想该位置没有任何元素和已经存在一个链表这两种情况 for (hashentry<k,v> e = first;;) { if (e != null) { k k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldvalue = e.value; if (!onlyifabsent) { // 覆盖旧值 e.value = value; ++modcount; } break; } // 继续顺着链表走 e = e.next; } else { // node 到底是不是 null,这个要看获取锁的过程,不过和这里都没有关系。 // 如果不为 null,那就直接将它设置为链表表头;如果是null,初始化并设置为链表表头。 if (node != null) node.setnext(first); else node = new hashentry<k,v>(hash, key, value, first); int c = count + 1; // 如果超过了该 segment 的阈值,这个 segment 需要扩容 if (c > threshold && tab.length < maximum_capacity) rehash(node); // 扩容后面也会具体分析 else // 没有达到阈值,将 node 放到数组 tab 的 index 位置, // 其实就是将新的节点设置成原链表的表头 setentryat(tab, index, node); ++modcount; count = c; oldvalue = null; break; } } } finally { // 解锁 unlock(); } return oldvalue; }
整体流程还是比较简单的,由于有独占锁的保护,所以 segment 内部的操作并不复杂。至于这里面的并发问题,我们稍后再进行介绍。
到这里 put 操作就结束了,接下来,我们说一说其中几步关键的操作。
获取写入锁: scanandlockforput
前面我们看到,在往某个 segment 中 put 的时候,首先会调用 node = trylock() ? null : scanandlockforput(key, hash, value),也就是说先进行一次 trylock() 快速获取该 segment 的独占锁,如果失败,那么进入到 scanandlockforput 这个方法来获取锁。
下面我们来具体分析这个方法中是怎么控制加锁的。
private hashentry<k,v> scanandlockforput(k key, int hash, v value) { hashentry<k,v> first = entryforhash(this, hash); hashentry<k,v> e = first; hashentry<k,v> node = null; int retries = -1; // negative while locating node // 循环获取锁 while (!trylock()) { hashentry<k,v> f; // to recheck first below if (retries < 0) { if (e == null) { if (node == null) // speculatively create node // 进到这里说明数组该位置的链表是空的,没有任何元素 // 当然,进到这里的另一个原因是 trylock() 失败,所以该槽存在并发,不一定是该位置 node = new hashentry<k,v>(hash, key, value, null); retries = 0; } else if (key.equals(e.key)) retries = 0; else // 顺着链表往下走 e = e.next; } // 重试次数如果超过 max_scan_retries(单核1多核64),那么不抢了,进入到阻塞队列等待锁 // lock() 是阻塞方法,直到获取锁后返回 else if (++retries > max_scan_retries) { lock(); break; } else if ((retries & 1) == 0 && // 这个时候是有大问题了,那就是有新的元素进到了链表,成为了新的表头 // 所以这边的策略是,相当于重新走一遍这个 scanandlockforput 方法 (f = entryforhash(this, hash)) != first) { e = first = f; // re-traverse if entry changed retries = -1; } } return node; }
这个方法有两个出口,一个是 trylock() 成功了,循环终止,另一个就是重试次数超过了 max_scan_retries,进到 lock() 方法,此方法会阻塞等待,直到成功拿到独占锁。
这个方法就是看似复杂,但是其实就是做了一件事,那就是获取该 segment 的独占锁,如果需要的话顺便实例化了一下 node。
获取锁时,并不直接使用lock来获取,因为该方法获取锁失败时会挂起。事实上,它使用了自旋锁,如果trylock获取锁失败,说明锁被其它线程占用,此时通过循环再次以trylock的方式申请锁。如果在循环过程中该key所对应的链表头被修改,则重置retry次数。如果retry次数超过一定值,则使用lock方法申请锁。
这里使用自旋锁是因为自旋锁的效率比较高,但是它消耗cpu资源比较多,因此在自旋次数超过阈值时切换为互斥锁。
扩容: rehash
重复一下,segment 数组不能扩容,扩容是 segment 数组某个位置内部的数组 hashentry\<k,v>[] 进行扩容,扩容后,容量为原来的 2 倍。
首先,我们要回顾一下触发扩容的地方,put 的时候,如果判断该值的插入会导致该 segment 的元素个数超过阈值,那么先进行扩容,再插值,读者这个时候可以回去 put 方法看一眼。
该方法不需要考虑并发,因为到这里的时候,是持有该 segment 的独占锁的。
// 方法参数上的 node 是这次扩容后,需要添加到新的数组中的数据。 private void rehash(hashentry<k,v> node) { hashentry<k,v>[] oldtable = table; int oldcapacity = oldtable.length; // 2 倍 int newcapacity = oldcapacity << 1; threshold = (int)(newcapacity * loadfactor); // 创建新数组 hashentry<k,v>[] newtable = (hashentry<k,v>[]) new hashentry[newcapacity]; // 新的掩码,如从 16 扩容到 32,那么 sizemask 为 31,对应二进制 ‘000...00011111' int sizemask = newcapacity - 1; // 遍历原数组,老套路,将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldcap 两个位置 for (int i = 0; i < oldcapacity ; i++) { // e 是链表的第一个元素 hashentry<k,v> e = oldtable[i]; if (e != null) { hashentry<k,v> next = e.next; // 计算应该放置在新数组中的位置, // 假设原数组长度为 16,e 在 oldtable[3] 处,那么 idx 只可能是 3 或者是 3 + 16 = 19 int idx = e.hash & sizemask; if (next == null) // 该位置处只有一个元素,那比较好办 newtable[idx] = e; else { // reuse consecutive sequence at same slot // e 是链表表头 hashentry<k,v> lastrun = e; // idx 是当前链表的头结点 e 的新位置 int lastidx = idx; // 下面这个 for 循环会找到一个 lastrun 节点,这个节点之后的所有元素是将要放到一起的 for (hashentry<k,v> last = next; last != null; last = last.next) { int k = last.hash & sizemask; if (k != lastidx) { lastidx = k; lastrun = last; } } // 将 lastrun 及其之后的所有节点组成的这个链表放到 lastidx 这个位置 newtable[lastidx] = lastrun; // 下面的操作是处理 lastrun 之前的节点, // 这些节点可能分配在另一个链表中,也可能分配到上面的那个链表中 for (hashentry<k,v> p = e; p != lastrun; p = p.next) { v v = p.value; int h = p.hash; int k = h & sizemask; hashentry<k,v> n = newtable[k]; newtable[k] = new hashentry<k,v>(h, p.key, v, n); } } } } // 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部 int nodeindex = node.hash & sizemask; // add the new node node.setnext(newtable[nodeindex]); newtable[nodeindex] = node; table = newtable; }
总结一下put的流程:
当执行put操作时,会进行第一次key的hash来定位segment的位置,如果该segment还没有初始化,即通过cas操作进行赋值,然后进行第二次hash操作,找到相应的hashentry的位置,这里会利用继承过来的锁的特性,在将数据插入指定的hashentry位置时(链表的尾端),会通过继承reentrantlock的trylock()方法尝试去获取锁,如果获取成功就直接插入相应的位置,如果已经有线程获取该segment的锁,那当前线程会以自旋的方式去继续的调用trylock()方法去获取锁,超过指定次数就挂起,等待唤醒。
get 过程分析
相对于 put 来说,get 真的不要太简单。
1.计算 hash 值,找到 segment 数组中的具体位置,或我们前面用的“槽”
2.槽中也是一个数组,根据 hash 找到数组中具体的位置
3.到这里是链表了,顺着链表进行查找即可
public v get(object key) { segment<k,v> s; // manually integrate access methods to reduce overhead hashentry<k,v>[] tab; // 1. hash 值 int h = hash(key); long u = (((h >>> segmentshift) & segmentmask) << sshift) + sbase; // 2. 根据 hash 找到对应的 segment if ((s = (segment<k,v>)unsafe.getobjectvolatile(segments, u)) != null && (tab = s.table) != null) { // 3. 找到segment 内部数组相应位置的链表,遍历 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; }
size操作
put、remove和get操作只需要关心一个segment,而size操作需要遍历所有的segment才能算出整个map的大小。一个简单的方案是,先锁住所有sgment,计算完后再解锁。但这样做,在做size操作时,不仅无法对map进行写操作,同时也无法进行读操作,不利于对map的并行操作。
为更好支持并发操作,concurrenthashmap会在不上锁的前提逐个segment计算3次size,如果某相邻两次计算获取的所有segment的更新次数(每个segment都与hashmap一样通过modcount跟踪自己的修改次数,segment每修改一次其modcount加一)相等,说明这两次计算过程中无更新操作,则这两次计算出的总size相等,可直接作为最终结果返回。如果这三次计算过程中map有更新,则对所有segment加锁重新计算size。该计算方法代码如下
public int size() { final segment<k,v>[] segments = this.segments; int size; boolean overflow; // true if size overflows 32 bits long sum; // sum of modcounts long last = 0l; // previous sum int retries = -1; // first iteration isn't retry try { for (;;) { if (retries++ == retries_before_lock) { for (int j = 0; j < segments.length; ++j) ensuresegment(j).lock(); // force creation } sum = 0l; size = 0; overflow = false; for (int j = 0; j < segments.length; ++j) { segment<k,v> seg = segmentat(segments, j); if (seg != null) { sum += seg.modcount; int c = seg.count; if (c < 0 || (size += c) < 0) overflow = true; } } if (sum == last) break; last = sum; } } finally { if (retries > retries_before_lock) { for (int j = 0; j < segments.length; ++j) segmentat(segments, j).unlock(); } } return overflow ? integer.max_value : size; }
concurrenthashmap的size方法是一个嵌套循环,大体逻辑如下:
1.遍历所有的segment。
2.把segment的元素数量累加起来。
3.把segment的修改次数累加起来。
4.判断所有segment的总修改次数是否大于上一次的总修改次数。如果大于,说明统计过程中有修改,重新统计,尝试次数+1;如果不是。说明没有修改,统计结束。
5.如果尝试次数超过阈值,则对每一个segment加锁,再重新统计。
6.再次判断所有segment的总修改次数是否大于上一次的总修改次数。由于已经加锁,次数一定和上次相等。
7.释放锁,统计结束。
并发问题分析
现在我们已经说完了 put 过程和 get 过程,我们可以看到 get 过程中是没有加锁的,那自然我们就需要去考虑并发问题。
添加节点的操作 put 和删除节点的操作 remove 都是要加 segment 上的独占锁的,所以它们之间自然不会有问题,我们需要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操作。
put 操作的线程安全性
- 初始化槽,这个我们之前就说过了,使用了 cas 来初始化 segment 中的数组。
- 添加节点到链表的操作是插入到表头的,所以,如果这个时候 get 操作在链表遍历的过程已经到了中间,是不会影响的。当然,另一个并发问题就是 get 操作在 put 之后,需要保证刚刚插入表头的节点被读取,这个依赖于 setentryat 方法中使用的 unsafe.putorderedobject。
- 扩容。扩容是新创建了数组,然后进行迁移数据,最后面将 newtable 设置给属性 table。所以,如果 get 操作此时也在进行,那么也没关系,如果 get 先行,那么就是在旧的 table 上做查询操作;而 put 先行,那么 put 操作的可见性保证就是 table 使用了 volatile 关键字。
remove 操作的线程安全性
- remove 操作我们没有分析源码,所以这里说的读者感兴趣的话还是需要到源码中去求实一下的。
- get 操作需要遍历链表,但是 remove 操作会"破坏"链表。
- 如果 remove 破坏的节点 get 操作已经过去了,那么这里不存在任何问题。
- 如果 remove 先破坏了一个节点,分两种情况考虑。 1、如果此节点是头结点,那么需要将头结点的 next 设置为数组该位置的元素,table 虽然使用了 volatile 修饰,但是 volatile 并不能提供数组内部操作的可见性保证,所以源码中使用了 unsafe 来操作数组,请看方法 setentryat。2、如果要删除的节点不是头结点,它会将要删除节点的后继节点接到前驱节点中,这里的并发保证就是 next 属性是 volatile 的。
最后我们来看看并发操作示意图
case1:不同segment的并发写入
不同segment的写入是可以并发执行的。
case2:同一segment的一写一读
同一segment的写和读是可以并发执行的。
case3:同一segment的并发写入
segment的写入是需要上锁的,因此对同一segment的并发写入会被阻塞。
由此可见,concurrenthashmap当中每个segment各自持有一把锁。在保证线程安全的同时降低了锁的粒度,让并发操作效率更高。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
推荐阅读
-
深入学习java并发包ConcurrentHashMap源码
-
java深入学习Hashmap源码(JDK8)
-
java深入学习ConcurrentHashmap (JDK8)
-
java深入学习Hashmap源码(JDK8)
-
java深入学习ConcurrentHashmap (JDK8)
-
java集合 ArrayDeque原理, 通过源码学习进行深入了解
-
Java并发编程—ConcurrentHashMap源码学习
-
java集合 PriorityQueue原理, 通过源码学习进行深入了解
-
深入学习java ThreadLocal的源码知识
-
深入并发包 ConcurrentHashMap 源码解析