并发容器之 ConCurrentHashMap（二）

有的时候博客内容会有变动，首发博客是最新的，其他博客地址可能会未同步, 认准https://blog.zysicyj.top

可点击链接https://blog-1253652709.cos.ap-guangzhou.myqcloud.com//picgo/202401180921373.png 解答疑问

扩容

前面提到了，超过一定阈值后，segment 会进行扩容，代码如下。

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
54
/**
 * 扩大哈希表的大小并重新打包条目，同时将给定的节点添加到新表中
 */
@SuppressWarnings("unchecked")
private void rehash(HashEntry<K,V> node) {
    /*
     * 将每个链表中的节点重新分类到新表中。由于我们使用的是二次方扩展，
     * 因此每个桶中的元素要么保持在相同的索引位置，要么移动到一个二次方偏移量的位置。
     * 我们通过捕捉那些旧节点可以重用的情况来消除不必要的节点创建，因为它们的 next 字段不会改变。
     * 统计上，在默认阈值下，当表大小翻倍时，只有大约六分之一的节点需要克隆。
     * 当这些节点不再被任何正在遍历表的读取线程引用时，它们将是可垃圾回收的。
     * 条目的访问使用普通的数组索引，因为它们之后是易失的表写入。
     */
    HashEntry<K,V>[] oldTable = table; // 保存旧表的引用
    int oldCapacity = oldTable.length; // 旧表的容量
    int newCapacity = oldCapacity << 1; // 新表容量是旧表的两倍
    threshold = (int)(newCapacity * loadFactor); // 重新计算阈值
    HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity]; // 创建新表
    int sizeMask = newCapacity - 1; // 用于计算索引的掩码
    for (int i = 0; i < oldCapacity; i++) { // 遍历旧表中的每个桶
        HashEntry<K,V> e = oldTable[i]; // 获取桶中的第一个节点
        if (e != null) {
            HashEntry<K,V> next = e.next; // 保存下一个节点的引用
            int idx = e.hash & sizeMask; // 计算新表中的索引
            if (next == null) { // 如果该桶中只有一个节点
                newTable[idx] = e; // 直接将该节点放到新表中
            } else { // 如果该桶中有多个节点，重用连续的节点序列
                HashEntry<K,V> lastRun = e;
                int lastIdx = idx;
                for (HashEntry<K,V> last = next; last != null; last = last.next) {
                    int k = last.hash & sizeMask;
                    if (k != lastIdx) {
                        lastIdx = k;
                        lastRun = last;
                    }
                }
                newTable[lastIdx] = lastRun; // 将最后一个连续的节点放到新表中
                // 克隆剩余的节点
                for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
                    V v = p.value;
                    int h = p.hash;
                    int k = h & sizeMask;
                    HashEntry<K,V> n = newTable[k];
                    newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
                }
            }
        }
    }
    int nodeIndex = node.hash & sizeMask; // 计算新节点在新表中的索引
    node.setNext(newTable[nodeIndex]); // 设置新节点的下一个节点
    newTable[nodeIndex] = node; // 将新节点放到新表中
    table = newTable; // 更新表引用
}

关于该扩容函数，有几点需要说明：

  1. 函数的参数，也就是将要加入的最新节点。在扩容完成之后，把该节点加入新的 Hash 表。
  2. 整个数组的长度是 2 的整数次方，每次按二倍扩容，而 hash 函数就是对数组长度取模，即 node.hash&sizeMask。因此，如果元素之前处于第 i 个位置，当再次 hash 时，必然处于第 i 个或者第 i+oldCapacity 个位置。
  3. 上面的扩容进行了一次优化，并没有对元素依次拷贝，而是先找到 lastRun 位置，也就是 for 循环。lastRun 到链表末尾的所有元素，其 hash 值没有改变，所以不需要依次重新拷贝，只需把这部分链表链接到新链表所对应的位置就可以，也就是 newTable[lastIdx]=lastRun。lastRun 之前的元素则需要依次拷贝。
    因此，即使把一段 for 循环去掉，整个程序的逻辑仍然是正确的。

