源码分析——ConcurrentHashMap

前言

上一篇文章我讲了一下HashMap的相关源码实现，并且我们知道它是线程不安全的，在并发环境中使用时，HashMap在扩容的时候有可能会生成一个环形链表，从而导致get形成死循环超时。那这篇我们就来介绍一下并发环境下使用的HashMap——ConcurrentHashMap，下面是它的类关系图。

JDK1.7中的实现

JDK1.7 中的ConcurrentHashMap采用了分段锁的设计，先来看一下它的数据结构。ConcurrentHashMap中含有一个Segment数组。每个Segment中又含有一个HashEntry数组。

Segment是一种可重入锁，在ConcurrentHashMap里扮演锁的角色；HashEntry则用于存储键值对数据。

一个ConcurrentHashMap里包含一个Segment数组。Segment的结构和HashMap类似，是一种数组和链表结构。一个Segment里包含一个HashEntry数组，每一个HashEntry是一个链表结构的元素，每个Segment守护着一个HashEntry数组里的元素，当对HashEntry数组的数据进行修改时，必须首先获得与它对应的Segment锁。

ConcurrentHashMap通过使用分段锁技术，将数据分成一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也能被其他线程访问，能够实现真正的并发访问。

来看看上述Segment结构的定义：

static final class Segment extends ReentrantLock implements Serializable {private static final long serialVersionUID = 2249069246763182397L;static final int MAX_SCAN_RETRIES =Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;transient volatile HashEntry[] table;transient int count;transient int modCount;transient int threshold;final float loadFactor;... ...
}复制代码

1. 存储结构

static final class HashEntry<K,V> {final int hash;final K key;volatile V value;volatile HashEntry<K,V> next;
}
复制代码

ConcurrentHashMap 和 HashMap 实现上类似，最主要的差别是 ConcurrentHashMap 采用了分段锁（Segment），每个分段锁维护着几个桶（HashEntry），多个线程可以同时访问不同分段锁上的桶，从而使其并发度更高（并发度就是 Segment 的个数）。

Segment 继承自 ReentrantLock。

static final class Segment<K,V> extends ReentrantLock implements Serializable {private static final long serialVersionUID = 2249069246763182397L;static final int MAX_SCAN_RETRIES =Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;transient volatile HashEntry<K,V>[] table;transient int count;transient int modCount;transient int threshold;final float loadFactor;
}
final Segment<K,V>[] segments;复制代码

不看下面的方法，可以看到几个熟悉的字段。HashEntry(哈希数组)，threshold(扩容阈值)，loadFactor(负载因子)表示segment是一个完整的HashMap。

接下来我们看看ConcurrentHashMap的构造函数

public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)复制代码

三个参数分别代表了：

初始容量：初始容量表示所有的segment数组中，一共含有多少个hashentry。若initialCapacity不为2的幂，会取一个大于initialCapacity的2的幂。
负载因子：默认0.75。
并发级别：可以同时允许多少个线程并发。concurrencyLevel为多少，就有多少个segment，当然也会取一个大于等于这个值的2的幂。

默认的并发级别为 16，也就是说默认创建 16 个 Segment。

static final int DEFAULT_CONCURRENCY_LEVEL = 16;复制代码

接下来我们看一下ConcurrentHashMap中的几个关键函数，get，put，rehash(扩容), size方法，看看他是如何实现并发的。

2. get 操作

get实现过程：

根据key，计算出hashCode；
根据步骤1计算出的hashCode定位segment，如果segment不为null && segment.table也不为null，跳转到步骤3，否则，返回null，该key所对应的value不存在；
根据hashCode定位table中对应的hashEntry，遍历hashEntry，如果key存在，返回key对应的value；
步骤3结束仍未找到key所对应的value，返回null，该key对应的value不存在。

比起Hashtable，ConcurrentHashMap的get操作高效之处在于整个get操作不需要加锁。如果不加锁，ConcurrentHashMap的get操作是如何做到线程安全的呢？原因是volatile，所有的value都定义成了volatile类型，volatile可以保证线程之间的可见性，这也是用volatile替换锁的经典应用场景。

