清屏死光,永城职业技术学院,小海狗资源搜索引擎
正文
以前写过介绍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
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 操作的线程安全性
remove 操作的线程安全性
最后我们来看看并发操作示意图
case1：不同segment的并发写入
不同segment的写入是可以并发执行的。
case2：同一segment的一写一读
同一segment的写和读是可以并发执行的。
case3：同一segment的并发写入
segment的写入是需要上锁的，因此对同一segment的并发写入会被阻塞。
由此可见，concurrenthashmap当中每个segment各自持有一把锁。在保证线程安全的同时降低了锁的粒度，让并发操作效率更高。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持移动技术网。
如对本文有疑问，请在下面进行留言讨论，广大热心网友会与你互动！！ 点击进行留言回复
浅析我对 String、StringBuilder、StringBuffer 的理解
使用IDEA搭建SSM框架的详细教程(spring + springMVC +MyBatis)
Springboot整合freemarker 404问题解决方案
引入mybatis-plus报 Invalid bound statement错误问题的解决方法
网友评论