ConcurrentHashMap源码解析 JDK8
超级小小黑 人气:1一、简介
上篇文章详细介绍了HashMap的源码及原理，本文趁热打铁继续分析ConcurrentHashMap的原理。
首先在看本文之前，希望对HashMap有一个详细的了解。不然看直接看ConcurrentHashMap的源码还是有些费劲的。
相信对HashMap，HashTable有一定了解，应该知道HashMap是不具备线程安全性的，在resize时会丢数据（JDK8），而HashTable虽然保证了线程安全性，但是其是通过给每个方法加Synchronized关键字达到的同步目的。但是都知道Synchronized在竞争激烈的多线程并发环境中，在性能上的表现是非常不如人意的。那在高并发环境中HashMap如何保证线程安全而又不浪费太多性能呢？答案就是Java J.U.C并发包中的ConcurrentHashMap。
依然开局一张图。JDK8中的ConcurrentHashMap数据结构。
呃呵，和HashMap的结构是一样的，没错在数据结构层面，ConcurrentHashMap和HashMap是完全一样的。有了这个基础继续往下看。
二、历史版本
ConcurrentHashMap的历史版本大致分界线在JDK8。也就是可以分为JDK8和JDK8以前版本。
数据结构的区别
在JDK8之前HashMap没有引入红黑树，同样的ConcurrentHashMap也没有引入红黑树。而且ConcurrentHashMap采用的是分段数组的底层数据结构。
在JDK7中的数据结构。
从上图我们不难看出其在数据结构方面的差别。
锁的区别
JDK7中为了提高并发性能采用了这种分段的设计。所以在JDK7中ConcurrentHashMap采用的是分段锁，也就是在每个Segment上加ReentrantLock实现的线程安全线。关于ReetrantLock后面有时间会介绍，大致来说ReetrantLoack是比Synchronized更细粒度的一种锁。使用得当的话其性能要比Synchronized表现要好，但是如果实现不得当容易造成死锁。
这种基于Segment和ReetrantLock的设计相对HashTable来说大大提高了并发性能。也就是说多个线程可以并发的操作多个Segment，而HashTable是通过给每个方法加Synchronized即将多线程串行而实现的。所以在一定程度上提高了并发性能。但是这种性能的提升表现相对JDK8来说显得不值一提。
如果说JDK7 ConcurrentHashMap相对HashTable来说是串行到多个线程并发的改进。而JDK8则是通过比Segment更细粒度的并发控制大大提高了其并发表现。
JDK8中ConcurrentHashMap采用的是CAS+Synchronized锁并且锁粒度是每一个桶。简单来说JDK7中锁的粒度是Segment，JDK8锁粒度细化到了桶级别。可想而知锁粒度是大大提到了。辅之以代码的优化，JDK8中的ConcurrentHashMap在性能上的表现非常优秀。
简单总结一下，从HashTable到JDK7 ConcurrentHashMap再到JDK8 ConcurrentHashMap。是从同步到并发再到高并发的进步。
三、基础知识
3.1、常量
//正在扩容，对应fwd类型的节点的hash static final int MOVED = -1; // hash for forwarding nodes //当前数组 transient volatile Node<K,V>[] table; //扩容时用到的，扩容后的数组。 private transient volatile Node<K,V>[] nextTable; //1，大于零，表示size * 0.75。 //2，等于-1，表示正在初始化。 //3，-(n + 1)，表示正在执行扩容的线程其只表示基数，而不是真正的数量，需要计算得出的哦 private transient volatile int sizeCtl;
3.2、Unsafe类方法
1 @SuppressWarnings("unchecked") //transient volatile Node<K,V>[] table; tab变量确实是volatile
2 static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {//获取table中索引 i 处的元素。
3 return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);//如果tab是volatile变量，则该方法保证其可见性。
4 }
5
6 static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,//通过CAS设置table索引为 i 处的元素。
7 Node<K,V> c, Node<K,V> v) {
8 return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
9 }
10 //transient volatile Node<K,V>[] table; tab变量确实是volatile
11 static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {//修改table 索引 i 处的元素。
12 U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);//如果tab是volatile变量，则该方法保证其可见性。
13 }
我们不难看出 以上三个方法都是调用的Unsafe（U）类中的方法，Unsafe类中定义了大量对内存的操作方法，是native的，不建议开发者直接使用。
tabAt和setTabAt最终调用的两个方法分别是 U.getObjectVolatile()和U.putObjectVolatile 顾名思义其是通过volatile保证的tab的可见性（Volatile只保证可见性不保证原子性哦）。前提是tab变量是Volatile修饰的变量。我们通过调用栈，最红可以看到其实tab就是ConcurrentHashMap中的table。而这个变量是这么定义的。
transient volatile Node<K,V>[] table;
可见其确实是Volatile修饰的变量。
再看
casTabAt方法，这个就是CAS方法了。
CAS：Compare and Swap三个单词的缩写，即：比较交换的意思。CAS在Java中又称之为乐观锁即我们总认为是没有锁的。
while(true){
CAS();
}
一般的通过上述用法达到自旋的目的。CAS一般通过自旋达到自旋锁的目的，即认为没有锁，失败重试，这种思路。更多内容请自行百度。CAS很重要哦。
四、put过程源码
1 public V put(K key, V value) {
2 return putVal(key, value, false);
3 }
4
5 /** Implementation for put and putIfAbsent */
6 final V putVal(K key, V value, boolean onlyIfAbsent) {
7 if (key == null || value == null) throw new NullPointerException();
8 int hash = spread(key.hashCode());//hash，对hashcode再散列
9 int binCount = 0;
10 for (Node<K,V>[] tab = table;;) {//迭代桶数组，自旋
11 Node<K,V> f; int n, i, fh;
12 if (tab == null || (n = tab.length) == 0)//懒加载。如果为空，则进行初始化
13 tab = initTable();//初始化桶数组
14 //(n - 1) & hash)计算下标，取值，为空即无hash碰撞
15 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
16 if (casTabAt(tab, i, null,
17 new Node<K,V>(hash, key, value, null)))//通过cas插入新值
18 break; // no lock when adding to empty bin
19 }
20 //判断是否正在扩容。如果正在扩容，当前线程帮助进行扩容。
21 //每个线程只能同时负责一个桶上的数据迁移，并且不影响其它桶的put和get操作。
22 //（很牛逼的思路，能这么做建立在更细粒度的锁基础上）
23 else if ((fh = f.hash) == MOVED)
24 tab = helpTransfer(tab, f);
25 else {//put5,存在hash碰撞
26 V oldVal = null;
27 //此处，f在上面已经被赋值，f为当前下标桶的首元素。对链表来说是链表头对红黑树来说是红黑树的头元素。
28 synchronized (f) {
29 //再次检查当前节点是否有变化，有变化进入下一轮自旋
30 //为什么再次检查？因为不能保证，当前线程到这里，有没有其他线程对该节点进行修改
31 if (tabAt(tab, i) == f) {
32 if (fh >= 0) {//当前桶为链表
33 binCount = 1;
34 for (Node<K,V> e = f;; ++binCount) {//迭代链表节点
35 K ek;
36 if (e.hash == hash &&//key相同，覆盖（onlyIfAbsent有什么用？）
37 ((ek = e.key) == key ||
38 (ek != null && key.equals(ek)))) {
39 oldVal = e.val;
40 if (!onlyIfAbsent)
41 e.val = value;
42 break;
43 }
44 Node<K,V> pred = e;
45 //找到链表尾部，插入新节点。（什么这里不用CAS？因为这在同步代码块里面）
46 if ((e = e.next) == null) {
47 pred.next = new Node<K,V>(hash, key,
48 value, null);
49 break;
50 }
51 }
52 }
53 else if (f instanceof TreeBin) {//当前桶为红黑树
54 Node<K,V> p;
55 binCount = 2;
56 if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
57 value)) != null) {//想红黑树插入新节点
58 oldVal = p.val;
59 if (!onlyIfAbsent)
60 p.val = value;
61 }
62 }
63 }
64 }
65 if (binCount != 0) {
66 //树化。binCount > 8，进行树化，链表转红黑树
67 if (binCount >= TREEIFY_THRESHOLD)
68 //如果容量 < 64则直接进行扩容；不转红黑树。
69 //（你想想，假如容量为16，你就插入了9个元素，巧了，都在同一个桶里面，
70 //如果这时进行树化，时间复杂度会增加，性能下降，不如直接进行扩容，空间换时间）
71 treeifyBin(tab, i);
72 if (oldVal != null)
73 return oldVal;
74 break;
75 }
76 }
77 }
78 addCount(1L, binCount);//扩容。addCount内部会进行判断要不要扩容
79 return null;
80 }
总结以上过程
1，懒加载，未初始化则初始化table 2，hash，hashcode再散列，并计算下标 3，无碰撞，通过CAS插入 4，有碰撞 4.1、如果正在扩容，协助其它线程去扩容 4.2、如果是链表，插入链表 4.3、如果是红黑树，插入红黑树 4.4、如果链表长度超过8，树化 4.5，如果key已经存在，覆盖旧值 5，需要扩容，则扩容
相比HashMap过程多了一个协助扩容。
以上源码需要注意的是
1 for (Node<K,V>[] tab = table;;) {//迭代桶数组，自旋
2
3 }
这是一个自旋的过程，如果CAS修改失败会进入下一轮自旋。很久以前看这段源码的时候，我总是在想，CAS失败了不就丢数据了吗？所以这个自旋，也称之为自旋锁会保证数据一定能插入成功。
说说上面锁竞争的情况，以上过程我们不难发现对table的修改都是通过CAS操作实现的。比如下面这行代码，如果已经有线程正在操作 i 位置的元素，则意味着本轮自旋将会失败，继续自旋，当其他线程修改完成，本线程再次运行到tabAt以为是Volatile操作，其他线程的修改对本线程立即可见（详见Volatile关键字内存语义的内容）。本线程通过tabAt发现该处已经存在元素，即发生碰撞，继续往下运行。
1 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
2 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))//通过cas插入新值
3 break; // no lock when adding to empty bin
4 }
线程的调度需要操作系统从用户态转为内核态，这是非常重量级的操作。CAS+自旋组成的自旋锁保证了线程不会进入阻塞态。
然后继续往下看
synchronized (f) {
//再次检查当前节点是否有变化，有变化进入下一轮自旋
//为什么再次检查？因为不能保证，当前线程运行到这里，有没有其他线程对该节点进行修改
if (tabAt(tab, i) == f) {
先看这行代码 synchronized (f) 这个f是一个桶的头元素。也就是说在JDK8中synchronized锁仅仅只锁链表头或者红黑树的头（其实就是锁一个桶，因为要访问链表或者红黑树总要从头开始访问吧）
再看 if (tabAt(tab, i) == f) {} 其实就是双重检测（参考单例的双重检测），为什么要再检查一遍呢？因为不能保证当前线程运行到这里，有没有其他线程已经对该节点进行了修改。
initTable()
1 private final Node<K,V>[] initTable() {
2 Node<K,V>[] tab; int sc;
3 while ((tab = table) == null || tab.length == 0) {
4 // 赋值sc。并当sizeCtl == -1 即当前有线程正在执行初始化
5 if ((sc = sizeCtl) < 0)
6 //yield()暂停当前正在执行的线程，执行其他线程
7 //（这是一个通知，但是这是不一定会让当前线程停止，要取决于线程调度器）
8 //就是我想让出资源，但是这只是一厢情愿的事情，线程调度器会考虑你的方法，但是不一定采纳。
9 Thread.yield();
10 //修改 sizeCtl 的值为 -1。 SIZECTL 为 sizeCtl 的内存地址。
11 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
12 try {
13 //执行初始化过程
14 if ((tab = table) == null || tab.length == 0) {
15 //sc在上面已经赋值，=原来 sizeCtl的值。是非讨厌JDK源码这种赋值方式。
16 int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
17 @SuppressWarnings("unchecked")
18 //创建一个sc长度的table。
19 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
20 table = tab = nt;
21 sc = n - (n >>> 2);
22 }
23 } finally {
24 //初始化完成, sizeCtl重新赋值为当前数组的长度。
25 sizeCtl = sc;
26 }
27 break;
28 }
29 }
30 return tab;
31 }
以上过程，同样是通过CAS实现的初始化控制，保证只有一个线程去执行初始化。
helpTransfer(tab, f);方法我们后面介绍完扩容再说。
看完以上put过程，我们能发现，JDK8通过CAS+自旋锁将锁的粒度控制在每一个桶上，相对于JDK7中Segment锁，锁粒度提高了很多。并且CAS+自旋锁保证了不会出现线程的切花这种重量级的操作。
五、扩容
1 //tab旧桶数组，nextTab新桶数组
2 private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
3 int n = tab.length, stride;
4 //控制并发数，控制CPU的资源
5 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
6 stride = MIN_TRANSFER_STRIDE; // subdivide range
7 if (nextTab == null) { // initiating//新数组为空，则初始化新数组
8 try {
9 @SuppressWarnings("unchecked")
10 //扩容为原来的两倍 n << 1
11 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
12 nextTab = nt;
13 } catch (Throwable ex) { // try to cope with OOME
14 sizeCtl = Integer.MAX_VALUE;
15 return;
16 }
17 nextTable = nextTab;
18 transferIndex = n;
19 }
20 int nextn = nextTab.length;
21 //在这里面进行new Node将node.hash置为-1。表示该桶正在进行移动。
22 //（这里很重要的一点是，只锁表头，所以只需要将链表（或者红黑树）头结点.hash置为-1即可）
23 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
24 //advance是控制是否继续进行移动的条件，当advance == false，表示正在移动一个桶。
25 //true表示可以继续进行下一个桶的移动
26 boolean advance = true;
27 boolean finishing = false; // to ensure sweep before committing nextTab
28 for (int i = 0, bound = 0;;) {//自旋
29 Node<K,V> f; int fh;
30 while (advance) {//start
31 int nextIndex, nextBound;
32 //当前桶是不是已经移动完了
33 if (--i >= bound || finishing)
34 advance = false;
35 //两个停止移动的条件。移动完了。（这个是真正停止的条件。下面那个条件会进行一次检查）
36 else if ((nextIndex = transferIndex) <= 0) {
37 i = -1;
38 advance = false;
39 }
40 else if (U.compareAndSwapInt
41 (this, TRANSFERINDEX, nextIndex,
42 nextBound = (nextIndex > stride ?
43 nextIndex - stride : 0))) {
44 bound = nextBound;
45 i = nextIndex - 1;
46 advance = false;
47 }
48 }
49 if (i < 0 || i >= n || i + n >= nextn) {
50 int sc;
51 if (finishing) {//结束扩容
52 nextTable = null;
53 table = nextTab;
54 sizeCtl = (n << 1) - (n >>> 1);
55 return;
56 }
57 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
58 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
59 return;
60 finishing = advance = true;
61 i = n; // recheck before commit 再次检查一遍，防止有桶中还有数据没移动。
62 }
63 }//end 从start到end可看可不看就是条件控制，包括结束条件的控制，移动进度的控制等。
64 //该桶没数据
65 else if ((f = tabAt(tab, i)) == null)
66 //将oldtab中的该桶设置为fwd节点，hash=-1
67 advance = casTabAt(tab, i, null, fwd);
68 //已经移动过的桶其hash=-1
69 else if ((fh = f.hash) == MOVED)
70 advance = true; // already processed
71 else {
72 synchronized (f) {//上锁
73 if (tabAt(tab, i) == f) {
74 //ln新链表，不需要移动的节点重新组组织成的链表。
75 //hn新链表，需要移动的节点重新组织成的链表
76 Node<K,V> ln, hn;
77 if (fh >= 0) {//链表
78 int runBit = fh & n;
79 Node<K,V> lastRun = f;
80 //start
81 //从start，到end之间。不看也行。实在费脑子。其实这段代码写的有点让人费解
82 //主要是不认真看不知道作者的意图。本意是这样的。判断是不是可以从某个节点n开始
83 //后面的节点是不是都是和节点n一样，移动的目标桶一样的。
84 //如果是一样的，则后面的这些节点就不用移动了，只需要移动n节点即可。
85 //（注意链表的引用，next指针就把后面的都带过去了）
86 //想一个极端情况，如果在这里迭代后发现，所有节点，扩容后数据移动的目标桶都是一样的。
87 //则只需要移动头结点即可。不用重新拼接链表了。
88 for (Node<K,V> p = f.next; p != null; p = p.next) {
89 int b = p.hash & n;
90 if (b != runBit) {
91 runBit = b;
92 lastRun = p;
93 }
94 }
95 if (runBit == 0) {// runBit== 0 表示该节点不需要移动
96 ln = lastRun;
97 hn = null;
98 }
99 else {
100 hn = lastRun;
101 ln = null;
102 }//end
103 for (Node<K,V> p = f; p != lastRun; p = p.next) {
104 int ph = p.hash; K pk = p.key; V pv = p.val;
105 if ((ph & n) == 0)
106 ln = new Node<K,V>(ph, pk, pv, ln);
107 else
108 hn = new Node<K,V>(ph, pk, pv, hn);
109 }
110 setTabAt(nextTab, i, ln);
111 setTabAt(nextTab, i + n, hn);
112 setTabAt(tab, i, fwd);
113 advance = true;
114 }
115 else if (f instanceof TreeBin) {//红黑树
116 TreeBin<K,V> t = (TreeBin<K,V>)f;
117 TreeNode<K,V> lo = null, loTail = null;
118 TreeNode<K,V> hi = null, hiTail = null;
119 int lc = 0, hc = 0;
120 for (Node<K,V> e = t.first; e != null; e = e.next) {
121 int h = e.hash;
122 TreeNode<K,V> p = new TreeNode<K,V>
123 (h, e.key, e.val, null, null);
124 if ((h & n) == 0) {
125 if ((p.prev = loTail) == null)
126 lo = p;
127 else
128 loTail.next = p;
129 loTail = p;
130 ++lc;
131 }
132 else {
133 if ((p.prev = hiTail) == null)
134 hi = p;
135 else
136 hiTail.next = p;
137 hiTail = p;
138 ++hc;
139 }
140 }
141 ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
142 (hc != 0) ? new TreeBin<K,V>(lo) : t;
143 hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
144 (lc != 0) ? new TreeBin<K,V>(hi) : t;
145 setTabAt(nextTab, i, ln);
146 setTabAt(nextTab, i + n, hn);
147 setTabAt(tab, i, fwd);
148 advance = true;
149 }
150 }
151 }
152 }
153 }
154 }
5.1、扩容前准备阶段
ForwardingNode
1 static final class ForwardingNode<K,V> extends Node<K,V> {
2 final Node<K,V>[] nextTable;
3 ForwardingNode(Node<K,V>[] tab) {
4 super(MOVED, null, null, null);
5 this.nextTable = tab;
6 }
7 }
看一下这个内部类，其实呢其就是一个起到标识作用的节点，该节点看上面代码可知，该节点最主要的特点就是hash=MOVED=-1。hash=-1的节点在ConcurrentHashMap中表示该桶是被扩容过程迁移过的桶。然后当前线程判断如果该桶已经被迁移。无论put还是get都去新的数组中操作。还有一点很重要，还可以通过ForwardingNode中 nextTable获取到新的数组。
1 //该桶没数据 2 else if ((f = tabAt(tab, i)) == null) 3 //将oldtab中的该桶设置为fwd节点，hash=-1 4 advance = casTabAt(tab, i, null, fwd);
看上面代码，先判断该桶还有没有数据。没数据不用迁移，等同于已经迁移完了。其他线程put会直接put到新的数组中。
1 //已经移动过的桶其hash=-1； 2 else if ((fh = f.hash) == MOVED) 3 advance = true; // already processed
如果该桶已经移动则跳过。
到此我们能看出什么？主要是已经移动完的设置成fwd节点，其它线程看到该桶已经移动，则会到新的table中操作。如果未移动，还直接操作当前table，因为就算put，待会处理到该桶，一样移动到新桶，也没啥影响。如果是正在移动的接下来会看到加了Synchronized锁，保证只有一个线程能操作当前桶。简直不要太妙。
5.2、扩容过程
画重点，扩容过程
1 synchronized (f) {//上锁
2 if (tabAt(tab, i) == f) {
3 //ln新链表，不需要移动的节点重新组组织成的链表。
4 //hn新链表，需要移动的节点重新组织成的链表
5 Node<K,V> ln, hn;
6 if (fh >= 0) {//链表
7 int runBit = fh & n;
8 Node<K,V> lastRun = f;
9 //start
10 //从start，到end之间。不看也行。实在费脑子。其实这段代码写的有点让人费解
11 //主要是不认真看不知道作者的意图。本意是这样的。判断是不是可以从某个节点n开始
12 //后面的节点是不是都是和节点n一样，移动的目标桶一样的。
13 //如果是一样的，则后面的这些节点就不用移动了，只需要移动n节点即可。
14 //（注意链表的引用，next指针就把后面的都带过去了）
15 //想一个极端情况，如果在这里迭代后发现，所有节点，扩容后数据移动的目标桶都是一样的。
16 //则只需要移动头结点即可。不用重新拼接链表了。
17 for (Node<K,V> p = f.next; p != null; p = p.next) {
18 int b = p.hash & n;
19 if (b != runBit) {
20 runBit = b;
21 lastRun = p;
22 }
23 }
24 if (runBit == 0) {// runBit== 0 表示该节点不需要移动
25 ln = lastRun;
26 hn = null;
27 }
28 else {
29 hn = lastRun;
30 ln = null;
31 }//end
32 for (Node<K,V> p = f; p != lastRun; p = p.next) {
33 int ph = p.hash; K pk = p.key; V pv = p.val;
34 if ((ph & n) == 0)
35 ln = new Node<K,V>(ph, pk, pv, ln);
36 else
37 hn = new Node<K,V>(ph, pk, pv, hn);
38 }
39 setTabAt(nextTab, i, ln);
40 setTabAt(nextTab, i + n, hn);
41 setTabAt(tab, i, fwd);
42 advance = true;
43 }
44 else if (f instanceof TreeBin) {//红黑树
45 //红黑树跳过
46 }
47 }
48 }
5.2.1、并发控制
首先扩容过程是在synchronized同步代码块中的。并且只锁了一个表头。可看到没有锁新数组nextTab的桶。想想，oldTab（tab变量）和nextTab都是多个线程共享的变量，为什么只有只锁了oldTab正在操作的桶？如果有多个线程向nextTab同时迁移数据怎么办？会不会存在线程安全性问题？
TIPS：
统一术语
tab = oldTab = table（旧数组）
newTab = nextTab（扩容后新数组）
oldIndex即在oldTab中的索引位
newIndex即在newTab中的位置
在上一篇文章中介绍HashMap的时候详细介绍了HashMap扩容中，oldTab旧桶迁移向newTab只有两个目标桶。再简单回顾一遍。
上面这张图形象的展示了旧桶在扩容后的两个去向：1，索引位原地不动，2，索引位为oldCap+oldIndex。（关于为什么是这两个去向，在HashMap扩容中已经详细介绍了）
如果你还没懂我的疑问，请参考下面这个图。
前提，ConcurrentHashMap是并发扩容，可以有多个线程同时扩容，其次如果如上图红线那样，oldTab中有多个桶中的数据迁移到newTab中的同一个桶中，如果出现这种情况就意味着存在线程安全性问题。
从上图5-1中，两个数据迁移的方向可知，扩容前，oldIndex不同就表示不在一个桶，扩容后的两个去向如果oldIndex不一样，也一定不在同一个桶。所以不会出现5-2图中红线的那种情况，也就说明在扩容过程中不需要锁newTab。佩服+2
5.2.2、数据迁移
//ln新链表，不需要移动的节点重新组组织成的链表。 //hn新链表，需要移动的节点重新组织成的链表 Node<K,V> ln, hn;
int runBit = fh & n;
看两个变量，上面说过扩容后，旧桶中的数据只有两个迁移的方向。ln正是数据迁移后索引位依然是oldIndex的数据的链表，hn是迁移后需要迁移到oldCap + oldIndex索引位的链表。
关注一下runBit变量，如果 runBit == 0 成立则说明迁移后桶的索引位依然是oldIndex。详见HashMap扩容分析。
重点关注一下start到end之间的代码
关于这段代码，首选我们假设一种极端情况，如果当前正在移动的桶中的数据在rehash之后，数据迁移的目标桶除了第一个节点的目标桶是oldIndex之外，后面的数据的目标桶都是oldIndex + oldCap。我们还需要处理后面的节点吗？不需要，因为只需要将第二个节点移动到newTab的oldIndex + oldCap位置即可。第二个元素也就是lastRun变量。相对于HashMap完全的将数据组织成两个链表，这也算得上是一个性能上的优化吧。
接着往下看
代码段1：
1 for (Node<K,V> p = f.next; p != null; p = p.next) {
2 int b = p.hash & n;
3 if (b != runBit) {//相同跳过
4 runBit = b;
5 lastRun = p;
6 }
7 }
以上代码通过对链表的一次扫描决定了lastRun。
代码段2：
1 if (runBit == 0) {// runBit== 0 表示该节点不需要移动
2 ln = lastRun;
3 hn = null;
4 }
5 else {
6 hn = lastRun;
7 ln = null;
8 }//end
根据lastRun指向的节点的runBit决定后续节点在扩容后是oldIndex + oldCap还是oldIndex。
代码段3：
1 for (Node<K,V> p = f; p != lastRun; p = p.next) {
2 int ph = p.hash; K pk = p.key; V pv = p.val;
3 if ((ph & n) == 0)
4 ln = new Node<K,V>(ph, pk, pv, ln);
5 else
6 hn = new Node<K,V>(ph, pk, pv, hn);
7 }
上述代码会重新组织两个新链表。注意这个迭代到lastRun位置结束，因为以上过程已经确定了lastRun的归属。
看一下 ln = new Node<K,V>(ph, pk, pv, ln); 重新组织链表的代码，也就是ln会成为新new出来的node的下一个节点。
这样有什么问题？问题就是节点在旧桶中的相对顺序在新桶中将相反。也就是next的指针翻转一下。可以看一下node的构造函数就明了了。
演示扩容过程
假设当前扩容前oldCap即oldTab的长度为2，扩容后newCap即newTab的长度为4。如下图看扩容过程，橘色的代表迁移后索引位依然是oldIndex，绿色代表扩容后索引位为oldIndex + oldCap。
上述代码段1迭代找到了lastRun即指向node（11），代码段2将lastRun赋值给hn。代码段3执行过程如下
1，将node(1)拼接到ln 2，将node(3)拼接到hn，此时注意，hn已经lastRun指向的节点node(11)，此时hn=3—>11—>15—>19—>null 3，处理node(5)拼接到ln 4，处理...
对比JDK7 HashMap，JDK8 HashMap，JDK8 ConcurrentHashMap在扩容后对节点相对顺序的保证方面，JDK7 HashMap是完全倒序。JDK8 HashMap不改变相对顺序。JDK8 ConcurrentHashMap 保证部分节点的相对顺序，其余的倒序。
题外话，从代码风格和死路上，猜测一下ConcurrentHashMap应该是来自JDK7的HashMap。
1 setTabAt(nextTab, i, ln); 2 setTabAt(nextTab, i + n, hn); 3 setTabAt(tab, i, fwd); 4 advance = true;
ln和hn两个链表各回各家各找各妈。
回过头来再看put方法中的帮助扩容
1 else if ((fh = f.hash) == MOVED) 2 tab = helpTransfer(tab, f);
在put方法中有这样一行判断，当f.hash = MOVED即当前HashMap正在扩容中，则当前线程会去尝试帮助扩容。
1 final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
2 Node<K,V>[] nextTab; int sc;
3 if (tab != null && (f instanceof ForwardingNode) &&//条件判断
4 (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {//从fwd节点中取出新table
5 int rs = resizeStamp(tab.length);
6 while (nextTab == nextTable && table == tab &&
7 (sc = sizeCtl) < 0) {
8 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
9 sc == rs + MAX_RESIZERS || transferIndex <= 0)
10 break;
11 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {//修改sizeCtl = sizeCtl + 1，表示多了一个线程参与扩容
12 transfer(tab, nextTab);
13 break;
14 }
15 }
16 return nextTab;
17 }
18 return table;
19 }
在helpTransfer方法中会首先做一系列判断，通过fwd节点获取到nextTab即新的数组。通过CAS 实现sizeCtl++操作，表示多了一个线程进行扩容，因为在扩容方法中对扩容线程数量有控制。
最后的最后，扩容的时机
说一下触发扩容的操作，总的来说就是put操作，但是有两个时机很重要，其一就是addCount方法中，每次put一个元素，在addCount方法中都会判断需不需要进行扩容。另外就是treeifyBin方法中，如果桶中数据超过了8个并且数组长度<64则不会进行树化，而是进行扩容。关于这个在HashMap源码介绍中也有介绍。你想想，假如容量为16，你就插入了9个元素，巧了，都在同一个桶里面，如果这时进行树化，树化本身就是一个耗时的过程。时间复杂度会增加，性能下降，不如直接进行扩容，空间换时间。
终于扩容过程写完了。很经典，想读懂也很费劲。
六、get过程源码
1 public V get(Object key) {
2 Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
3 int h = spread(key.hashCode());//hash
4 if ((tab = table) != null && (n = tab.length) > 0 &&
5 (e = tabAt(tab, (n - 1) & h)) != null) {//取桶
6 if ((eh = e.hash) == h) {//key相同直接返回
7 if ((ek = e.key) == key || (ek != null && key.equals(ek)))
8 return e.val;
9 }
10 else if (eh < 0)//hash < 0 表示正在扩容
11 //在这里需要非常注意的一点，扩容后的桶会放入fwd节点
12 //该节点hash = MOVED，fwd.nextTable为扩容后新的数组。
13 return (p = e.find(h, key)) != null ? p.val : null;
14 while ((e = e.next) != null) {//迭代链表
15 if (e.hash == h &&
16 ((ek = e.key) == key || (ek != null && key.equals(ek))))
17 return e.val;
18 }
19 }
20 return null;
21 }
get源码只关注下面这行
return (p = e.find(h, key)) != null ? p.val : null;
当该桶已经被移动，则通过e.find方法去nextTab新数组查找。首先在5章节resize扩容方法中，已经扩容的桶会被塞进去一个ForwardingNode节点 setTabAt(tab, i, fwd); 继续看resize方法中ForwardingNode的初始化会发现是这样初始化的 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); ，看它的构造方法
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;
}
}
不难发下其初始化方法接收一个nextTab也就是扩容后的新数组，并将该数组赋值给其内部变量nextTable。也就是说当get发现桶已经扩容后，我们可以从fwd节点中找到新的数组。并从新的数组中找到新的目标桶并进行元素查找。
看了以上代码，回到 e.find(h, key)) ，需要明确的是e就是ForwardingNode节点。看看find方法
1 Node<K,V> find(int h, Object k) {
2 // loop to avoid arbitrarily deep recursion on forwarding nodes
3 outer: for (Node<K,V>[] tab = nextTable;;) {//迭代nextTable
4 Node<K,V> e; int n;
5 if (k == null || tab == null || (n = tab.length) == 0 ||
6 (e = tabAt(tab, (n - 1) & h)) == null)
7 return null;
8 for (;;) {
9 int eh; K ek;
10 if ((eh = e.hash) == h &&
11 ((ek = e.key) == k || (ek != null && k.equals(ek))))
12 return e;
13 if (eh < 0) {
14 if (e instanceof ForwardingNode) {
15 tab = ((ForwardingNode<K,V>)e).nextTable;
16 continue outer;
17 }
18 else
19 return e.find(h, k);
20 }
21 if ((e = e.next) == null)
22 return null;
23 }
24 }
25 }
OK，很明确了，确实是从nextTable中查找的。
得出一个结论，ConcurrentHashMap扩容不影响get操作。也就是在扩容过程中可以并发读。
七、ConcurrentHashMap并发控制
详细看了ConcurrentHashMap put resize get过程的源码，本章从整体上看一下ConcurrentHashMap的并发控制。
下面结合图片我们看一下ConcurrentHashMap的并发过程。
如上图，线程1进行put操作，这时发现size > sizeCtl。开始进行扩容
此时线程1已经完成oldTab中索引[2，16)中的扩容。正在进行索引为1的桶的扩容。接下来线程2执行get。
线程2根据get逻辑和key的hash，可能访问的三种情况如上图所示
情况一：访问蓝色号桶，即未扩容的桶。该桶还未进行扩容，所以在桶中找到对应元素，返回。
情况二：访问绿色桶，即正在扩容的桶。该桶正在扩容，在扩容过程中，线程1持有Synchronized锁，线程2只能自旋等待。
情况三：访问橘色桶，该桶已扩容的桶。该桶已扩容，oldTab中是fwd节点，hash=-1，所以执行fwd节点的find逻辑，fwd节点持有newTab(nextTable)，所以线程2去newTab中查找对应元素，返回。
如上图4，当线程1进行扩容时，线程3进来执行put，同样存在三种可能的情况
情况一：访问蓝色桶，即未扩容的桶。正常执行put逻辑。
情况二：访问绿色桶，即正扩容的桶。因为线层1持有Synchronized锁，线程3将一直自旋，等待扩容结束。
情况三：访问橘色桶，即已扩容的桶。因为已扩容的桶，在oldTab中是fwd节点，hash = -1 = MOVED，所以线程3执行帮助扩容的逻辑。等待扩容完成，线程3继续完成put逻辑。
OK，以上就是ConcurrentHashMap关于get put resize的并发控制，从以上过程可见，存在锁竞争的情况很有限，即使存在锁竞争，也是进行自旋，而不会阻塞线程。可见ConcurrentHashMap能做到高效的并发读。
在put过程中，因为如果存在线程正在已经扩容，则帮助进行扩容（协助扩容这块，有一个步长的概念，同时进行扩容的线程和table的长度有关）。如果当前桶正在进行扩容，则被Synchronized锁拒之门外，自旋等待扩容结束。如果访问的是未扩容的桶，则执行正常的put逻辑。可见整个过程中，由于锁的粒度很小，put做到了高效的并发写，也做到了高效的扩容。
总之一句话ConcurrentHashMap的高并发是通过 CAS乐观锁 + 自旋锁 + 细粒度 保证的。
如有错误的地方还请留言指正。
原创不易，转载请注明原文地址：https://www.cnblogs.com/hello-shf/p/12183263.html
加载全部内容