get 实现分析

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
/**
 * 返回指定键映射到的值，如果此映射不包含该键的映射，则返回 {@code null}。
 *
 * <p> 更正式地说，如果此映射包含一个键 {@code k} 到值 {@code v} 的映射，
 * 使得 {@code key.equals(k)}，则此方法返回 {@code v}；
 * 否则返回 {@code null}。 (最多只能有一个这样的映射。)
 *
 * @throws NullPointerException 如果指定的键为 null
 */
public V get(Object key) {
    Segment<K,V> s; // 手动整合访问方法以减少开销
    HashEntry<K,V>[] tab;
    int h = hash(key); // 计算键的哈希值
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; // 计算段的偏移地址
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && // 获取对应的段
        (tab = s.table) != null) { // 获取段中的哈希表
        // 获取哈希表中对应桶的头节点
        for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
                 (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
             e != null; e = e.next) { // 遍历桶中的链表
            K k;
            if ((k = e.key) == key || (e.hash == h && key.equals(k))) // 如果找到匹配的键
                return e.value; // 返回对应的值
        }
    }
    return null; // 如果没有找到匹配的键，返回 null
}

详细解释

  1. 哈希计算：int h = hash(key);

    • 计算键的哈希值，用于定位该键在哈希表中的位置。

  2. 段的偏移地址计算：long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;

    • 这行代码计算键所在的段在 segments 数组中的偏移量。
    • segmentShift 和 segmentMask 是 ConcurrentHashMap 中的常量，用于确定段的索引。
    • SSHIFT 和 SBASE 是用于计算段的地址偏移的常量。

  3. 获取段和哈希表：

    • s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)：通过计算出的偏移量，从 segments 数组中获取对应的段。
    • tab = s.table：获取段中的哈希表。

  4. 桶的头节点：

    • UNSAFE.getObjectVolatile(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE)：
      • 计算键在哈希表中对应桶的头节点位置。
      • TSHIFT 和 TBASE 是用于计算桶位置的常量。
    • for (HashEntry<K,V> e = ...; e != null; e = e.next)：遍历桶中的链表，寻找匹配的键。

  5. 键匹配：

    • if ((k = e.key) == key || (e.hash == h && key.equals(k)))：
      • 检查当前节点的键是否与目标键匹配。
      • 如果匹配，则返回对应的值。

  6. 返回结果：

    • 如果找到匹配的键，返回对应的值。
    • 如果没有找到，返回 null。

关键信息和注意点

  • 哈希计算和段的定位：代码中首先计算键的哈希值，然后通过一系列位运算确定键所在的段。这种方法保证了段的快速定位。
  • 并发控制：使用 UNSAFE.getObjectVolatile 方法来保证在多线程环境下的可见性和一致性。
  • 链表遍历：在哈希冲突时，通过链表遍历查找匹配的键。尽管在高并发环境下链表的长度一般不会很长，但链表遍历仍然是一个线性时间复杂度的操作。

总结

  • 这段代码展示了 ConcurrentHashMap 在高并发环境下的高效键查找过程。
  • 通过哈希值计算和段的定位，代码能够快速确定键的位置。
  • 使用 UNSAFE 类方法确保多线程环境下的数据一致性。
  • 总体来说，这段代码体现了 ConcurrentHashMap 在设计 docker 上的巧妙性和高效性，适用于高并发读操作的场景。

整个 get 过程也就是两次 hash：

  • 第一次 hash，函数为（h＞＞＞segmentShift）&segmentMask，计算出所在的 Segment；
  • 第二次 hash，函数为 h&（tab.length-1），即 h 对数组长度取模，找到 Segment 里面对应的 HashEntry 数组下标，然后遍历该位置的链表。
    整个读的过程完全没有加锁，而是使用了 UNSAFE.getObjectVolatile 函数。

下篇我们继续讲解 JDK8 中是怎么实现的