ConcurrentHashMap 源码解析

前言

学习 ConcurrentHashMap 的源码，可以得到 Doug Lea 大佬对于并发编程的理解，编写更加高效的线程安全的代码。ConcurrentHashMap 在 jdk 1.7 版本和 jdk 1.8 版本的实现不同，现在使用的都是 jdk 1.8 以上的版本，本文主要分析 jdk 1.8 版本的源码。考虑到知识的完整性，也会对 jdk 1.7 的代码做简要分析。

jdk 1.7

在 jdk 1.7 版本中，ConcurrentHashMap 使用分段锁，将 HashMap 分为若干个段，每个段中都有一个 HashMap。根据 key 的 hash 值，先确定 key 所在的段，然后再确定在 HashMap 的哪个桶中。对同一个 key 的并发访问，只需要锁住 key 所在的段，降低锁粒度，从而提升并发性能。ConcurrentHashMap 的数据结构如下图所示，

使用segment[]数组表示数据段，段的数组是不能扩容的，即段的大小在初始化的时候就确定了。每个段中的 HashMap 能够扩容，扩容机制与 HashMap 类似。

jdk 1.8

在 jdk 1.8 中，ConcurrentHashMap 的设计与 HashMap 保持一致，使用了链表和红黑树的组合，当一个链表上的节点数量超过 8 时，就会自动转换为红黑树。在代码上，ConcurrentHashMap 与 HashMap 的 get 和 put 方法类似，最大的不同是扩容。

构造方法

构造方法主要是初始化 ConcurrentHashMap 的容量，即所需桶的个数。构造方法有多个，本文主要讲述ConcurrentHashMap(int initialCapacity)，其它构造方法类似。ConcurrentHashMap 的构造方法和 HashMap 的几乎一致，如果了解 HashMap 的构造方法，这个步骤可以跳过。

public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    // 计算桶的数量，使得数量为2的整数幂
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
               MAXIMUM_CAPACITY :
               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    this.sizeCtl = cap;
}
1
2
3
4
5
6
7
8
9

tableSizeFor()方法是将任意设置的容量转换为 $2^n-1$ 的形式，将结果加一，就变成了 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;
}
1
2
3
4
5
6
7
8
9

2 的整数次幂的二进制形式为：2 -> 10，4 -> 100，8 -> 1000，依次类推。3~7行代码就是把任意的整数转换为有效位数都是 1，例如 10 -> 1010，经过位运算就变成 15 -> 1111，此时把 n + 1，即 15 + 1 = 16，就得到了大于 10 的最小 2 的整数次幂的数 16。2 行先把 c 减一，是防止 c 本身就是 2 的整数次幂，经过位运算变成了原来的 2 倍。如果 c = 0，即 n = -1，经过位运算后 n 仍然是 -1。计算机使用补码存储数字，-1 的补码全是 1，所以无论怎么移位，与 -1 取或运算，其值仍是 -1。
证明：
假设存在一个正整数 n，其二进制形式为xxxx xxx1x xxxx xxxx，因为正整数的二进制形式至少存在一个 1。当执行n |= n >>> 1后，n 的二进制形式就变成 xxxx xx11 xxxx xxxx，因为 1 与任何位取或都是 1。就相当于在原来 1 的右边增加了一个 1。执行n |= n >>> 2后，n 的二进制形式变成xxxx xx11 11xx xxxx，依次类推，执行完所有的或运算后，n 的最高位的 1 右边的位全部变成 1。

get

get 方法的主要步骤：

1. 计算出 key 的 hash 值。
2. 通过 hash 值确定 key 可能存在的桶的下标。
3. 遍历桶中的每个元素，寻找具有相同 key 的节点。

代码如下，

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // 计算哈希值
    int h = spread(key.hashCode());
    // 确定hash表是否初始化，key所在的桶是否有元素，如果都有则进入if
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        // 判断链表头节点是否就是要找的节点
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        // hash值为负数与扩容有关，等讲了扩容再回来说明，现在先忽略这个分支
        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;
        }
    }
    // 没找到相同key的节点
    return null;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

现在讲下 get 方法中，如何快速的根据 hash 值找到对应的桶的下标。首先是 spread(key.hashCode())方法，

static final int spread(int h) {
	// 将hash值的低16位图片与高16位进行异或计算，而高16位保持不变。
	// HASH_BITS=0x7fffffff，将hash值的符号位置为0，其它位不变，确保hash值非负。
    return (h ^ (h >>> 16)) & HASH_BITS;
}
1
2
3
4
5

