深入并发包 ConcurrentHashMap 源码解析
Posted java-chen-hao
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了深入并发包 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 内部使用的。
1 public ConcurrentHashMap(int initialCapacity, 2 float loadFactor, int concurrencyLevel) { 3 if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0) 4 throw new IllegalArgumentException(); 5 if (concurrencyLevel > MAX_SEGMENTS) 6 concurrencyLevel = MAX_SEGMENTS; 7 // Find power-of-two sizes best matching arguments 8 int sshift = 0; 9 int ssize = 1; 10 // 计算并行级别 ssize，因为要保持并行级别是 2 的 n 次方 11 while (ssize < concurrencyLevel) { 12 ++sshift; 13 ssize <<= 1; 14 } 15 // 我们这里先不要那么烧脑，用默认值，concurrencyLevel 为 16，sshift 为 4 16 // 那么计算出 segmentShift 为 28，segmentMask 为 15，后面会用到这两个值 17 this.segmentShift = 32 - sshift; 18 this.segmentMask = ssize - 1; 19 20 if (initialCapacity > MAXIMUM_CAPACITY) 21 initialCapacity = MAXIMUM_CAPACITY; 22 23 // initialCapacity 是设置整个 map 初始的大小， 24 // 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小 25 // 如 initialCapacity 为 64，那么每个 Segment 或称之为"槽"可以分到 4 个 26 int c = initialCapacity / ssize; 27 if (c * ssize < initialCapacity) 28 ++c; 29 // 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2，这个值也是有讲究的，因为这样的话，对于具体的槽上， 30 // 插入一个元素不至于扩容，插入第二个的时候才会扩容 31 int cap = MIN_SEGMENT_TABLE_CAPACITY; 32 while (cap < c) 33 cap <<= 1; 34 35 // 创建 Segment 数组， 36 // 并创建数组的第一个元素 segment[0] 37 Segment<K,V> s0 = 38 new Segment<K,V>(loadFactor, (int)(cap * loadFactor), 39 (HashEntry<K,V>[])new HashEntry[cap]); 40 Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize]; 41 // 往数组写入 segment[0] 42 UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0] 43 this.segments = ss; 44 }
初始化完成，我们得到了一个 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，姑且把它们简单翻译为移位数和掩码，这两个值马上就会用到
put 过程分析
我们先看 put 的主流程，对于其中的一些关键细节操作，后面会进行详细介绍。
1 public V put(K key, V value) { 2 Segment<K,V> s; 3 if (value == null) 4 throw new NullPointerException(); 5 // 1. 计算 key 的 hash 值 6 int hash = hash(key); 7 // 2. 根据 hash 值找到 Segment 数组中的位置 j 8 // hash 是 32 位，无符号右移 segmentShift(28) 位，剩下高 4 位， 9 // 然后和 segmentMask(15) 做一次与操作，也就是说 j 是 hash 值的高 4 位，也就是槽的数组下标 10 int j = (hash >>> segmentShift) & segmentMask; 11 // 刚刚说了，初始化的时候初始化了 segment[0]，但是其他位置还是 null， 12 // ensureSegment(j) 对 segment[j] 进行初始化 13 if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck 14 (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment 15 s = ensureSegment(j); 16 // 3. 插入新值到 槽 s 中 17 return s.put(key, hash, value, false); 18 }
初始化槽: ensureSegment
ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0]，对于其他槽来说，在插入第一个值的时候进行初始化。
这里需要考虑并发，因为很可能会有多个线程同时进来初始化同一个槽 segment[k]，不过只要有一个成功了就可以。
1 private Segment<K,V> ensureSegment(int k) { 2 final Segment<K,V>[] ss = this.segments; 3 long u = (k << SSHIFT) + SBASE; // raw offset 4 Segment<K,V> seg; 5 if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { 6 // 这里看到为什么之前要初始化 segment[0] 了， 7 // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k] 8 // 为什么要用“当前”，因为 segment[0] 可能早就扩容过了 9 Segment<K,V> proto = ss[0]; 10 int cap = proto.table.length; 11 float lf = proto.loadFactor; 12 int threshold = (int)(cap * lf); 13 14 // 初始化 segment[k] 内部的数组 15 HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap]; 16 if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) 17 == null) { // 再次检查一遍该槽是否被其他线程初始化了。 18 19 Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); 20 // 使用 while 循环，内部用 CAS，当前线程成功设值或其他线程成功设值后，退出,如果其他线程成功设置后，这里获取到直接返回 21 while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) 22 == null) { 23 if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) 24 break; 25 } 26 } 27 } 28 return seg; 29 }
总的来说，ensureSegment(int k) 比较简单，对于并发操作使用 CAS 进行控制。
第一层很简单，根据 hash 值很快就能找到相应的 Segment，之后就是 Segment 内部的 put 操作了。
Segment 内部是由 数组+链表 组成的。
1 final V put(K key, int hash, V value, boolean onlyIfAbsent) { 2 // 在往该 segment 写入前，需要先获取该 segment 的独占锁 3 // 先看主流程，后面还会具体介绍这部分内容 4 HashEntry<K,V> node = tryLock() ? null : 5 scanAndLockForPut(key, hash, value); 6 V oldValue; 7 try { 8 // 这个是 segment 内部的数组 9 HashEntry<K,V>[] tab = table; 10 // 再利用 hash 值，求应该放置的数组下标 11 int index = (tab.length - 1) & hash; 12 // first 是数组该位置处的链表的表头 13 HashEntry<K,V> first = entryAt(tab, index); 14 15 // 下面这串 for 循环虽然很长，不过也很好理解，想想该位置没有任何元素和已经存在一个链表这两种情况 16 for (HashEntry<K,V> e = first;;) { 17 if (e != null) { 18 K k; 19 if ((k = e.key) == key || 20 (e.hash == hash && key.equals(k))) { 21 oldValue = e.value; 22 if (!onlyIfAbsent) { 23 // 覆盖旧值 24 e.value = value; 25 ++modCount; 26 } 27 break; 28 } 29 // 继续顺着链表走 30 e = e.next; 31 } 32 else { 33 // node 到底是不是 null，这个要看获取锁的过程，不过和这里都没有关系。 34 // 如果不为 null，那就直接将它设置为链表表头；如果是null，初始化并设置为链表表头。 35 if (node != null) 36 node.setNext(first); 37 else 38 node = new HashEntry<K,V>(hash, key, value, first); 39 40 int c = count + 1; 41 // 如果超过了该 segment 的阈值，这个 segment 需要扩容 42 if (c > threshold && tab.length < MAXIMUM_CAPACITY) 43 rehash(node); // 扩容后面也会具体分析 44 else 45 // 没有达到阈值，将 node 放到数组 tab 的 index 位置， 46 // 其实就是将新的节点设置成原链表的表头 47 setEntryAt(tab, index, node); 48 ++modCount; 49 count = c; 50 oldValue = null; 51 break; 52 } 53 } 54 } finally { 55 // 解锁 56 unlock(); 57 } 58 return oldValue; 59 }
整体流程还是比较简单的，由于有独占锁的保护，所以 segment 内部的操作并不复杂。至于这里面的并发问题，我们稍后再进行介绍。
到这里 put 操作就结束了，接下来，我们说一说其中几步关键的操作。
获取写入锁: scanAndLockForPut
前面我们看到，在往某个 segment 中 put 的时候，首先会调用 node = tryLock() ? null : scanAndLockForPut(key, hash, value)，也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁，如果失败，那么进入到 scanAndLockForPut 这个方法来获取锁。
下面我们来具体分析这个方法中是怎么控制加锁的。
1 private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) { 2 HashEntry<K,V> first = entryForHash(this, hash); 3 HashEntry<K,V> e = first; 4 HashEntry<K,V> node = null; 5 int retries = -1; // negative while locating node 6 7 // 循环获取锁 8 while (!tryLock()) { 9 HashEntry<K,V> f; // to recheck first below 10 if (retries < 0) { 11 if (e == null) { 12 if (node == null) // speculatively create node 13 // 进到这里说明数组该位置的链表是空的，没有任何元素 14 // 当然，进到这里的另一个原因是 tryLock() 失败，所以该槽存在并发，不一定是该位置 15 node = new HashEntry<K,V>(hash, key, value, null); 16 retries = 0; 17 } 18 else if (key.equals(e.key)) 19 retries = 0; 20 else 21 // 顺着链表往下走 22 e = e.next; 23 } 24 // 重试次数如果超过 MAX_SCAN_RETRIES（单核1多核64），那么不抢了，进入到阻塞队列等待锁 25 // lock() 是阻塞方法，直到获取锁后返回 26 else if (++retries > MAX_SCAN_RETRIES) { 27 lock(); 28 break; 29 } 30 else if ((retries & 1) == 0 && 31 // 这个时候是有大问题了，那就是有新的元素进到了链表，成为了新的表头 32 // 所以这边的策略是，相当于重新走一遍这个 scanAndLockForPut 方法 33 (f = entryForHash(this, hash)) != first) { 34 e = first = f; // re-traverse if entry changed 35 retries = -1; 36 } 37 } 38 return node; 39 }
这个方法有两个出口，一个是 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 的独占锁的。
1 // 方法参数上的 node 是这次扩容后，需要添加到新的数组中的数据。 2 private void rehash(HashEntry<K,V> node) { 3 HashEntry<K,V>[] oldTable = table; 4 int oldCapacity = oldTable.length; 5 // 2 倍 6 int newCapacity = oldCapacity << 1; 7 threshold = (int)(newCapacity * loadFactor); 8 // 创建新数组 9 HashEntry<K,V>[] newTable = 10 (HashEntry<K,V>[]) new HashEntry[newCapacity]; 11 // 新的掩码，如从 16 扩容到 32，那么 sizeMask 为 31，对应二进制 ‘000...00011111’ 12 int sizeMask = newCapacity - 1; 13 14 // 遍历原数组，老套路，将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置 15 for (int i = 0; i < oldCapacity ; i++) { 16 // e 是链表的第一个元素 17 HashEntry<K,V> e = oldTable[i]; 18 if (e != null) { 19 HashEntry<K,V> next = e.next; 20 // 计算应该放置在新数组中的位置， 21 // 假设原数组长度为 16，e 在 oldTable[3] 处，那么 idx 只可能是 3 或者是 3 + 16 = 19 22 int idx = e.hash & sizeMask; 23 if (next == null) // 该位置处只有一个元素，那比较好办 24 newTable[idx] = e; 25 else { // Reuse consecutive sequence at same slot 26 // e 是链表表头 27 HashEntry<K,V> lastRun = e; 28 // idx 是当前链表的头结点 e 的新位置 29 int lastIdx = idx; 30 31 // 下面这个 for 循环会找到一个 lastRun 节点，这个节点之后的所有元素是将要放到一起的 32 for (HashEntry<K,V> last = next; 33 last != null; 34 last = last.next) { 35 int k = last.hash & sizeMask; 36 if (k != lastIdx) { 37 lastIdx = k; 38 lastRun = last; 39 } 40 } 41 // 将 lastRun 及其之后的所有节点组成的这个链表放到 lastIdx 这个位置 42 newTable[lastIdx] = lastRun; 43 // 下面的操作是处理 lastRun 之前的节点， 44 // 这些节点可能分配在另一个链表中，也可能分配到上面的那个链表中 45 for (HashEntry<K,V> p = e; p != lastRun; p = p.next) { 46 V v = p.value; 47 int h = p.hash; 48 int k = h & sizeMask; 49 HashEntry<K,V> n = newTable[k]; 50 newTable[k] = new HashEntry<K,V>(h, p.key, v, n); 51 } 52 } 53 } 54 } 55 // 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部 56 int nodeIndex = node.hash & sizeMask; // add the new node 57 node.setNext(newTable[nodeIndex]); 58 newTable[nodeIndex] = node; 59 table = newTable; 60 }
总结一下put的流程：
当执行put操作时，会进行第一次key的hash来定位Segment的位置，如果该Segment还没有初始化，即通过CAS操作进行赋值，然后进行第二次hash操作，找到相应的HashEntry的位置，这里会利用继承过来的锁的特性，在将数据插入指定的HashEntry位置时（链表的尾端），会通过继承ReentrantLock的tryLock（）方法尝试去获取锁，如果获取成功就直接插入相应的位置，如果已经有线程获取该Segment的锁，那当前线程会以自旋的方式去继续的调用tryLock（）方法去获取锁，超过指定次数就挂起，等待唤醒。
get 过程分析
相对于 put 来说，get 真的不要太简单。
- 计算 hash 值，找到 segment 数组中的具体位置，或我们前面用的“槽”
- 槽中也是一个数组，根据 hash 找到数组中具体的位置
- 到这里是链表了，顺着链表进行查找即可
1 public V get(Object key) { 2 Segment<K,V> s; // manually integrate access methods to reduce overhead 3 HashEntry<K,V>[] tab; 4 // 1. hash 值 5 int h = hash(key); 6 long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; 7 // 2. 根据 hash 找到对应的 segment 8 if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && 9 (tab = s.table) != null) { 10 // 3. 找到segment 内部数组相应位置的链表，遍历 11 for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile 12 (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE); 13 e != null; e = e.next) { 14 K k; 15 if ((k = e.key) == key || (e.hash == h && key.equals(k))) 16 return e.value; 17 } 18 } 19 return null; 20 }
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。该计算方法代码如下
1 public int size() { 2 final Segment<K,V>[] segments = this.segments; 3 int size; 4 boolean overflow; // true if size overflows 32 bits 5 long sum; // sum of modCounts 6 long last = 0L; // previous sum 7 int retries = -1; // first iteration isn‘t retry 8 try { 9 for (;;) { 10 if (retries++ == RETRIES_BEFORE_LOCK) { 11 for (int j = 0; j < segments.length; ++j) 12 ensureSegment(j).lock(); // force creation 13 } 14 sum = 0L; 15 size = 0; 16 overflow = false; 17 for (int j = 0; j < segments.length; ++j) { 18 Segment<K,V> seg = segmentAt(segments, j); 19 if (seg != null) { 20 sum += seg.modCount; 21 int c = seg.count; 22 if (c < 0 || (size += c) < 0) 23 overflow = true; 24 } 25 } 26 if (sum == last) 27 break; 28 last = sum; 29 } 30 } finally { 31 if (retries > RETRIES_BEFORE_LOCK) { 32 for (int j = 0; j < segments.length; ++j) 33 segmentAt(segments, j).unlock(); 34 } 35 } 36 return overflow ? Integer.MAX_VALUE : size; 37 }
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各自持有一把锁。在保证线程安全的同时降低了锁的粒度，让并发操作效率更高。
以上是关于深入并发包 ConcurrentHashMap 源码解析的主要内容，如果未能解决你的问题，请参考以下文章
JDK源码那些事儿之并发ConcurrentHashMap上篇