1. 云图网首页
  2. 技术专区
  3. 编程笔记

ConcurrentHashMap 原理和源码分析(一)详解编程语言

• 编程笔记

通过之前几篇文章《HashMap原理和源码分析》 《HashTable原理和源码分析》《LinkedHashMap原理和源码分析》的理解和分析，终于引出来了重头戏ConcurrentHashMap的分析。

说实话，之前几个数据结构HashMap、HashTable 复杂度跟ConcurrentHashMap相比简直是小儿科。

少说废话，直接引入正题。

其实有个疑问，既然有了HashMap和HashTable，为什么会需要ConcurrentHashMap？

因为呢，HashMap是线程不安全的，而HashTable虽然是线程安全的，方法都是用Synchronized修饰的，但争夺的都是同一个对象锁，在高并发的情况下，会产生效率低，等待时间长的问题。这个时候，ConcurrentHashMap就荣耀登场了，至于为什么ConcurrentHashMap能解决高并发的情况？下面会详细解释。

ConcurrentHashMap的特性和原理

JDK1.8 跟之前的版本，ConcurrentHashMap的实现变化了很大。以下都是基于JDK1.8的源码和资料。

  1. concurrentHashMap 不支持null的key和value

  2. concurrentHashMap 很多地方使用了cas操作和分段加锁，加锁的最小单位是Hash桶，这使得ConcurrentHashMap效率大大提升。

  3. 数据结构：

数据结构描述

ConcurrentHashMap的数据结构跟HashMap一样：数组 + 链表 + 红黑树。当hash桶的节点数量超过8个，链表就会转化为红黑树，反之，节点数量减少到6就会转化链表

源码分析

1. 重要的常量
// 初始化容量 
private static final int DEFAULT_CAPACITY = 16; 
 
// 加载因子，跟hashmap一样 
private static final float LOAD_FACTOR = 0.75f; 
 
// 如果发现链表长度小于8，会从链表转化成树，跟hashmap一样 
static final int TREEIFY_THRESHOLD = 8; 
 
// 在哈希表扩容时，如果发现链表长度小于 6，则会由树变成链表，跟hashmap一样 
static final int UNTREEIFY_THRESHOLD = 6; 
 
// 最低树化的容量，如果 容量 < MIN_TREEIFY_CAPACITY 会发生一次resize() 
static final int MIN_TREEIFY_CAPACITY = 64; 
 
// 迁移桶的最低数量 
// 表示扩容中，一个线程的一次任务负责迁移最少16个hash桶 
// 后面结合代码再理解下 
private static final int MIN_TRANSFER_STRIDE = 16; 
 
// 用于生成每次扩容都唯一的生成戳的数 
// 后面结合代码再理解下 
private static final int RESIZE_STAMP_BITS = 16; 
 
// 最大的扩容线程的数量 
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1; 
 
// 移位量 
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS; 
 
// 以下是标记几个特殊的节点的hash值，都是负数 
// ForwardingNode节点，表示该节点正在处于扩容工作，内部有个指针指向nextTable 
static final int MOVED     = -1; 
 
// 红黑树的首节点，内部不存key、value，只是用来表示红黑树 
static final int TREEBIN   = -2;  
 
// ReservationNode保留节点， 
// 当hash桶为空时，充当首结点占位符，用来加锁，在compute/computeIfAbsent使用 
static final int RESERVED  = -3;  
  
// 用于普通节点hash计算 
// 结合上面三个变量,特殊节点的hash值都是负数，普通节点为正数 
static final int HASH_BITS = 0x7fffffff; 
 
// CPU数量 
static final int NCPU = Runtime.getRuntime().availableProcessors();
2. 后面常用到的方法：
// 计算hash值 
// 让高16位 亦或 低16位，再把高的16位置为0 
 static final int spread(int h) {
    
        // & HASH_BITS用于把hash值转化为正数 
        return (h ^ (h >>> 16)) & HASH_BITS; 
 } 
 
// 跟hashmap扩容一样，计算出比c大，最小的2次幂的数，如14->16,29->32 
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; 
} 