3. put操作

ConcurrentHashMap提供两个方法put和putIfAbsent来完成put操作，它们之间的区别在于put方法做插入时key存在会更新key所对应的value，而putIfAbsent不会更新。

put实现过程：

参数校验，value不能为null，为null时抛出空指针异常；
计算key的hashCode；
定位segment，如果segment不存在，创建新的segment；
调用segment的put方法在对应的segment做插入操作。

putIfAbsent实现过程：

putIfAbsent的执行过程与put方法是一致的，除了最后调用的segment的put方法参数onlyIfAbsent传参不一样。

segment的put方法实现

segment的put方法是整个put操作的核心，它实现了在segment的HashEntry数组中做插入（segment的HashEntry数组采用拉链法来处理冲突）。

segment put实现过程：

1. 获取锁，保证put操作的线程安全；
2. 定位到HashEntry数组中具体的HashEntry；
3. 遍历HashEntry链表，假若待插入key已存在：

需要更新key所对应value（!onlyIfAbsent），更新oldValue=newValue，跳转到步骤5；
否则，直接跳转到步骤5；

4. 遍历完HashEntry链表，key不存在，插入HashEntry节点，oldValue=null，跳转到步骤5；
5. 释放锁，返回oldValue。

步骤4在做插入的时候实际上经历了两个步骤：

第一：HashEntry数组扩容；

是否需要扩容

在插入元素前会先判断Segment的HashEntry数组是否超过threshold，如果超过阀值，则需要对HashEntry数组扩容；

如何扩容

在扩容的时候，首先创建一个容量是原来容量两倍的数组，将原数组的元素再散列后插入到新的数组里。为了高效，ConcurrentHashMap只对某个Segment进行扩容，不会对整个容器扩容。

第二：定位添加元素对应的位置，然后将其放到HashEntry数组中。

4. size 操作

每个 Segment 维护了一个 count 变量来统计该 Segment 中的键值对个数。

/*** The number of elements. Accessed only either within locks* or among other volatile reads that maintain visibility.*/
transient int count;复制代码

在执行 size 操作时，需要遍历所有 Segment 然后把 count 累计起来。

ConcurrentHashMap 在执行 size 操作时先尝试不加锁，如果连续两次不加锁操作得到的结果一致，那么可以认为这个结果是正确的。

尝试次数使用 RETRIES_BEFORE_LOCK 定义，该值为 2，retries 初始值为 -1，因此尝试次数为 3。

如果尝试的次数超过 3 次，就需要对每个 Segment 加锁。

static final int RETRIES_BEFORE_LOCK = 2;
public int size() {// Try a few times to get accurate count. On failure due to// continuous async changes in table, resort to locking.final Segment<K,V>[] segments = this.segments;int size;boolean overflow; // true if size overflows 32 bitslong sum;         // sum of modCountslong last = 0L;   // previous sumint retries = -1; // first iteration isn't retrytry {for (;;) {// 超过尝试次数，则对每个 Segment 加锁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;
}复制代码

由于在累加count的操作的过程中之前累加过的count发生变化的几率非常小，所以ConcurrentHashMap先尝试2次不锁住Segment的方式来统计每个Segment的大小，如果在统计的过程中Segment的count发生了变化，这时候再加锁统计Segment的count。

ConcurrentHashMap如何判断统计过程中Segment的cout发生了变化？

Segment使用变量modCount来表示Segment大小是否发生变化，在put/remove/clean操作里都会将modCount加1，那么在统计size的前后只需要比较modCount是否发生了变化，如果发生变化，Segment的大小肯定发生了变化。

JDK 1.8 的改动

JDK 1.7 使用分段锁机制来实现并发更新操作，核心类为 Segment，它继承自重入锁 ReentrantLock，并发度与 Segment 数量相等。

JDK 1.8 使用了 CAS 操作来支持更高的并发度，在 CAS 操作失败时使用内置锁 synchronized。

1.7中Segment[]最大是16，也就是最大支持16个并发。1.8改成Node[]，课并行的数量远远大于16。

