探索 ConcurrentHashMap 高并发性的实现机制

核心提示： 图 2. 插入三个节点后 Segment 的结构示意图： ConcurrentHashMap 类ConcurrentHashMap 在默认并发级别会创建包含 16 个 Segment 对象的数组，每个 Segment 的成员对象 table 包含若干个散列表的桶，探索 ConcurrentHashMap 高并发性

图 2. 插入三个节点后 Segment 的结构示意图：  

ConcurrentHashMap 类

ConcurrentHashMap 在默认并发级别会创建包含 16 个 Segment 对象的数组。每个 Segment 的成员对象 table 包含若干个散列表的桶。每个桶是由 HashEntry 链接起来的一个链表。如果键能均匀散列，每个 Segment 大约守护整个散列表中桶总数的 1/16。

清单 3.ConcurrentHashMap 类的定义  

				 
 public class ConcurrentHashMap extends AbstractMap 
        implements ConcurrentMap, Serializable { 

    /** 
     * 散列映射表的默认初始容量为 16，即初始默认为 16 个桶
     * 在构造函数中没有指定这个参数时，使用本参数
     */ 
    static final 	 int DEFAULT_INITIAL_CAPACITY= 16; 

    /** 
     * 散列映射表的默认装载因子为 0.75，该值是 table 中包含的 HashEntry 元素的个数与
 * table 数组长度的比值
     * 当 table 中包含的 HashEntry 元素的个数超过了 table 数组的长度与装载因子的乘积时，
 * 将触发 再散列
     * 在构造函数中没有指定这个参数时，使用本参数
     */ 
    static final float DEFAULT_LOAD_FACTOR= 0.75f; 

    /** 
     * 散列表的默认并发级别为 16。该值表示当前更新线程的估计数
     * 在构造函数中没有指定这个参数时，使用本参数
     */ 
    static final int DEFAULT_CONCURRENCY_LEVEL= 16; 

    /** 
     * segments 的掩码值
     * key 的散列码的高位用来选择具体的 segment 
     */ 
    final int segmentMask; 

    /** 
     * 偏移量
     */ 
    final int segmentShift; 

    /** 
     * 由 Segment 对象组成的数组
     */ 
    final Segment[] segments; 

    /** 
     * 创建一个带有指定初始容量、加载因子和并发级别的新的空映射。
     */ 
    public ConcurrentHashMap(int initialCapacity, 
                             float loadFactor, int concurrencyLevel) { 
        if(!(loadFactor > 0) || initialCapacity < 0 || 
 concurrencyLevel <= 0) 
            throw new IllegalArgumentException(); 

        if(concurrencyLevel > MAX_SEGMENTS) 
            concurrencyLevel = MAX_SEGMENTS; 

        // 寻找最佳匹配参数（不小于给定参数的最接近的 2 次幂） 
        int sshift = 0; 
        int ssize = 1; 
        while(ssize < concurrencyLevel) { 
            ++sshift; 
            ssize <<= 1; 
        } 
        segmentShift = 32 - sshift;       // 偏移量值
        segmentMask = ssize - 1;           // 掩码值 
        this.segments = Segment.newArray(ssize);   // 创建数组

        if (initialCapacity > MAXIMUM_CAPACITY) 
            initialCapacity = MAXIMUM_CAPACITY; 
        int c = initialCapacity / ssize; 
        if(c * ssize < initialCapacity) 
            ++c; 
        int cap = 1; 
        while(cap < c) 
            cap <<= 1; 

        // 依次遍历每个数组元素
        for(int i = 0; i < this.segments.length; ++i) 
            // 初始化每个数组元素引用的 Segment 对象
 this.segments[i] = new Segment(cap, loadFactor); 
    } 

    /** 
     * 创建一个带有默认初始容量 (16)、默认加载因子 (0.75) 和 默认并发级别 (16) 
  * 的空散列映射表。
     */ 
    public ConcurrentHashMap() { 
        // 使用三个默认参数，调用上面重载的构造函数来创建空散列映射表
 this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); 
 } 

}

下面是 ConcurrentHashMap 的结构示意图。

图 3.ConcurrentHashMap 的结构示意图：

用分离锁实现多个线程间的并发写操作

在 ConcurrentHashMap 中，线程对映射表做读操作时，一般情况下不需要加锁就可以完成，对容器做结构性修改的操作才需要加锁。下面以 put 操作为例说明对 ConcurrentHashMap 做结构性修改的过程。

首先，根据 key 计算出对应的 hash 值：

清单 4.Put 方法的实现  

				 
 public V put(K key, V value) { 
        if (value == null)          //ConcurrentHashMap 中不允许用 null 作为映射值
            throw new NullPointerException(); 
        int hash = hash(key.hashCode());        // 计算键对应的散列码
        // 根据散列码找到对应的 Segment 
        return segmentFor(hash).put(key, hash, value, false); 
 } 

然后，根据 hash 值找到对应的Segment 对象：

清单 5.根据 hash 值找到对应的 Segment  

				 
 /** 
     * 使用 key 的散列码来得到 segments 数组中对应的 Segment 
     */ 
 final Segment segmentFor(int hash) { 
    // 将散列值右移 segmentShift 个位，并在高位填充 0 
    // 然后把得到的值与 segmentMask 相“与”
 // 从而得到 hash 值对应的 segments 数组的下标值
 // 最后根据下标值返回散列码对应的 Segment 对象
        return segments[(hash >>> segmentShift) & segmentMask]; 
 } 

最后，在这个 Segment 中执行具体的 put 操作：

清单 6.在 Segment 中执行具体的 put 操作  

				 
 V put(K key, int hash, V value, boolean onlyIfAbsent) { 
            lock();  // 加锁，这里是锁定某个 Segment 对象而非整个 ConcurrentHashMap 
            try { 
                int c = count; 

                if (c++ > threshold)     // 如果超过再散列的阈值
                    rehash();              // 执行再散列，table 数组的长度将扩充一倍

                HashEntry[] tab = table; 
                // 把散列码值与 table 数组的长度减 1 的值相“与”
                // 得到该散列码对应的 table 数组的下标值
                int index = hash & (tab.length - 1); 
                // 找到散列码对应的具体的那个桶
                HashEntry first = tab[index]; 

                HashEntry e = first; 
                while (e != null && (e.hash != hash || !key.equals(e.key))) 
                    e = e.next; 

                V oldValue; 
                if (e != null) {            // 如果键 / 值对以经存在
                    oldValue = e.value; 
                    if (!onlyIfAbsent) 
                        e.value = value;    // 设置 value 值
                } 
                else {                        // 键 / 值对不存在 
                    oldValue = null; 
                    ++modCount;         // 要添加新节点到链表中，所以 modCont 要加 1  
                    // 创建新节点，并添加到链表的头部 
                    tab[index] = new HashEntry(key, hash, first, value); 
                    count = c;               // 写 count 变量
                } 
                return oldValue; 
            } finally { 
                unlock();                     // 解锁
            } 
        } 

注意：这里的加锁操作是针对（键的 hash 值对应的）某个具体的 Segment，锁定的是该 Segment 而不是整个 ConcurrentHashMap。因为插入键 / 值对操作只是在这个 Segment 包含的某个桶中完成，不需要锁定整个ConcurrentHashMap。此时，其他写线程对另外 15 个Segment 的加锁并不会因为当前线程对这个 Segment 的加锁而阻塞。同时，所有读线程几乎不会因本线程的加锁而阻塞（除非读线程刚好读到这个 Segment 中某个 HashEntry 的 value 域的值为 null，此时需要加锁后重新读取该值）。