// 根据table的长度n生成一个戳，表示要扩容n长度的table，会构造出sizeCtl 
// RESIZE_STAMP_BITS = 16 
static final int resizeStamp(int n) {
    
    return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1)); 
}

下面是CAS方法，CAS（compareAndSwap） 比较并交换，是无锁操作的重要手段，是一个原子操作

CAS(V, E, N)：如果变量V跟旧的预期值E相同，则修改成新值N，否则什么都不做

// 获取table[i] 得到hash桶首结点 
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); 
} 
 
// 更新table[i]，Node链表 或者TreeBin 
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); 
} 
 
// 修改table[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); 
}
3. 重要的变量
// 存储node节点的数组 
transient volatile Node<K,V>[] table; 
 
// 扩容后的新的table数组，只有在扩容时才有用 
// nextTable != null，说明在进行扩容 
private transient volatile Node<K,V>[] nextTable; 
 
// 在初始化或resize时控制参数，重要，后面细说 
private transient volatile int sizeCtl; 
 
// 扩容下个表的索引，重要，后面细说 
 private transient volatile int transferIndex; 
 
// Node节点的数量，根据cas更新的，有可能不准确，需要结合counterCells一起计算 
private transient volatile long baseCount; 
 
// 操作counterCells的自旋锁 
private transient volatile int cellsBusy; 
 
// counterCell表，大小是2的幂数， 
// 并行计算每个bucket的元素数量，结合baseCount算法出size，下面细说 
private transient volatile CounterCell[] counterCells;

以上是理解ConcurrentHashMap非常重要的几个变量，其中有几个没有细说

sizeCtl

这个控制参数贯穿了初始化或扩容，而且不同状态下表达不同的含义。

  1. sizeCtl == 0时候，默认情况
  2. sizeCtl == -1 时候，说明table正在初始化
  3. sizeCtl > 0 时候，说明接下来初始化要的初始化容量或者是扩容成功后threadshold的值
  4. sizeCtl < 0 时候，说明正在扩容，而此刻的sizeCtl是怎么来的呢？
// 假如有一个线程，准备加入扩容，下面开始计算sizeCtl	 
int rs = resizeStamp(table长度); 
if (sizeCtl 未初始化 或准备初始化) {
    // sizeCtl >= 0 
	sizeCtl =  (rs << RESIZE_STAMP_SHIFT) + 2 // RESIZE_STAMP_SHIFT = 16 
} else {
    // sizeCtl < 0 ，正在扩容 
	sizeCtl = rs + 1; //增加一个扩容的线程 
}

rs << 16 左移16位，这样低16位都是0；
rs << 16 + 2 应该理解成 rs << 16 + 1 + 1，第一个1表示初始状态，第二个1表示目前有一个线程参与扩容。
sizeCtl分成了高16位，做验证使用，防止扩容重叠；低16位表示 n – 1个线程在参与扩容线程数

transferIndex

这个变量跟扩容迁移有关，原来table扩容新的nextTable，需要多个线程参与节点的迁移。

transferIndex 从 table.length开始，表示需要迁移的桶的数量或者可以说是索引。