并且 JDK 1.8 的实现也在链表过长时会转换为红黑树，这点与HashMap1.8的实现是一样的。

数据结构采用数组 + 链表 + 红黑树的方式实现。当链表中(bucket)的节点个数超过8个时，会转换成红黑树的数据结构存储，这样设计的目的是为了提高同一个链表冲突过大情况下的读取效率。

Java8中主要做了如下优化:

将Segment抛弃掉了，直接采用Node（继承自Map.Entry）作为table元素。
修改时，不再采用ReentrantLock加锁，直接用内置synchronized加锁，Java8的内置锁比之前版本优化了很多，相较ReentrantLock，性能不并差。
size方法优化，增加CounterCell内部类，用于并行计算每个bucket的元素数量。

JDK1.8中，出现了较大的改动。没有使用段锁，改成了Node数组 + 链表 + 红黑树的方式。

其中有个重要的变量：sizeCtl

负数表示正在进行初始化或者扩容，-1表示正在初始化，-N表示有N - 1个线程正在扩容
正数0，表示还没有被初始化。其他正数表示下一次扩容的大小。

Node核心数据结构：

static class Node<K,V> implements Map.Entry<K,V> {final int hash;final K key;volatile V val;volatile Node<K,V> next;
}复制代码

还有两个数据结构TreeNode、TreeBin，用来当链表的大小超过阈值的时候，将链表变作红黑树。

CAS 操作

这一版本大量使用了CAS操作。所谓的CAS就是，比较内存对应的区域的值，和期望值是不是相等，如果相等，就设置一个新的值进去。

一般是这样使用，先获取对象中的某个域的值，并以这个值为期望值去调用CAS算法。

ConcurrentHashMap中有三个核心的CAS操作

tabAt：获得数组中位置i上的节点
casTabAt：设置数组位置i上的节点
setTabAt：利用volatile设置位置i上的节点。

//获取索引i处Node
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);
}
//利用CAS算法设置i位置上的Node节点（将c和table[i]比较，相同则插入v）
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);
}
//利用volatile设置节点位置i的值，仅在上锁区被调用
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {  U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}  复制代码

initTable() 方法

private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;while ((tab = table) == null || tab.length == 0) {//如果一个线程发现sizeCtl<0，意味着另外的线程//执行CAS操作成功，当前线程只需要让出cpu时间片if ((sc = sizeCtl) < 0)Thread.yield();else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//CAS方法把sizectl置为-1，表示本线程正在进行初始化try {if ((tab = table) == null || tab.length == 0) {//DEFAULT_CAPACITY 默认初始容量是 16int n = (sc > 0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")//初始化数组，长度为 16 或初始化时提供的长度Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//将这个数组赋值给 table，table 是 volatile 的table = tab = nt;//如果 n 为 16 的话，那么这里 sc = 12//其实就是 0.75 * nsc = n - (n >>> 2);}} finally {sizeCtl = sc;}break;}}return tab;
}复制代码

在put方法调用时，回去判断table是不是为null，若为null就调用initTable去初始化。

调用initTable会判断sizeCtl的值，若值为-1则表示正在初始化，会调用yield()去等待。

若值为0，这时先调用CAS算法去设置为-1，再初始化。

所以执行第一次put操作的线程会执行Unsafe.compareAndSwapInt方法修改sizeCtl为-1，有且只有一个线程能够修改成功，其它线程通过Thread.yield()让出CPU时间片等待table初始化完成。

综上所述，可以知道初始化是单线程操作。

put()方法

假设table已经初始化完成，put操作采用CAS+synchronized实现并发插入或更新操作，具体实现如下。

