源码分析——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实现过程：

1. 根据key，计算出hashCode；
2. 根据步骤1计算出的hashCode定位segment，如果segment不为null && segment.table也不为null，跳转到步骤3，否则，返回null，该key所对应的value不存在；
3. 根据hashCode定位table中对应的hashEntry，遍历hashEntry，如果key存在，返回key对应的value；
4. 步骤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实现过程：

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

segment的put方法实现

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

segment put实现过程：

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

• 第一：HashEntry数组扩容；

是否需要扩容

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

如何扩容

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

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

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

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 bits
    long sum;         // sum of modCounts
    long last = 0L;   // previous sum
    int retries = -1; // first iteration isn't retry
    try {
        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;
}复制代码

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中主要做了如下优化:

1. 将Segment抛弃掉了，直接采用Node（继承自Map.Entry）作为table元素。
2. 修改时，不再采用ReentrantLock加锁，直接用内置synchronized加锁，Java8的内置锁比之前版本优化了很多，相较ReentrantLock，性能不并差。
3. 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 默认初始容量是 16
                    int 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 * n
                    sc = 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为空，初始化table
            tab = 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相同，则修改value
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                //仅putIfAbsent()方法中onlyIfAbsent为true
                                if (!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一样也是8
                if (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判断，防止被其它线程修改。

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;
                        else
                            tl.next = p;
                        tl = p;
                    }
                    // 将红黑树设置到数组相应位置中
                    setTabAt(tab, index, new TreeBin(hd));
                }
            }
        }
    }
}复制代码

// 首先要说明的是，方法参数 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 不为 null
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            // 1. 将 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2)
            // 调用 transfer 方法，此时 nextTab 参数为 null
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
        }
    }
}复制代码

至于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的实现之精妙，个人觉得还需以后再多看几次，相信每次都会有新的收获。