跳至主要內容

线程安全集合类

高松灯大约 7 分钟

分类

  1. 遗留的安全集合(使用synchornized关键字进行修饰，并发性能比较低，fail-fast机制，有人修改就失败)
    1. HashTable
    2. Vector
  2. 修饰的安全集合(底层用synchornized(mutex)关键字修饰，然后调用HashMap的方法，并发性能也很低)
    1. SynchornizedMap
    2. SynchornizedList
  3. JUC安全集合
    1. Blocking类(基于锁)
    2. CopyOnWrite类(基于拷贝，适用于读多写少的场景)
    3. Concurrent类(基于CAS，缺点:弱一致性，fail-safe机制，有人修改不会失败，允许数据不一致)

HashMap的并发安全问题

在并发扩容过程中出现死链问题，或者扩容丢失数据

ConcurrentHashMap如何解决并发安全问题

使用volatile修饰的ForwardingNode节点进行占位，保证可见性；

使用cas操作进行Node的写操作

JDK 1.8 的 ConcurrentHashMap 是 Java 并发编程的集大成者，其内部结构相比 1.7 发生了翻天覆地的变化。它抛弃了 Segment 分段锁，采用了 CAS + synchronized + Node 数组 + 链表/红黑树 的结构。

以下是核心属性和关键方法的深度解析：

1. 关键属性 (Fields)

这些属性都使用了 volatile 修饰，保证内存可见性。

  1. table (Node<K,V>[] volatile)

    • 含义：核心数据存储数组（哈希桶）。
    • 作用：默认为 null，第一次 put 时才初始化（懒加载）。
    • 特点：每个位置存放链表头节点或红黑树根节点。

  2. nextTable (Node<K,V>[] volatile)

    • 含义：扩容时的临时新数组。
    • 作用：只在扩容（Resizing）期间不为 null。当扩容结束后，它会被置为 null。
    • 关键：通过判断 nextTable 是否为 null，线程可以感知当前是否正在扩容。

  3. sizeCtl (int volatile) —— 最核心的状态控制变量

    • 这是一个多义词变量，不同状态下含义不同：
    • -1：表示正在进行初始化。
    • -N：表示正在进行扩容。具体数值代表有多少个线程正在参与扩容搬迁工作。
    • 0：默认值，表示尚未初始化。
    • > 0：
      • 如果未初始化：表示初始容量。
      • 如果已初始化：表示扩容阈值（capacity * 0.75）。

  4. ForwardingNode (特殊的 Node)

    • 含义：转发节点。
    • 特征：它的 hash 值固定为 MOVED (-1)。
    • 作用：扩容时的占位符。当一个桶（Bucket）里的数据被迁移到了新数组 nextTable 后，旧数组的这个位置就会放一个 ForwardingNode。
    • 意义：如果有其他线程来读旧数组，发现这个节点，就知道“这里的数据搬家了”，它会跳转到 nextTable 去查找。

2. 关键内部类 (Inner Classes)

  1. Node：普通的链表节点。val 和 next 是 volatile 的。
  2. TreeNode：红黑树节点。
  3. TreeBin：红黑树的代理节点。当链表转红黑树时，桶里放的是 TreeBin，它内部持有红黑树的引用。

3. 关键方法 (Methods) 解析

A. put(K key, V value) —— 写入操作

核心流程是一个死循环（自旋），直到插入成功为止：

  1. 计算 Hash：计算 key 的 hash 值（高低位异或）。

  2. 初始化：如果 table 为空，先调用 initTable() 初始化（CAS 设置 sizeCtl 为 -1 抢占初始化权）。

  3. 定位桶：计算下标 i。

  4. Case 1：桶是空的 (null)

    • 动作：使用 CAS (casTabAt) 尝试把新节点放入该位置。
    • 结果：成功则退出；失败（被别人抢先了）则进入下一轮循环。
    • 锁：完全无锁，基于 CAS。

  5. Case 2：桶里是 ForwardingNode (hash == MOVED)

    • 动作：说明正在扩容，当前位置的数据已经搬走了。
    • 结果：调用 helpTransfer()，当前线程加入扩容大军，帮着一起搬运数据（多线程协同扩容）。

  6. Case 3：桶里有数据 (Hash 冲突)

    • 动作：使用 synchronized 锁住当前桶的头节点。
    • 逻辑：
      • 如果是链表：遍历链表，覆盖旧值或追加到尾部。
      • 如果是红黑树：调用树的插入逻辑。
    • 锁：只锁当前这一个桶。

  7. 树化检查：插入后，如果链表长度 > 8，尝试转换为红黑树。

  8. 计数：调用 addCount() 增加元素个数（LongAdder 思想，高并发下分散热点），并检查是否需要扩容。

B. get(Object key) —— 读取操作

