并发容器之ConcurrentHashMap(JDK 1.8版本)
1.concurrenthashmap简介
在使用hashmap时在多线程情况下扩容会出现cpu接近100%的情况，因为hashmap并不是线程安全的，通常我们可以使用在java体系中古老的hashtable类，该类基本上所有的方法都采用synchronized进行线程安全的控制，可想而知，在高并发的情况下，每次只有一个线程能够获取对象监视器锁，这样的并发性能的确不令人满意。另外一种方式通过collections的map<k,v> synchronizedmap(map<k,v> m)将hashmap包装成一个线程安全的map。比如synchronzedmap的put方法源码为：
public v put(k key, v value) {
synchronized (mutex) {return m.put(key, value);}
}
实际上synchronizedmap实现依然是采用synchronized独占式锁进行线程安全的并发控制的。同样，这种方案的性能也是令人不太满意的。针对这种境况，doug lea大师不遗余力的为我们创造了一些线程安全的并发容器，让每一个java开发人员倍感幸福。相对于hashmap来说，concurrenthashmap就是线程安全的map，其中利用了锁分段的思想提高了并发度。
concurrenthashmap在jdk1.6的版本网上资料很多，有兴趣的可以去看看。 jdk 1.6版本关键要素：
- segment继承了reentrantlock充当锁的角色，为每一个segment提供了线程安全的保障；
- segment维护了哈希散列表的若干个桶，每个桶由hashentry构成的链表。
而到了jdk 1.8的concurrenthashmap就有了很大的变化，光是代码量就足足增加了很多。1.8版本舍弃了segment，并且大量使用了synchronized，以及cas无锁操作以保证concurrenthashmap操作的线程安全性。
至于为什么不用reentrantlock而是synchronzied呢？实际上，synchronzied做了很多的优化，包括偏向锁，轻量级锁，重量级锁，可以依次向上升级锁状态，但不能降级，因此，使用synchronized相较于reentrantlock的性能会持平甚至在某些情况更优，具体的性能测试可以去网上查阅一些资料。另外，底层数据结构改变为采用数组+链表+红黑树的数据形式。
2.关键属性及类
在了解concurrenthashmap的具体方法实现前，我们需要系统的来看一下几个关键的地方。
concurrenthashmap的关键属性
- table volatile node<k,v>[] table://装载node的数组，作为concurrenthashmap的数据容器，采用懒加载的方式，直到第一次插入数据的时候才会进行初始化操作，数组的大小总是为2的幂次方。
- nexttable volatile node<k,v>[] nexttable; //扩容时使用，平时为null，只有在扩容的时候才为非null
- sizectl volatile int sizectl; 该属性用来控制table数组的大小，根据是否初始化和是否正在扩容有几种情况： 当值为负数时：如果为-1表示正在初始化，如果为-n则表示当前正有n-1个线程进行扩容操作； 当值为正数时：如果当前数组为null的话表示table在初始化过程中，sizectl表示为需要新建数组的长度；
若已经初始化了，表示当前数据容器（table数组）可用容量也可以理解成临界值（插入节点数超过了该临界值就需要扩容），具体指为数组的长度n 乘以 加载因子loadfactor； 当值为0时，即数组长度为默认初始值。
- sun.misc.unsafe u 在concurrenthashmapde的实现中可以看到大量的u.compareandswapxxxx的方法去修改concurrenthashmap的一些属性。这些方法实际上是利用了cas算法保证了线程安全性，这是一种乐观策略，假设每一次操作都不会产生冲突，当且仅当冲突发生的时候再去尝试。
而cas操作依赖于现代处理器指令集，通过底层cmpxchg指令实现。cas(v,o,n)核心思想为：若当前变量实际值v与期望的旧值o相同，则表明该变量没被其他线程进行修改，因此可以安全的将新值n赋值给变量；若当前变量实际值v与期望的旧值o不相同，则表明该变量已经被其他线程做了处理，此时将新值n赋给变量操作就是不安全的，在进行重试。
而在大量的同步组件和并发容器的实现中使用cas是通过sun.misc.unsafe类实现的，该类提供了一些可以直接操控内存和线程的底层操作，可以理解为java中的“指针”。该成员变量的获取是在静态代码块中：
```
static {
try {
u = sun.misc.unsafe.getunsafe();
.......
} catch (exception e) {
throw new error(e);
}
}
```
concurrenthashmap中关键内部类
-
node node类实现了map.entry接口，主要存放key-value对，并且具有next域
static class node<k,v> implements map.entry<k,v> { final int hash; final k key; volatile v val; volatile node<k,v> next; ...... }
另外可以看出很多属性都是用volatile进行修饰的，也就是为了保证内存可见性。
-
treenode 树节点，继承于承载数据的node类。而红黑树的操作是针对treebin类的，从该类的注释也可以看出，也就是treebin会将treenode进行再一次封装
** * nodes for use in treebins */ static final class treenode<k,v> extends node<k,v> { treenode<k,v> parent; // red-black tree links treenode<k,v> left; treenode<k,v> right; treenode<k,v> prev; // needed to unlink next upon deletion boolean red; ...... } -
treebin 这个类并不负责包装用户的key、value信息，而是包装的很多treenode节点。实际的concurrenthashmap“数组”中，存放的是treebin对象，而不是treenode对象。
static final class treebin<k,v> extends node<k,v> { treenode<k,v> root; volatile treenode<k,v> first; volatile thread waiter; volatile int lockstate; // values for lockstate static final int writer = 1; // set while holding write lock static final int waiter = 2; // set when waiting for write lock static final int reader = 4; // increment value for setting read lock ...... } -
forwardingnode 在扩容时才会出现的特殊节点，其key,value,hash全部为null。并拥有nexttable指针引用新的table数组。
static final class forwardingnode<k,v> extends node<k,v> { final node<k,v>[] nexttable; forwardingnode(node<k,v>[] tab) { super(moved, null, null, null); this.nexttable = tab; } ..... }
cas关键操作
在上面我们提及到在concurrenthashmap中会大量使用cas修改它的属性和一些操作。因此，在理解concurrenthashmap的方法前我们需要了解下面几个常用的利用cas算法来保障线程安全的操作。
-
tabat
static final <k,v> node<k,v> tabat(node<k,v>[] tab, int i) { return (node<k,v>)u.getobjectvolatile(tab, ((long)i << ashift) + abase); }
该方法用来获取table数组中索引为i的node元素。
-
castabat
static final <k,v> boolean castabat(node<k,v>[] tab, int i, node<k,v> c, node<k,v> v) { return u.compareandswapobject(tab, ((long)i << ashift) + abase, c, v); }
利用cas操作设置table数组中索引为i的元素 -
settabat
static final <k,v> void settabat(node<k,v>[] tab, int i, node<k,v> v) { u.putobjectvolatile(tab, ((long)i << ashift) + abase, v); }
该方法用来设置table数组中索引为i的元素
3.重点方法讲解
在熟悉上面的这核心信息之后，我们接下来就来依次看看几个常用的方法是怎样实现的。
3.1 实例构造器方法
在使用concurrenthashmap第一件事自然而然就是new 出来一个concurrenthashmap对象，一共提供了如下几个构造器方法：
// 1\. 构造一个空的map，即table数组还未初始化，初始化放在第一次插入数据时，默认大小为16
concurrenthashmap()
// 2\. 给定map的大小
concurrenthashmap(int initialcapacity)
// 3\. 给定一个map
concurrenthashmap(map<? extends k, ? extends v> m)
// 4\. 给定map的大小以及加载因子
concurrenthashmap(int initialcapacity, float loadfactor)
// 5\. 给定map大小，加载因子以及并发度（预计同时操作数据的线程）
concurrenthashmap(int initialcapacity,float loadfactor, int concurrencylevel)
concurrenthashmap一共给我们提供了5中构造器方法，具体使用请看注释，我们来看看第2种构造器，传入指定大小时的情况，该构造器源码为：
public concurrenthashmap(int initialcapacity) {
//1\. 小于0直接抛异常
if (initialcapacity < 0)
throw new illegalargumentexception();
//2\. 判断是否超过了允许的最大值，超过了话则取最大值，否则再对该值进一步处理
int cap = ((initialcapacity >= (maximum_capacity >>> 1)) ?
maximum_capacity :
tablesizefor(initialcapacity + (initialcapacity >>> 1) + 1));
//3\. 赋值给sizectl
this.sizectl = cap;
}
这段代码的逻辑请看注释，很容易理解，如果小于0就直接抛出异常，如果指定值大于了所允许的最大值的话就取最大值，否则，在对指定值做进一步处理。最后将cap赋值给sizectl,关于sizectl的说明请看上面的说明，当调用构造器方法之后，sizectl的大小应该就代表了concurrenthashmap的大小，即table数组长度。tablesizefor做了哪些事情了？源码为：
/**
* returns a power of two table size for the given desired capacity.
* see hackers delight, sec 3.2
*/
private static final int tablesizefor(int c) {
int n = c - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= maximum_capacity) ? maximum_capacity : n + 1;
}
通过注释就很清楚了，该方法会将调用构造器方法时指定的大小转换成一个2的幂次方数，也就是说concurrenthashmap的大小一定是2的幂次方，比如，当指定大小为18时，为了满足2的幂次方特性，实际上concurrenthashmapd的大小为2的5次方（32）。另外，需要注意的是，调用构造器方法的时候并未构造出table数组（可以理解为concurrenthashmap的数据容器），只是算出table数组的长度，当第一次向concurrenthashmap插入数据的时候才真正的完成初始化创建table数组的工作。
3.2 inittable方法
直接上源码：
private final node<k,v>[] inittable() {
node<k,v>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizectl) < 0)
// 1\. 保证只有一个线程正在进行初始化操作
thread.yield(); // lost initialization race; just spin
else if (u.compareandswapint(this, sizectl, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
// 2\. 得出数组的大小
int n = (sc > 0) ? sc : default_capacity;
@suppresswarnings("unchecked")
// 3\. 这里才真正的初始化数组
node<k,v>[] nt = (node<k,v>[])new node<?,?>[n];
table = tab = nt;
// 4\. 计算数组中可用的大小：实际大小n*0.75（加载因子）
sc = n - (n >>> 2);
}
} finally {
sizectl = sc;
}
break;
}
}
return tab;
}
代码的逻辑请见注释，有可能存在一个情况是多个线程同时走到这个方法中，为了保证能够正确初始化，在第1步中会先通过if进行判断，若当前已经有一个线程正在初始化即sizectl值变为-1，这个时候其他线程在if判断为true从而调用thread.yield()让出cpu时间片。正在进行初始化的线程会调用u.compareandswapint方法将sizectl改为-1即正在初始化的状态。
另外还需要注意的事情是，在第四步中会进一步计算数组中可用的大小即为数组实际大小n乘以加载因子0.75.可以看看这里乘以0.75是怎么算的，0.75为四分之三，这里n - (n >>> 2)是不是刚好是n-(1/4)n=(3/4)n，挺有意思的吧:)。如果选择是无参的构造器的话，这里在new node数组的时候会使用默认大小为default_capacity（16），然后乘以加载因子0.75为12，也就是说数组的可用大小为12。
3.3 put方法
使用concurrenthashmap最长用的也应该是put和get方法了吧，我们先来看看put方法是怎样实现的。调用put方法时实际具体实现是putval方法，源码如下：
/** implementation for put and putifabsent */
final v putval(k key, v value, boolean onlyifabsent) {
if (key == null || value == null) throw new nullpointerexception();
//1\. 计算key的hash值
int hash = spread(key.hashcode());
int bincount = 0;
for (node<k,v>[] tab = table;;) {
node<k,v> f; int n, i, fh;
//2\. 如果当前table还没有初始化先调用inittable方法将tab进行初始化
if (tab == null || (n = tab.length) == 0)
tab = inittable();
//3\. tab中索引为i的位置的元素为null，则直接使用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
}
//4\. 当前正在扩容
else if ((fh = f.hash) == moved)
tab = helptransfer(tab, f);
else {
v oldval = null;
synchronized (f) {
if (tabat(tab, i) == f) {
//5\. 当前为链表，在链表中插入新的键值对
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;
}
}
}
// 6.当前为红黑树，将新的键值对插入到红黑树中
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;
}
}
}
}
// 7.插入完键值对后再根据实际大小看是否需要转换成红黑树
if (bincount != 0) {
if (bincount >= treeify_threshold)
treeifybin(tab, i);
if (oldval != null)
return oldval;
break;
}
}
}
//8.对当前容量大小进行检查，如果超过了临界值（实际大小*加载因子）就需要扩容
addcount(1l, bincount);
return null;
}
put方法的代码量有点长，我们按照上面的分解的步骤一步步来看。从整体而言，为了解决线程安全的问题，concurrenthashmap使用了synchronzied和cas的方式。在之前了解过hashmap以及1.8版本之前的concurrenhashmap都应该知道concurrenthashmap结构图，为了方面下面的讲解这里先直接给出，如果对这有疑问的话，可以在网上随便搜搜即可。
[图片上传中...(image-326780-1575107646328-1)]
<figcaption></figcaption>
如图（图片摘自网络），concurrenthashmap是一个哈希桶数组，如果不出现哈希冲突的时候，每个元素均匀的分布在哈希桶数组中。当出现哈希冲突的时候，是标准的链地址的解决方式，将hash值相同的节点构成链表的形式，称为“拉链法”，另外，在1.8版本中为了防止拉链过长，当链表的长度大于8的时候会将链表转换成红黑树。table数组中的每个元素实际上是单链表的头结点或者红黑树的根节点。当插入键值对时首先应该定位到要插入的桶，即插入table数组的索引i处。那么，怎样计算得出索引i呢？当然是根据key的hashcode值。
- spread()重哈希，以减小hash冲突
我们知道对于一个hash表来说，hash值分散的不够均匀的话会大大增加哈希冲突的概率，从而影响到hash表的性能。因此通过spread方法进行了一次重hash从而大大减小哈希冲突的可能性。spread方法为：
static final int spread(int h) {
return (h ^ (h >>> 16)) & hash_bits;
}
该方法主要是将key的hashcode的低16位于高16位进行异或运算，这样不仅能够使得hash值能够分散能够均匀减小hash冲突的概率，另外只用到了异或运算，在性能开销上也能兼顾，做到平衡的trade-off。
2.初始化table
紧接着到第2步，会判断当前table数组是否初始化了，没有的话就调用inittable进行初始化，该方法在上面已经讲过了。
3.能否直接将新值插入到table数组中
从上面的结构示意图就可以看出存在这样一种情况，如果插入值待插入的位置刚好所在的table数组为null的话就可以直接将值插入即可。那么怎样根据hash确定在table中待插入的索引i呢？很显然可以通过hash值与数组的长度取模操作，从而确定新值插入到数组的哪个位置。而之前我们提过concurrenthashmap的大小总是2的幂次方，(n - 1) & hash运算等价于对长度n取模，也就是hash%n，但是位运算比取模运算的效率要高很多，doug lea大师在设计并发容器的时候也是将性能优化到了极致，令人钦佩。
确定好数组的索引i后，就可以可以tabat()方法（该方法在上面已经说明了，有疑问可以回过头去看看）获取该位置上的元素，如果当前node f为null的话，就可以直接用castabat方法将新值插入即可。
4.当前是否正在扩容
如果当前节点不为null，且该节点为特殊节点（forwardingnode）的话，就说明当前concurrenthashmap正在进行扩容操作，关于扩容操作，下面会作为一个具体的方法进行讲解。那么怎样确定当前的这个node是不是特殊的节点了？是通过判断该节点的hash值是不是等于-1（moved）,代码为(fh = f.hash) == moved，对moved的解释在源码上也写的很清楚了：
static final int moved = -1; // hash for forwarding nodes
5.当table[i]为链表的头结点，在链表中插入新值
在table[i]不为null并且不为forwardingnode时，并且当前node f的hash值大于0（fh >= 0）的话说明当前节点f为当前桶的所有的节点组成的链表的头结点。那么接下来，要想向concurrenthashmap插入新值的话就是向这个链表插入新值。通过synchronized (f)的方式进行加锁以实现线程安全性。往链表中插入节点的部分代码为：
if (fh >= 0) {
bincount = 1;
for (node<k,v> e = f;; ++bincount) {
k ek;
// 找到hash值相同的key,覆盖旧值即可
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;
}
}
}
这部分代码很好理解，就是两种情况：1. 在链表中如果找到了与待插入的键值对的key相同的节点，就直接覆盖即可；2. 如果直到找到了链表的末尾都没有找到的话，就直接将待插入的键值对追加到链表的末尾即可
6.当table[i]为红黑树的根节点，在红黑树中插入新值
按照之前的数组+链表的设计方案，这里存在一个问题，即使负载因子和hash算法设计的再合理，也免不了会出现拉链过长的情况，一旦出现拉链过长，甚至在极端情况下，查找一个节点会出现时间复杂度为o(n)的情况，则会严重影响concurrenthashmap的性能，于是，在jdk1.8版本中，对数据结构做了进一步的优化，引入了红黑树。而当链表长度太长（默认超过8）时，链表就转换为红黑树，利用红黑树快速增删改查的特点提高concurrenthashmap的性能，其中会用到红黑树的插入、删除、查找等算法。当table[i]为红黑树的树节点时的操作为：
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中通过f instanceof treebin判断当前table[i]是否是树节点，这下也正好验证了我们在最上面介绍时说的treebin会对treenode做进一步封装，对红黑树进行操作的时候针对的是treebin而不是treenode。这段代码很简单，调用puttreeval方法完成向红黑树插入新节点，同样的逻辑，如果在红黑树中存在于待插入键值对的key相同（hash值相等并且equals方法判断为true）的节点的话，就覆盖旧值，否则就向红黑树追加新节点。
7.根据当前节点个数进行调整
当完成数据新节点插入之后，会进一步对当前链表大小进行调整，这部分代码为：
if (bincount != 0) {
if (bincount >= treeify_threshold)
treeifybin(tab, i);
if (oldval != null)
return oldval;
break;
}
很容易理解，如果当前链表节点个数大于等于8（treeify_threshold）的时候，就会调用treeifybin方法将tabel[i]（第i个散列桶）拉链转换成红黑树。
至此，关于put方法的逻辑就基本说的差不多了，现在来做一些总结：
整体流程：
- 首先对于每一个放入的值，首先利用spread方法对key的hashcode进行一次hash计算，由此来确定这个值在 table中的位置；
- 如果当前table数组还未初始化，先将table数组进行初始化操作；
- 如果这个位置是null的，那么使用cas操作直接放入；
- 如果这个位置存在结点，说明发生了hash碰撞，首先判断这个节点的类型。如果该节点fh==moved(代表forwardingnode,数组正在进行扩容)的话，说明正在进行扩容；
- 如果是链表节点（fh>0）,则得到的结点就是hash值相同的节点组成的链表的头节点。需要依次向后遍历确定这个新加入的值所在位置。如果遇到key相同的节点，则只需要覆盖该结点的value值即可。否则依次向后遍历，直到链表尾插入这个结点；
- 如果这个节点的类型是treebin的话，直接调用红黑树的插入方法进行插入新的节点；
- 插入完节点之后再次检查链表长度，如果长度大于8，就把这个链表转换成红黑树；
- 对当前容量大小进行检查，如果超过了临界值（实际大小*加载因子）就需要扩容。
3.4 get方法
看完了put方法再来看get方法就很容易了，用逆向思维去看就好，这样存的话我反过来这么取就好了。get方法源码为：
public v get(object key) {
node<k,v>[] tab; node<k,v> e, p; int n, eh; k ek;
// 1\. 重hash
int h = spread(key.hashcode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabat(tab, (n - 1) & h)) != null) {
// 2\. table[i]桶节点的key与查找的key相同，则直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// 3\. 当前节点hash小于0说明为树节点，在红黑树中查找即可
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {
//4\. 从链表中查找，查找到则返回该节点的value，否则就返回null即可
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
代码的逻辑请看注释，首先先看当前的hash桶数组节点即table[i]是否为查找的节点，若是则直接返回；若不是，则继续再看当前是不是树节点？通过看节点的hash值是否为小于0，如果小于0则为树节点。如果是树节点在红黑树中查找节点；如果不是树节点，那就只剩下为链表的形式的一种可能性了，就向后遍历查找节点，若查找到则返回节点的value即可，若没有找到就返回null。
3.5 transfer方法
当concurrenthashmap容量不足的时候，需要对table进行扩容。这个方法的基本思想跟hashmap是很像的，但是由于它是支持并发扩容的，所以要复杂的多。原因是它支持多线程进行扩容操作，而并没有加锁。我想这样做的目的不仅仅是为了满足concurrent的要求，而是希望利用并发处理去减少扩容带来的时间影响。transfer方法源码为：
private final void transfer(node<k,v>[] tab, node<k,v>[] nexttab) {
int n = tab.length, stride;
if ((stride = (ncpu > 1) ? (n >>> 3) / ncpu : n) < min_transfer_stride)
stride = min_transfer_stride; // subdivide range
//1\. 新建node数组，容量为之前的两倍
if (nexttab == null) { // initiating
try {
@suppresswarnings("unchecked")
node<k,v>[] nt = (node<k,v>[])new node<?,?>[n << 1];
nexttab = nt;
} catch (throwable ex) { // try to cope with oome
sizectl = integer.max_value;
return;
}
nexttable = nexttab;
transferindex = n;
}
int nextn = nexttab.length;
//2\. 新建forwardingnode引用，在之后会用到
forwardingnode<k,v> fwd = new forwardingnode<k,v>(nexttab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nexttab
for (int i = 0, bound = 0;;) {
node<k,v> f; int fh;
// 3\. 确定遍历中的索引i
while (advance) {
int nextindex, nextbound;
if (--i >= bound || finishing)
advance = false;
else if ((nextindex = transferindex) <= 0) {
i = -1;
advance = false;
}
else if (u.compareandswapint
(this, transferindex, nextindex,
nextbound = (nextindex > stride ?
nextindex - stride : 0))) {
bound = nextbound;
i = nextindex - 1;
advance = false;
}
}
//4.将原数组中的元素复制到新数组中去
//4.5 for循环退出，扩容结束修改sizectl属性
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
nexttable = null;
table = nexttab;
sizectl = (n << 1) - (n >>> 1);
return;
}
if (u.compareandswapint(this, sizectl, sc = sizectl, sc - 1)) {
if ((sc - 2) != resizestamp(n) << resize_stamp_shift)
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
//4.1 当前数组中第i个元素为null，用cas设置成特殊节点forwardingnode(可以理解成占位符)
else if ((f = tabat(tab, i)) == null)
advance = castabat(tab, i, null, fwd);
//4.2 如果遍历到forwardingnode节点 说明这个点已经被处理过了 直接跳过 这里是控制并发扩容的核心
else if ((fh = f.hash) == moved)
advance = true; // already processed
else {
synchronized (f) {
if (tabat(tab, i) == f) {
node<k,v> ln, hn;
if (fh >= 0) {
//4.3 处理当前节点为链表的头结点的情况，构造两个链表，一个是原链表 另一个是原链表的反序排列
int runbit = fh & n;
node<k,v> lastrun = f;
for (node<k,v> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runbit) {
runbit = b;
lastrun = p;
}
}
if (runbit == 0) {
ln = lastrun;
hn = null;
}
else {
hn = lastrun;
ln = null;
}
for (node<k,v> p = f; p != lastrun; p = p.next) {
int ph = p.hash; k pk = p.key; v pv = p.val;
if ((ph & n) == 0)
ln = new node<k,v>(ph, pk, pv, ln);
else
hn = new node<k,v>(ph, pk, pv, hn);
}
//在nexttable的i位置上插入一个链表
settabat(nexttab, i, ln);
//在nexttable的i+n的位置上插入另一个链表
settabat(nexttab, i + n, hn);
//在table的i位置上插入forwardnode节点 表示已经处理过该节点
settabat(tab, i, fwd);
//设置advance为true 返回到上面的while循环中 就可以执行i--操作
advance = true;
}
//4.4 处理当前节点是treebin时的情况，操作和上面的类似
else if (f instanceof treebin) {
treebin<k,v> t = (treebin<k,v>)f;
treenode<k,v> lo = null, lotail = null;
treenode<k,v> hi = null, hitail = null;
int lc = 0, hc = 0;
for (node<k,v> e = t.first; e != null; e = e.next) {
int h = e.hash;
treenode<k,v> p = new treenode<k,v>
(h, e.key, e.val, null, null);
if ((h & n) == 0) {
if ((p.prev = lotail) == null)
lo = p;
else
lotail.next = p;
lotail = p;
++lc;
}
else {
if ((p.prev = hitail) == null)
hi = p;
else
hitail.next = p;
hitail = p;
++hc;
}
}
ln = (lc <= untreeify_threshold) ? untreeify(lo) :
(hc != 0) ? new treebin<k,v>(lo) : t;
hn = (hc <= untreeify_threshold) ? untreeify(hi) :
(lc != 0) ? new treebin<k,v>(hi) : t;
settabat(nexttab, i, ln);
settabat(nexttab, i + n, hn);
settabat(tab, i, fwd);
advance = true;
}
}
}
}
}
}
代码逻辑请看注释,整个扩容操作分为两个部分：
第一部分是构建一个nexttable,它的容量是原来的两倍，这个操作是单线程完成的。新建table数组的代码为:node<k,v>[] nt = (node<k,v>[])new node<?,?>[n << 1],在原容量大小的基础上右移一位。
第二个部分就是将原来table中的元素复制到nexttable中，主要是遍历复制的过程。 根据运算得到当前遍历的数组的位置i，然后利用tabat方法获得i位置的元素再进行判断：
- 如果这个位置为空，就在原table中的i位置放入forwardnode节点，这个也是触发并发扩容的关键点；
- 如果这个位置是node节点（fh>=0），如果它是一个链表的头节点，就构造一个反序链表，把他们分别放在nexttable的i和i+n的位置上
- 如果这个位置是treebin节点（fh<0），也做一个反序处理，并且判断是否需要untreefi，把处理的结果分别放在nexttable的i和i+n的位置上
- 遍历过所有的节点以后就完成了复制工作，这时让nexttable作为新的table，并且更新sizectl为新容量的0.75倍 ，完成扩容。设置为新容量的0.75倍代码为
sizectl = (n << 1) - (n >>> 1)，仔细体会下是不是很巧妙，n<<1相当于n右移一位表示n的两倍即2n,n>>>1左右一位相当于n除以2即0.5n,然后两者相减为2n-0.5n=1.5n,是不是刚好等于新容量的0.75倍即2n*0.75=1.5n。
3.6 与size相关的一些方法
对于concurrenthashmap来说，这个table里到底装了多少东西其实是个不确定的数量，因为不可能在调用size()方法的时候像gc的“stop the world”一样让其他线程都停下来让你去统计，因此只能说这个数量是个估计值。对于这个估计值，concurrenthashmap也是大费周章才计算出来的。
为了统计元素个数，concurrenthashmap定义了一些变量和一个内部类
/**
* a padded cell for distributing counts. adapted from longadder
* and striped64\. see their internal docs for explanation.
*/
@sun.misc.contended static final class countercell {
volatile long value;
countercell(long x) { value = x; }
}
/******************************************/
/**
* 实际上保存的是hashmap中的元素个数 利用cas锁进行更新
但它并不用返回当前hashmap的元素个数
*/
private transient volatile long basecount;
/**
* spinlock (locked via cas) used when resizing and/or creating countercells.
*/
private transient volatile int cellsbusy;
/**
* table of counter cells. when non-null, size is a power of 2.
*/
private transient volatile countercell[] countercells;
mappingcount与size方法
mappingcount与size方法的类似 从给出的注释来看，应该使用mappingcount代替size方法 两个方法都没有直接返回basecount 而是统计一次这个值，而这个值其实也是一个大概的数值，因此可能在统计的时候有其他线程正在执行插入或删除操作。
public int size() {
long n = sumcount();
return ((n < 0l) ? 0 :
(n > (long)integer.max_value) ? integer.max_value :
(int)n);
}
/**
* returns the number of mappings. this method should be used
* instead of {@link #size} because a concurrenthashmap may
* contain more mappings than can be represented as an int. the
* value returned is an estimate; the actual count may differ if
* there are concurrent insertions or removals.
*
* @return the number of mappings
* @since 1.8
*/
public long mappingcount() {
long n = sumcount();
return (n < 0l) ? 0l : n; // ignore transient negative values
}
final long sumcount() {
countercell[] as = countercells; countercell a;
long sum = basecount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;//所有counter的值求和
}
}
return sum;
}
addcount方法
在put方法结尾处调用了addcount方法，把当前concurrenthashmap的元素个数+1这个方法一共做了两件事,更新basecount的值，检测是否进行扩容。
private final void addcount(long x, int check) {
countercell[] as; long b, s;
//利用cas方法更新basecount的值
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();
}
//如果check值大于等于0 则需要检验是否需要进行扩容操作
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);
}
//当前线程是唯一的或是第一个发起扩容的线程 此时nexttable=null
else if (u.compareandswapint(this, sizectl, sc,
(rs << resize_stamp_shift) + 2))
transfer(tab, null);
s = sumcount();
}
}
}
4. 总结
jdk6,7中的concurrenthashmap主要使用segment来实现减小锁粒度，分割成若干个segment，在put的时候需要锁住segment，get时候不加锁，使用volatile来保证可见性，当要统计全局时（比如size），首先会尝试多次计算modcount来确定，这几次尝试中，是否有其他线程进行了修改操作，如果没有，则直接返回size。如果有，则需要依次锁住所有的segment来计算。
1.8之前put定位节点时要先定位到具体的segment，然后再在segment中定位到具体的桶。而在1.8的时候摒弃了segment臃肿的设计，直接针对的是node[] tale数组中的每一个桶，进一步减小了锁粒度。并且防止拉链过长导致性能下降，当链表长度大于8的时候采用红黑树的设计。
主要设计上的变化有以下几点:
- 不采用segment而采用node，锁住node来实现减小锁粒度。
- 设计了moved状态 当resize的中过程中 线程2还在put数据，线程2会帮助resize。
- 使用3个cas操作来确保node的一些操作的原子性，这种方式代替了锁。
- sizectl的不同值来代表不同含义，起到了控制的作用。
- 采用synchronized而不是reentrantlock
上一篇： delete误删数据使用SCN恢复