这个操作是将 hash 值的低 16 位进行了第二次哈希计算，将低 16 位的值打散。因为计算下标的方式是(n - 1) & h)，n 为 2 的整数幂，其二进制形式为 00...100...00，而 n - 1其实就是一个掩码，其二进制形式为00...011...11，其实就是将 hash 值的低若干位取出来作为桶的下标，这就要求 hash 低位值要比较分散，这样才能尽可能的减少 hash 冲突。
讲到这里，大家就应该知道 ConcurrentHashMap 为什么要把容量设为 2 的整数幂了，就是为了能够快速找到桶的下标。HashMap 的桶下标计算也是这样设计的，一知双解。

put

put 方法就比较复杂了，需要考虑到扩容。

public V put(K key, V value) {
    return putVal(key, value, false);
}
1
2
3

put()方法调用了putVal()方法，这是 put 逻辑的核心方法，由于代码较长，不会贴出所有代码，会根据功能分开讲解。

final V putVal(K key, V value, boolean onlyIfAbsent) {
   	if (key == null || value == null) throw new NullPointerException();
   	int hash = spread(key.hashCode());
   	int binCount = 0;
	for (Node<K,V>[] tab = table;;) {
		// 核心逻辑
	}
	addCount(1L, binCount);
    return null;
}
1
2
3
4
5
6
7
8
9
10

核心逻辑分为如下步骤：
1）将所有的桶初始化

if (tab == null || (n = tab.length) == 0)
    tab = initTable();
1
2

tab 是一个数组，每个元素就是桶的头节点。是在第一次 put 的时候，这个数组才被创建出来，这其实是一种懒加载的策略。

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
    	// 表示其它线程正在初始化数组，当前线程就不用管了
        if ((sc = sizeCtl) < 0)
            Thread.yield(); // lost initialization race; just spin
        // CAS操作，将sizeCtl设为-1，表示当前线程正在初始化数组
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
            	// double-check
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    // sc = 0.75n，0.75就是默认的装填因子，map中的总元素数量达到该值就会触发扩容
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

经过这个步骤，map 创建完成，我们可以学习到，使用volatile声明的成员变量作为状态，通过CAS操作，使得多线程可以互斥的访问相同资源。
2）在 map 中寻找指定元素

// key所在的桶没有一个元素，那就直接将当前key作为头节点，put操作完成
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
	// CAS设置头节点，因为可能有其它线程也在设置该头节点
	if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
    	break;
}
1
2
3
4
5
6

如果桶中的元素不为空，那就需要从头节点开始遍历，注意，这里遍历的可能是链表也可能是红黑树。

// 表示map正在扩容
else if ((fh = f.hash) == MOVED)
	tab = helpTransfer(tab, f);
1
2
3

当 put 的时候如果发现正在扩容，则需要帮助 map 完成扩容，相当于扩容可能由多个线程来完成，这个方法等到扩容章节再讲解。
接下来就要从头节点开始遍历了，链表使用的是尾插法，即新节点作为链表最后一个节点。

else {
    V oldVal = null;
    // 开始遍历，锁住头节点，此时这个桶无法put其它节点了
    synchronized (f) {
    	// 在加锁前，头节点可能被删除，所以这里要再判断一下，确认锁住的是最新的头节点
        if (tabAt(tab, i) == f) {
        	// 头节点的hash值大于0表示这是一个正经的节点，如果小于0，与扩容有关，稍后再讲
            if (fh >= 0) {
            	// 记录当前桶中的节点数量，如果超过8，就要转换为红黑树
                binCount = 1;
                // 开始遍历
                for (Node<K,V> e = f;; ++binCount) {
                    K ek;
                    if (e.hash == hash &&
                        ((ek = e.key) == key ||
                         (ek != null && key.equals(ek)))) {
                         // 发现map中存在key了，就根据情况进行覆盖，如果是put方法进来的，onlyIfAbsent=false
                        oldVal = e.val;
                        if (!onlyIfAbsent)
                            e.val = value;
                        break;
                    }
                    // 找到最后一个节点还没有找到重复的key，则表示当前key是全新的，那就将节点加入链表末尾
                    Node<K,V> pred = e;
                    if ((e = e.next) == null) {
                        pred.next = new Node<K,V>(hash, key,
                                                  value, null);
                        break;
                    }
                }
            }
            // 如果头节点是树节点，则使用红黑树的查询方式
            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;
                }
            }
        }
    }
    if (binCount != 0) {
    	// 节点数超过8就变成红黑树
        if (binCount >= TREEIFY_THRESHOLD)
            treeifyBin(tab, i);
        if (oldVal != null)
            return oldVal;
        break;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

3）判断是否需要扩容

addCount(1L, binCount);

private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    // counterCells 记录每个桶中元素的个数，这个分支主要是 counterCells 的维护工作。
    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;
        // map 中节点总数
        s = sumCount();
    }
    // 桶中的节点数大于0，表示可能需要扩容
    if (check >= 0) {
        Node<K,V>[] tab, nt; int n, sc;
        // 当前map中的节点数超过了sizeCtl=容量*装载因子，并且能够扩容
        while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
               (n = tab.length) < MAXIMUM_CAPACITY) {
            // rs 没看懂是什么，这里我们关注扩容逻辑就行
            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);
            }
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