public V put(K key, V value) {return putVal(key, value, false);
}/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {//不允许key、value为空if (key == null || value == null) throw new NullPointerException();//返回 (h ^ (h >>> 16)) & HASH_BITS;int hash = spread(key.hashCode());int binCount = 0;//循环，直到插入成功for (Node[] tab = table;;) {Node f; int n, i, fh;if (tab == null || (n = tab.length) == 0)//table为空，初始化tabletab = initTable();else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//索引处无值if (casTabAt(tab, i, null,new Node(hash, key, value, null)))break;                   // no lock when adding to empty bin}else if ((fh = f.hash) == MOVED)// MOVED=-1;//检测到正在扩容，则帮助其扩容tab = helpTransfer(tab, f);else {V oldVal = null;//上锁（hash值相同的链表的头节点）synchronized (f) {if (tabAt(tab, i) == f) {if (fh >= 0) {//遍历链表节点binCount = 1;for (Node e = f;; ++binCount) {K ek;// hash和key相同，则修改valueif (e.hash == hash &&((ek = e.key) == key ||(ek != null && key.equals(ek)))) {oldVal = e.val;//仅putIfAbsent()方法中onlyIfAbsent为trueif (!onlyIfAbsent)//putIfAbsent()包含key则返回get，否则put并返回e.val = value;break;}Node pred = e;//已遍历到链表尾部，直接插入if ((e = e.next) == null) {pred.next = new Node(hash, key,value, null);break;}}}else if (f instanceof TreeBin) {// 树节点Node p;binCount = 2;if ((p = ((TreeBin)f).putTreeVal(hash, key,value)) != null) {oldVal = p.val;if (!onlyIfAbsent)p.val = value;}}}}if (binCount != 0) {//判断是否要将链表转换为红黑树，临界值和HashMap一样也是8if (binCount >= TREEIFY_THRESHOLD)//若length<64,直接tryPresize,两倍table.length;不转树treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}addCount(1L, binCount);return null;
}复制代码

1. hash算法

static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;
}
复制代码

2. table中定位索引位置，n是table的大小

int index = (n - 1) & hash复制代码

3. 获取table中对应索引的元素f。

采用Unsafe.getObjectVolatile来获取。在java内存模型中，我们已经知道每个线程都有一个工作内存，里面存储着table的副本，虽然table是volatile修饰的，但不能保证线程每次都拿到table中的最新元素，Unsafe.getObjectVolatile可以直接获取指定内存的数据，保证了每次拿到数据都是最新的。

4. 如果f为null，说明table中这个位置第一次插入元素，利用Unsafe.compareAndSwapObject方法插入Node节点。

如果CAS成功，说明Node节点已经插入，break跳出，随后addCount(1L, binCount)方法会检查当前容量是否需要进行扩容。
如果CAS失败，说明有其它线程提前插入了节点，自旋重新尝试在这个位置插入节点。

5. 如果f的hash值为-1，说明当前f是ForwardingNode节点，意味有其它线程正在扩容，则一起进行扩容操作。

6. 其余情况把新的Node节点按链表或红黑树的方式插入到合适的位置，这个过程采用同步内置锁实现并发，代码如上。

在节点f上进行同步，节点插入之前，再次利用tabAt(tab, i) == f判断，防止被其它线程修改。

如果f.hash >= 0，说明f是链表结构的头结点，遍历链表，如果找到对应的node节点，则修改value，否则在链表尾部加入节点。
如果f是TreeBin类型节点，说明f是红黑树根节点，则在树结构上遍历元素，更新或增加节点。
如果链表中节点数binCount >= TREEIFY_THRESHOLD(默认是8)，则把链表转化为红黑树结构。

链表转红黑树: treeifyBin()

treeifyBin 不一定就会进行红黑树转换，也可能是仅仅做数组扩容。我们还是看源码吧。