每个线程每次进来，如果发现正在扩容并且 transferIndex > 0 的时候，会停止手头的工作，加入帮助扩容，从中分配得到一个 [trasferIndex – stride, transferIndex）区间对应的hash桶的迁移工作，transferIndex 慢慢的减少直至为0。

也有可能是同一个线程负责了多次任务，迁移了多个stride数量hash桶。

下面只是一个示意图，并不表示一次任务只能移动四个hash桶，并且stride的数量是根据CPU数量和tab的数量决定的，最小是MIN_TRANSFER_STRIDE（16）。

示意图

baseCount, cellBusy, counterCells 计数统计

这三个变量跟统计节点数量有关，

  • baseCount用于记录节点的个数

  • cellsBusy是一个只有0和1两个状态的volatile整数，它被当做一个自旋锁，0代表无锁，1代表加锁，只要对counterCells操作，都需要先CAS更新cellsBusy加锁

  • counterCells 是一个辅助baseCount计数的数组，每个counterCell存着部分的节点数量，这样做的目的就是尽可能地减少冲突，看完下面的流程就可以明白了。

  • table节点的数量 = baseCount + counterCells每个cell记录下来的节点数量

总体的原则就是：先尝试更新baseCount，失败再利用CounterCell

  1. 通过CAS尝试更新baseCount ，如果更新成功则完成，如果CAS更新失败会进入下一步；

  2. 线程通过随机数ThreadLocalRandom.getProbe() & (n-1) 计算出在counterCells数组的位置，如果不为null，则CAS尝试在couterCell上直接增加数量，如果失败会进入下一步；

  3. counterCells数组会进行扩容为原来的两倍，继续随机，继续添加；

  4. 最后，table节点的数量 = baseCount + counterCells每个cell记录下来的节点数量

重要的内部类

// ConcurrentHashMap的节点 
static class Node<K,V> implements Map.Entry<K,V> {
    
        final int hash; 
        final K key; 
        volatile V val; 
        volatile Node<K,V> next; 
 
	// 比较节点是否相同：value跟key都相同 
 	public final boolean equals(Object o) {
    
            Object k, v, u; Map.Entry<?,?> e; 
            return ((o instanceof Map.Entry) && 
                    (k = (e = (Map.Entry<?,?>)o).getKey()) != null && 
                    (v = e.getValue()) != null && 
                    (k == key || k.equals(key)) && 
                    (v == (u = val) || v.equals(u))); 
    } 
} 

//红黑树的根节点 
 static final class TreeBin<K,V> extends Node<K,V> {
    
        TreeNode<K,V> root; 
        volatile TreeNode<K,V> first; 
        volatile Thread waiter; 
        volatile int lockState；// 当前的锁状态 
        static final int WRITER = 1; // 正在写 
        static final int WAITER = 2; // 等待写 
        static final int READER = 4; // 正在读 
 		...... 
 } 

// 红黑树的节点 
static final class TreeNode<K,V> extends Node<K,V> {
    
        TreeNode<K,V> parent;  // red-black tree links 
        TreeNode<K,V> left; 
        TreeNode<K,V> right; 
        TreeNode<K,V> prev;    // needed to unlink next upon deletion 
        boolean red; 
}
ForwardingNode

下面还有一个比较特别的节点

ForwardingNode 是临时节点，这个节点会出现在扩容的时候，不存储实际的数据数据。

如果Hash桶被迁移到新的table中，会在旧的table插入一个ForwardingNode临时节点，内部会指向新的table。

当读操作碰到ForwardingNode，会通过ForwardingNode内部的nextTable找到新的table，继续读。

当写操作碰到ForwadingNode，加入帮助扩容。

static final class ForwardingNode<K,V> extends Node<K,V> {
 
final Node<K,V>[] nextTable; 
ForwardingNode(Node<K,V>[] tab) {
 
super(MOVED, null, null, null); // hash设置为move，为-1 
this.nextTable = tab; 
} 
//重写了Node中的find方法 
Node<K,V> find(int h, Object k) {
 
// 避免多次碰到ForwardingNode导致递归过深 
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) {
// 还碰到ForwardingNode，往下递归接着找 
tab = ((ForwardingNode<K,V>)e).nextTable; 
continue outer; 
} 
else 
return e.find(h, k); 
} 
if ((e = e.next) == null) 
return null; 
} 
} 
} 
}

总结

本文将ConcurrentHashMap重要的数据结构和思想做了大致的介绍，ConcurrentHashMap效率高的原因主要是：

  1. 对Hash桶分段加锁
  2. 尽可能尝试CAS更新，否则才升级到Synchronized同步（put等方法）
  3. 增加多线程协助扩容，帮助迁移
  4. 增加counterCells帮助计数，减少冲突

参考

IT虾米网
IT虾米网

原创文章，作者：奋斗，如若转载，请注明出处：https://blog.ytso.com/19397.html

赞 (0)
0
上一篇 2021年7月19日
下一篇 2021年7月19日

相关推荐

发表回复

登录后才能评论