全程无锁，性能极高。

  1. 计算 Hash，定位下标。
  2. 检查头节点：
    • 如果头节点就是目标，直接返回。
    • 如果头节点的 hash < 0（说明是 ForwardingNode 或 TreeBin）：调用多态方法 find()。
      • 如果是 ForwardingNode，它会把你引导到 nextTable 去找。
  3. 遍历链表查找。

C. transfer() —— 扩容与数据迁移

这是最复杂的方法。

  • 分片迁移：把数组切分成多个段（stride），每个线程领一段任务去搬运。
  • 状态标记：搬完一个桶，就在旧位置放一个 ForwardingNode。
  • 多线程协作：多个线程可以通过 transferIndex 全局变量来领取不同的下标区间，并发进行迁移，大大加快了扩容速度。

D.构造方法

JDK 1.8 中 ConcurrentHashMap 的构造方法有一个非常重要的特点：懒加载（Lazy Loading）。

这意味着，当你在 new ConcurrentHashMap() 时，它并不会立即初始化内部的数组（table），仅仅是计算并保存了一些参数（如 sizeCtl）。真正的初始化被推迟到了第一次调用 put() 方法时。

让我们逐个分析主要的构造方法：

1. 无参构造方法
public ConcurrentHashMap() {
}
  • 分析：什么都不做。
  • 状态：table 为 null，sizeCtl 为 0（默认值）。
  • 目的：极速创建对象，如果不使用就不占用内存。
2. 指定初始容量
public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    
    // 核心逻辑：计算 sizeCtl
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
               MAXIMUM_CAPACITY :
               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    
    this.sizeCtl = cap;
}
  • 核心逻辑：
    • 这里并没有创建数组！
    • 它计算了一个 cap，这个 cap 是大于等于 1.5 倍 initialCapacity 的最小的 2 的幂次方。
    • 为什么是 1.5 倍？
      • HashMap 的默认负载因子是 0.75。
      • 如果你想要存 100 个元素，为了不触发扩容，你的容量至少应该是 100 / 0.75 ≈ 133。
      • 这里的算法直接帮你把容量预估大了，避免你刚存满就触发扩容。
    • tableSizeFor：这个方法保证计算出的容量永远是 2 的 n 次幂（例如 16, 32, 64）。
    • 赋值：将计算好的容量赋值给 sizeCtl。注意：此时 sizeCtl 存储的是【初始容量】，而不是扩容阈值。
3. 指定初始容量和负载因子
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
    this(initialCapacity, loadFactor, 1);
}
  • 调用了下面那个最全的构造器，并发度（concurrencyLevel）默认为 1。
4. 全参数构造方法
public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    
    // 1. 保证初始容量至少能容纳 concurrencyLevel 这么多元素
    if (initialCapacity < concurrencyLevel)   
        initialCapacity = concurrencyLevel;   
    
    // 2. 根据负载因子计算实际需要的数组大小
    long size = (long)(1.0 + (long)initialCapacity / loadFactor);
    
    // 3. 规范化为 2 的幂次方
    int cap = (size >= (long)MAXIMUM_CAPACITY) ?
        MAXIMUM_CAPACITY : tableSizeFor((int)size);
    
    // 4. 保存到 sizeCtl
    this.sizeCtl = cap;
}
  • concurrencyLevel 参数：在 JDK 1.8 中，这个参数实际上已经没什么用了。
    • 在 JDK 1.7 中它决定了 Segment 的个数。
    • 在 JDK 1.8 中，它仅仅用于辅助计算初始容量，保证初始容量不小于并发度，除此之外不再影响内部结构。这是为了兼容旧版本的 API。
总结
  1. 懒惰：构造方法里绝对不分配数组内存（不 new Node[]）。
  2. sizeCtl 的临时身份：在初始化前，sizeCtl 变量暂时充当了“配置项寄存器”，存储的是经过计算后的初始数组大小。
  3. 什么时候初始化？
    • 当你第一次调用 put 方法时。
    • 会进入 initTable() 方法。
    • 该方法会读取 sizeCtl 的值作为数组长度来创建数组。
    • 创建完后，立刻把 sizeCtl 更新为 capacity * 0.75（变成扩容阈值）。

总结 JDK 1.8 的改进

  1. 数据结构：引入红黑树，解决 Hash 冲突严重时链表退化为 O(n) 的问题，优化为 O(log n)。
  2. 锁粒度：从 Segment 细化到 Node（桶头），并发度大幅提升。
  3. 锁实现：大量使用 CAS（无锁算法）和 synchronized（JDK 1.8 对 synchronized 做了大量优化，如偏向锁、轻量级锁，性能不输 ReentrantLock，且节省内存）。
  4. 扩容：引入多线程协同扩容机制。