其中transfer()就是真正的扩容方法，第一个参数是当前的 map，第二个参数是扩容后的 map。

扩容

扩容是创建一个容量为原来 2 倍大小的桶数组，然后将原桶数组的元素迁移到新的桶数组上，迁移的最小单位是桶。由于数组可能比较大，这个搬迁的过程不是一蹴而就的，需要多个线程共同努力完成，前面put()方法中的helpTransfer()加入到扩容的工作中来，每个线程负责数组的一部分，这样就不会冲突。
在扩容的过程中可能出现一部分节点在旧的数组中，一部分节点在新的数组中，为了确保能够获取到数据，当把一个桶的节点迁移之后，会在旧桶的对应位置留下一个哨兵节点，如果碰到这个节点，就知道要去新的数组中查询。
1）初始化新数组

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    // NCPU 表示当前机器的 CPU 核心数，计算扩容的布长，一个线程负责一个布长范围内的桶的迁移工作，布长最小为 16
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range
    // 创建扩容后的新数组
    if (nextTab == null) {            // initiating
        try {
            // 创建原数组两倍大小的新数组
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            nextTab = nt;
        // 需要注意 OOM 的问题
        } catch (Throwable ex) {
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        nextTable = nextTab;
        // 确认开始迁移的桶的下标
        transferIndex = n;
    }
    int nextn = nextTab.length;
    // 留在旧数组的哨兵节点，其hash值为-1
    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;;) {
		// 扩容核心逻辑
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

2）确定每个线程迁移桶的下标的范围，从下标大的桶开始迁移。

boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab}
for (int i = 0, bound = 0;;) {
    Node<K,V> f; int fh;
    // 先看最后一个else if
    while (advance) {
        int nextIndex, nextBound;
        // 当前桶迁移完成，开始迁移下一个桶
        if (--i >= bound || finishing)
            advance = false;
        // 说明迁移完成了
        else if ((nextIndex = transferIndex) <= 0) {
            i = -1;
            advance = false;
        }
        // 第一次进while循环通常从这里开始，前面那个if是给变量赋值nextIndex = transferIndex
        // 迁移的下界是bound，上界为nextIndex - 1
        // 其中transferIndex是全局变量，通过CAS操作，就能为每个线程分配到不同的迁移段
        else if (U.compareAndSwapInt
                 (this, TRANSFERINDEX, nextIndex,
                  nextBound = (nextIndex > stride ?
                               nextIndex - stride : 0))) {
            bound = nextBound;
            i = nextIndex - 1;
            advance = false;
        }
    }
    // 后续步骤
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

3）边界条件判断

for (int i = 0, bound = 0;;) {
	Node<K,V> f; int fh;
	// 所有桶都迁移完成，将新map的信息赋值到当前map
	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
	    }
	// 如果下标为i的桶完成迁移，就在原位置增加一个哨兵节点
	} else if ((f = tabAt(tab, i)) == null)
        advance = casTabAt(tab, i, null, fwd);
    // 如果下标为i的桶已经设置了哨兵节点，那就溜了
    else if ((fh = f.hash) == MOVED)
        advance = true; // already processed
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

4）实际的迁移过程
重头戏开始了