private final void treeifyBin(Node[] tab, int index) {Node b; int n, sc;if (tab != null) {// MIN_TREEIFY_CAPACITY 为 64// 所以，如果数组长度小于 64 的时候，其实也就是 32 或者 16 或者更小的时候，会进行数组扩容if ((n = tab.length) < MIN_TREEIFY_CAPACITY)// 后面我们再详细分析这个方法tryPresize(n << 1);// b 是头结点else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {// 加锁synchronized (b) {if (tabAt(tab, index) == b) {// 下面就是遍历链表，建立一颗红黑树TreeNode hd = null, tl = null;for (Node e = b; e != null; e = e.next) {TreeNode p =new TreeNode(e.hash, e.key, e.val,null, null);if ((p.prev = tl) == null)hd = p;elsetl.next = p;tl = p;}// 将红黑树设置到数组相应位置中setTabAt(tab, index, new TreeBin(hd));}}}}
}复制代码

扩容：tryPresize()

如果说 Java8 ConcurrentHashMap 的源码不简单，那么说的就是扩容操作和迁移操作。

这里的扩容也是做翻倍扩容，扩容后数组容量为原来的 2 倍。

// 首先要说明的是，方法参数 size 传进来的时候就已经翻了倍了
private final void tryPresize(int size) {// c：size 的 1.5 倍，再加 1，再往上取最近的 2 的 n 次方。int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :tableSizeFor(size + (size >>> 1) + 1);int sc;while ((sc = sizeCtl) >= 0) {Node<K,V>[] tab = table; int n;// 这个 if 分支和之前说的初始化数组的代码基本上是一样的if (tab == null || (n = tab.length) == 0) {n = (sc > c) ? sc : c;if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if (table == tab) {@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = nt;sc = n - (n >>> 2); // 0.75 * n}} finally {sizeCtl = sc;}}}else if (c <= sc || n >= MAXIMUM_CAPACITY)break;else if (tab == table) {int rs = resizeStamp(n);if (sc < 0) {Node<K,V>[] nt;if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||transferIndex <= 0)break;// 2. 用 CAS 将 sizeCtl 加 1，然后执行 transfer 方法// 此时 nextTab 不为 nullif (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))transfer(tab, nt);}// 1. 将 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2)// 调用 transfer 方法，此时 nextTab 参数为 nullelse if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))transfer(tab, null);}}
}复制代码

这个方法的核心在于 sizeCtl 值的操作，首先将其设置为一个负数，然后执行 transfer(tab, null)，再下一个循环将 sizeCtl 加 1，并执行 transfer(tab, nt)，之后可能是继续 sizeCtl 加 1，并执行 transfer(tab, nt)。

至于transfer()方法的源码这里我就不分析了，它的大概功能就是将原来的 tab 数组的元素迁移到新的 nextTab 数组中。

get()方法

get方法不用加锁。利用CAS操作，可以达到无锁的访问。

public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;int h = spread(key.hashCode());if ((tab = table) != null && (n = tab.length) > 0 &&(e = tabAt(tab, (n - 1) & h)) != null) {//tabAt(i),获取索引i处Node// 判断头结点是否就是我们需要的节点if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}// 如果头结点的 hash<0，说明正在扩容，或者该位置是红黑树else if (eh < 0)return (p = e.find(h, key)) != null ? p.val : null;//遍历链表while ((e = e.next) != null) {if (e.hash == h &&((ek = e.key) == key || (ek != null && key.equals(ek))))return e.val;}}return null;
}复制代码

Node<K,V> find(int h, Object k) {Node<K,V> e = this;if (k != null) {do {K ek;if (e.hash == h &&((ek = e.key) == k || (ek != null && k.equals(ek))))return e;} while ((e = e.next) != null);}return null;
}复制代码

get() 执行过程：

1. 计算 hash 值

2. 根据 hash 值找到数组对应位置: (n – 1) & h

3. 根据该位置处结点性质进行相应查找

如果该位置为 null，那么直接返回 null 就可以了
如果该位置处的节点刚好就是我们需要的，返回该节点的值即可
如果该位置节点的 hash 值小于 0，说明正在扩容，或者是红黑树
如果以上 3 条都不满足，那就是链表，进行遍历比对即可

小结

到这里我就基本把ConcurrentHashMap在 JDK 1.7和1.8中的实现大概捋了一遍，并详细分析了几个重要的方法实现：初始化、put、get。在 JDK1.8中ConcurrentHashMap发生了较大的变化，通过使用CAS+synchronized的实现取代了原先 1.7 中的Segment分段锁机制，从而支持更高的并发量。

这只是我对ConcurrentHashMap的第二次学习，若想更好地理解掌握ConcurrentHashMap的实现之精妙，个人觉得还需以后再多看几次，相信每次都会有新的收获。

转载于:https://juejin.im/post/5d06f6cb6fb9a07ec42b5ae0