else {
	// 锁住旧桶的头节点
    synchronized (f) {
    	// 再次校验
        if (tabAt(tab, i) == f) {
            Node<K,V> ln, hn;
            // 头节点的hash值大于0，表示是数据节点，还未开始迁移
            if (fh >= 0) {
            	// 这块机制比较复杂，不在注释里面讲了
                int runBit = fh & n;
                Node<K,V> lastRun = f;
                // lastRun 机制选出最后hash值相同的链表的头节点
                for (Node<K,V> p = f.next; p != null; p = p.next) {
                    int b = p.hash & n;
                    if (b != runBit) {
                        runBit = b;
                        lastRun = p;
                    }
                }
                // 根据最高位比特为1还是0，决定放入新数组的高位还是低位
                if (runBit == 0) {
                    ln = lastRun;
                    hn = null;
                }
                else {
                    hn = lastRun;
                    ln = null;
                }
                // 迁移剩余的节点，0入低位，1入高位
                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);
                }
                setTabAt(nextTab, i, ln);
                setTabAt(nextTab, i + n, hn);
                setTabAt(tab, i, fwd);
                advance = true;
            }
            // 如果是红黑树
            else if (f instanceof TreeBin) {
                // 处理逻辑与链表相同，不过多展示
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

前面讲过了，通过hash & (n - 1)来确定桶下标，迁移到新数组前，需要计算当前 hash 值在新数组的下标，而新数组的容量是就数组的两倍，也就是旧数组容量左移 1 位，新下标为hash & (m - 1)，其中 $m=2n$。举个例子，n = 8，掩码 = 0111，hash = 1011，那么旧桶的下标为 3（0011）。扩容后，m = 16，掩码 = 1111，那么新桶的下标为 11（1011），这个新桶的下标其实可以不用根据掩码计算，而是可以直接根据 hash 值来推算。

还是上面那个例子，扩容后，掩码多了一位，那么掩码能够截取的 hash 的位数也多了一位。扩容前取的是 011，扩容后取的是 1011，新的下标 = 0011 + 1000 = 1011。如果 hash = 0011，那么新的下标还是 0011 = 0011 + 0000，可以看出多取的 hash 值的比特位是 0 还是 1，就决定了新的下标会不会发生变化，如果发生了变化，变化的幅度为 $+n$。

int runBit = fh & n;这就是取 hash 值中多出来的那个比特，如果比特为 0，则 runBit = 0，如果比特为 1，则 runBit = n。后面的 for 循环使用了 lastRun 机制。旧桶中所有节点的 hash 的低位都是一样的，不然也不会映射到同一个桶中，唯一可能不同的就是多出来的比特的值，作者认为，多出来的那个比特也相同的节点在链表尾部大量存在，且都是连接在一起的，这样 lastRun 机制就能选出这个子链表的头节点，只要将头节点移动到新桶中，那么后面的节点也自动到了新桶中，避免将节点一个一个移动，提高了迁移效率。

接着再回头看get()方法中的find()方法，那就通过哨兵节点去新的数组中寻找 key。

Node<K,V> find(int h, Object k) {
    // loop to avoid arbitrarily deep recursion on forwarding nodes
    // 直接在新数组中查询
    outer: for (Node<K,V>[] tab = nextTable;;) {
        Node<K,V> e; int n;
        if (k == null || tab == null || (n = tab.length) == 0 ||
            (e = tabAt(tab, (n - 1) & h)) == null)
            return null;
        for (;;) {
            int eh; K ek;
            if ((eh = e.hash) == h &&
                ((ek = e.key) == k || (ek != null && k.equals(ek))))
                return e;
            // 可能新数组又发生扩容了，又要接着寻找，开始套娃
            if (eh < 0) {
                if (e instanceof ForwardingNode) {
                    tab = ((ForwardingNode<K,V>)e).nextTable;
                    continue outer;
                }
                else
                    return e.find(h, k);
            }
            if ((e = e.next) == null)
                return null;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

再看put()方法中的helpTansfer()方法，

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
        int rs = resizeStamp(tab.length);
        // 反复确认当前处于扩容阶段
        while (nextTab == nextTable && table == tab &&
               (sc = sizeCtl) < 0) {
            if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                sc == rs + MAX_RESIZERS || transferIndex <= 0)
                break;
            // 加入扩容工作
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

总结

ConcurrentHashMap 使用了大量的轻量的 CAS 操作来代替重量加锁操作，提高并发性能。我们看到只有在节点实际变动的过程才会加锁。

参考资料

• Java 并发工具包 java.util.concurrent 用户南
• Java Tutorial
• 并发容器之ConcurrentHashMap(JDK 1.8版本)
• Java7/8 中的 HashMap 和 ConcurrentHashMap 全解析
• Java并发包concurrent——ConcurrentHashMap
• 【JUC】JDK1.8源码分析之ConcurrentHashMap（一）
• 探索jdk8之ConcurrentHashMap 的实现机制