[toc]
前言
分析ConturrentHashMap 1.8的实现，JDK1.8实现屏蔽了Segment(分段代码锁)的概念，而是直接用Node数组+链表+红黑树的数据结构实现的，并发机制使用synchronized和CAS来操作，整个看起来就像是优化后且线程安全的HashMap，虽然JDK1.8还能看见Segment，但是已经简化了属性，只是为了兼容旧版本。
注：ConcurrentHashMap和HashTable都是支持并发的，但是不允许key/value为null，因为当你通过get(key)获取value时，如果获取到的是null，你无法判断他是put(key,v)的时候value为null，还是这个key从来没用做过映射，HashMap是非并发的，可以通过contains(key)判断，而支持并发的Map在调用contains(key)和get(key)，map可能已经不同了，所以不允许key/value为null
结构及核心组成
在深入JDK1.8的put和get实现前要知道一些设计和数据结构，这些都是构成ConcurrentHashMap实现的基础，下面看一下：
常量
//node数组的最大容量2^30次方
private static final int MAXIMUN_CAPACITY = 1 << 30;
//默认数组容量的初始值，必须是2的n次幂
private static final int DEFAULT_CAPACITY = 16;
//数组可能的最大值，需要与toArray()方法相关联
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
//默认并发级别，遗留下来，未使用，为兼容以前版本
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
//负载因子
private static final float LOAD_FACTOR = 0.75f;
//链表转为红黑树的阈值>8时候链表转为红黑树
static final int TREEIFY_THRESHOLD = 8;
//红黑树转为链表的阈值，小于等于6
static final int UNTREEIFY_THRESHOLD = 6;
//最小树化容量，当数组长度小于64时候不会将链表转为红黑树而是先扩容
static final int MIN_TREEIFY_CAPACITY = 64;
//转移最小值
private static final int MIN_TRANSFER_STRIDE = 16;
//生成sizeCtl的最小位数
private static int RESIZE_STAMP_BITS = 16;
//调整大小时候最大允许的线程数
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
//sizeCtl中记录size大小的偏移量
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
//forwarding nodes的hash值
static final int MOVED = -1; // hash for forwarding nodes
//treeBin节点的固定hash值
static final int TREEBIN = -2; // hash for roots of trees
//临时保留的哈希
static final int RESERVED = -3; // hash for transient reservations
//普通节点哈希的可用位,用于hash取正,使其成为正常节点
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
//可用处理器的数量
static final int NCPU = Runtime.getRuntime().availableProcessors();
成员变量
//第一次插入时候进行初始化必须是2^n次幂
transient volatile Node<K,V>[] table;
//用于扩容，下一次调整大小时为非null
private transient volatile Node<K,V>[] nextTable;
//基本计算器值，通过CAS修改值，没用竞争是时使用，或者出现多线程初始化时回滚
private transient volatile long baseCount;
//初始化和扩容的标志，concurrent包中很多类似用法
//-1 初始化中 -N (N-1)个线程在扩容。
//非常重要的一个属性，源码中英文翻译，直译过来下面四行文字意思：
//sizeCtl = -1 表示有线程正在进行真正的初始化操作
//sizeCtl = -(1-nThreads),表示有nThreads个线程正在进行扩容操作
// sizeCtl>0，表示接下来真正的初始化操作中使用的容量，或者初始化/扩容完成后的threshold
// sizeCtl = 0,默认值，此时在真正的初始化操作中使用默认容量
private transient volatile int sizeCtl;
//transfer的table索引
private transient volatile int transferIndex;
//扩容或创建counterCells的自旋锁,通过CAS锁定
private transient volaile int cellsBusy;
//计数器原件。如果为非null，则大小为2的幂。大小为2^n
private transient volatile CounterCell[] counterCells;
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;
重点解释一下sizeCtl这个属性，可以说是ConcurrentHashMap中出镜率最高的属性，因为他是一个控制标识符，在不同的地方有不同的用途，而且他的取值不通，也代表不同的含义
- 负数代表正在初始化或扩容操作
- -1代表正在有线程进行初始化
- -N表示用N-1个线程正在进行初始化
- 0代表hash表还没有被初始化
- 正数表示初始化或下一次进行扩容的大小，这一点类似于扩容阈值的概念，后面可以看到,他的值使用是当前ConcurrentHashMap容量的0.75倍，这与loadfactor对应
Node<K,V>[] table是整个hash表的结构，也被称为hash桶数组，他是Node<K,V>的数组，因此1.8的ConccurrentHashMap可以看做如下结构：
image
table在第一次插入时候被初始化，默认大小是16，长度一定是2的n次方
四种Node结构
//以下是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;
Node(int hash,K key,V val,Node<K,V> next){
this.hash = hash;
this.key = key;
this.val = val;
this.next = next;
}
public final K getKey(){
return key;
}
public final V getValue(){
return val;
}
public final int hashCode(){
return key.hashCode() ^ val.hashCode();
}
public final String toString(){
return key + "=" + val;
}
//不允许set值
public final V setValue(){
throw new UnsupportedOperationException();
}
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)));
}
//对map.get（）的虚拟支持；在子类中重写。
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))))
returnn e;
}while((e = e.next) != null);
}
return null;
}
}
//在TreeBins中使用的节点
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;
//prev指针是为了删除方便，删除链表的非头节点的节点，都需要知道它的前一个节点才能进行删除，所以直接提供一个prev指针
//就是链表形式时候的节点的前一个节点，如果是头结点就为null
TreeNode<K,V> prev; // needed to unlink next upon deletion
boolean red;
TreeNode(int hash, K key, V val, Node<K,V> next,
TreeNode<K,V> parent) {
super(hash, key, val, next);
this.parent = parent;
}
Node<K,V> find(int h, Object k) {
return findTreeNode(h, k, null);
}
//从根节点开始查找，返回给定key的TreeNode，如果没有返回null
final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
.....
}
}
//树的包装类Node
static final class TreeBin<K,V> extends Node<K,V> {
//红黑树根节点
TreeNode<K,V> root;
//链表的头结点，TreeBin仍然保存了链表结构
volatile TreeNode<K,V> first;
//最近设置WAITER标志位的线程
volatile Thread waiter;
//锁状态标志位
volatile int lockState;
// 锁状态可取的值，状态可以多选，读状态可以叠加
static final int WRITER = 1; // 写锁标志位
static final int WAITER = 2; // 等待锁标志位
static final int READER = 4; //读锁标志位
}
//只在扩容时候出现，实现了扩容时旧表和新表的链接
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);
this.nextTable = tab;
}
//查找节点
Node<K,V> find(int h, Object k) {
.....
}
}
-
Node<K,V>：继承了Map.Entry<K,V>是其余三种Node的基类。ConcurrentHashMap中链表的普通节点就是Node<K,V>，整个表结构就是Node<K,V>的数组，Node<K,V>与HashMap定义相似，他对value和next的设置都是volatile，保证了其他线程的可见性。
-
TreeNode<K,V>继承了Node<K,V>与1.8的hashMap一样，当链表长度超过8时，并且table的length的长度大于64(小于64先扩容时)时，链表转为红黑树。TreeNode<K,V>就是红黑树节点由于继承了Node<K,V>，红黑树节点本身保存着普通链表节点Node的所有属性，因此可以使用两种方式进行读操作。 -
TreeBin<K,V>：继承了Node<K,V>，hash固定为-2，是红黑树的包装节点，与hashMap不同的是，ConcurrentHashMap数组中放入的不是TreeNode节点，而是将TreeNode包装起来的TreeBin的对象中，此外可以从TreeBin包含的字段中处理包含树的根节点root，还有读写锁方面的状态变量。
-
ForwardingNode<K,V>：继承了Node<K,V>，hash固定是-1，只在扩容tranfer的时候吃醋先，当数组槽位空或已经完成数组槽的扩容，将该节点插入数组槽中告知其他线程，如果就数组的一个hash桶中全部节点都迁移到新数组中，就数组就在这个hash桶中放置一个ForwardingNode，读操作或者迭代时碰到ForwardingNode时，将操作转发到扩容新的table数组上去执行，如果写操作碰见ForwardingNode时，则尝试帮助扩容
构造方法
//创建一个新的map，默认大小为16
public ConcurrentHashMap() {
}
//创建一个新的map，其初始表大小可以容纳指定数量的元素，而不用动态调整大小
public ConcurrentHashMap(int initialCapacity) {
//初始化大小小于0则抛出异常
if (initialCapacity < 0)
throw new IllegalArgumentException();
//调整初始化大小必须为2^n次幂，而且最大是2^30次方
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
//调整大小
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
//设置sizeCtl等于初始化容量
this.sizeCtl = cap;
}
//创建一个与给定map具有相同映射的新map
public ConcurrentHashMap(Map<? extends K, ? extends V> m){
//sizeCtl为默认的DDFAULT_CAPACITY=16
this,sizeCtl = DEFAULT_CAPACITY;
//调用putAll方法
putAll(m);
}
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1);
}
//concurrencyLevel主要为了兼容1.7及之前版本,它并不是实际的并发等级
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
//负载因子小于0或初始化容量小于0或并发级别小于等于0直接抛异常
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
//并发等级大于初始化容量，将并发等级赋值给初始化容量
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
//调整容量大小
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
//设置sizeCtl的值
this.sizeCtl = cap;
}
其实和HashMap的构造方法大同小异，此时ConcurrentHashMap的构造方法逻辑和HashMap基本一致，只是多一个concurrenyLevel和sizeCtl,而且也是此此时没有初始化table，要等等到第一次put时候才初始化(HashMap，也是一样的)。
成员方法
initTable
initTable：在put方法中，首先判断存放数据的table是否为null，如果为null，这时候需要初始化table就要调用这个方法。
private final Node<K,V> initTable(){
Node<K,V> tab;int sc;
//使用自旋一直判断sizeCtl的值如果小于0说明有线程正在进行初始化或者扩容操作
while((tab = table) == null || tab.length == 0){
//如果sizeCtl当前小于0说明有线程正在进行初始化或者扩容操作，一直自旋判断
if ((sc = sizeCtl) < 0)
//让出cpu时间碎片，重新自旋竞争
Thread.yield();
//通过CAS的形式更改sizeCtl的值，将值改为-1，如果设置成功，则当前线程进入初始化
else if(U.compareAndSwapInt(this,SIZECTL，sc,-1)){
// 如果原子更新失败则说明有其它线程先一步进入初始化了，则进入下一次循环
// 如果下一次循环时还没初始化完毕，则sizeCtl<0进入上面if的逻辑让出CPU
// 如果下一次循环更新完毕了，则table.length!=0，退出循环
try{
//为什么还要判断，因为：如果走到下面的finally改变了sizeCtl值，有可能其他线程是会进入这个逻辑的
if((tab = table) == null || table.length == 0){
//sc如果大于0说明设置了初始化容量，如果没有设置初始化容量默认是16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//设置下一次扩容的阈值，因为如果这个值是正数，sizeCtl的值是下一次扩容的值
//n - (n >>>2) 这个值就是 初始化的容量*0.75 (和用户设置的负载因子无关，就是默认负载因子0.75)
sc = n - (n >>>2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
//上面代出现了CAS操作(compareAndSwapInt).ConcurrentHashMap的源码中除了普通的CAS操作,
//还提供了更强的数组元素volatile读写以及CAS更新.代码如下:
//volatile读取table[]
@SuppressWarnings("unchecked")
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更新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);
}
//volatile设置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);
}
在这个过程中，就已经有乐观锁的实现，可以看出table的初始化在一次CAS方法中进行，当table为null或者长度为0时进入while内。while内的执行过程：
- 首先判断sizeCtl的值，如果小于0则线程让步，自旋等待，由于初始化状态sizeCtl是等于0的，如果是小于0，说明前面有线程进入了
else if这部分，将sizeCtl设置为-1.表示有一个线程正在初始化。 - 如果sc不小于0进入
else if中，通过CAS更新sizeCtl的值更新为-1，表示有线程正在进行else if操作，如果更新成功则进入下一步，否则继续循环判断。 - 进入
else if如果sc大于0，则取sc，否则是取默认容量16，然后计算下一次元素数量达到多少时需要resize (n-(n>>>2),容量✖️0.75，与用户设置负载因子无关，就是默认的0.75，与用户设置的负载因子无关)
总结初始化方法如下：
- 如果sizeCtl小于0，说明数组正在初始化，让出执行权
- 如果sizeCtl大于0，则初始化一个大小为sizeCtl的数组
- 如果sizeCtl等于0，初始化一个默认大小为16的数组
- 最后设置sizeCtl的值数组长度的3/4(容量✖️0.75,与负载因子无关)
get方法
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//计算hash
int h = spread(key.hashCode());
//判断table!=null并且length>0,通过k的hash获取桶中的链表头结点或者treeBin节点
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//判断如果e的hash等于key的hash
if ((eh = e.hash) == h) {
//判断e的key是否等于key
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
//如果全相等说明头结点就是对应的节点直接返回value
return e.val;
}
//如果eh小于0说明不是正常链表的节点,有可能的情况:
//-1是ForwardingNode(旧数组的一个hash桶中全部节点都迁移到新数组中,旧数组就在这个hash桶中放置一个ForwardinngNode)则用find函数转发到nextTable上查找
//-2是TreeBin，调用TreeBin的find函数。根据自身读写锁情况，判断是用红黑树方式查找，还是用链表方式查找
//transient reservations 固定是 -3
else if (eh < 0)
//调用对应类型节点中的find方法找出对应的value如果没有返回null
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;
}
get操作总结：
首先定位到具体的Hash槽，若hash槽不为空，判断第一个节点是否是要查找的节点(判断方法是比较hash值，若相等，则需要比较地址相等或者equals为true中的一个成立，则是要从查找的节点)，否则根据hash值是否是负数，将查找操作分为相应的find函数。
- 若是
FowaardingNode则用find函数转发到nextTable上查找。 - 若是TreeBin节点，调用TreeBing的函数，根据自身读写锁情况，去查找红黑树。
- 最后是普通节点，则遍历链表寻找
从代码中看出get是无锁的，即使TreeBing的find函数虽然有可能会加TreeBin的内部读锁。但也是非阻塞的。
定位方法和hashMap基本一致，首先求出hash值，然后利用hash&(n-1)来快速定位，但有一点不同是ConcurrentHashMap用了spread函数来求hash值，与hashMap略有不同代码如下：
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
除了高16位和低16位操作之外，最后还和HASH_BITS在上文中提到过值为0x7fffffff。他的作用就是让hash值为正数，在CouncurrentHashMap中hash值负数有特别的含义，如-1表示Forwarding节点，-2表示TreeBinn节点，而hashMap不用。
当hash数组中第一个节点是负数时，会根据节点的类型利用多态调用相应的find函数。
比如是-1是ForwardingNode节点，则调用ForwardingNode的find函数，代码如下：
//ForwardingNode节点
static final class ForwardingNode<K,V> extends Node<K,V> {
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
// MOVED = -1
super(MOVED, null, null, null);
this.nextTable = tab;
}
// ForwardingNode的查找操作，直接在新数组nextTable上去进行查找
Node<K,V> find(int h, Object k) {
// loop to avoid arbitrarily deep recursion on forwarding nodes
//使用循环，避免多次碰到ForwardingNode导致递归过深
//将nextTable赋值到tab,将查找操作在nextTable中进行查找
outer: for (Node<K,V>[] tab = nextTable;;) {
Node<K,V> e; int n;
//k==null或者nextTable=null或nextTable.length=0,或者数组中没用对应节点直接返回null
//最后一个条件定位了在新表中的位置，方法与HashMap一致
if (k == null || tab == null || (n = tab.length) == 0 ||
//获取头结点赋值到e中
(e = tabAt(tab, (n - 1) & h)) == null)
return null;
//如果是普通链表相当于遍历
//如果是TreeBin节点就调用对应的find函数,
//如果还是ForwardinngNode节点调到外层循环(相当于递归操作)
for (;;) {
int eh; K ek;
//第一个节点就是要查找的节点直接返回
if ((eh = e.hash) == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
//返回头结点
return e;
if (eh < 0) {
//继续碰到ForwardingNode节点情况,这里相当于递归调用本次方法
if (e instanceof ForwardingNode) {
//重新赋值nextTable
tab = ((ForwardingNode<K,V>)e).nextTable;
//返回到指定位置
continue outer;
}
else
//碰见其他的特殊节点,调用对应的find方法进行查找
return e.find(h, k);
}
//普通节点直接遍历链表,如果e.next为空直接返回
if ((e = e.next) == null)
return null;
}
}
}
}
扩容是当数组槽位为空或已经完成数组槽的扩容从操作后，将ForwardingNode节点插入到姐的数组槽中，而find操作在新表中进行查询，敲门的利用了ForwardingNode节点将旧表和新表链接起来，保证了其他线程扩容时也能对节点正常访问。TreeBin节点下面分析。
put代码分析
put操作代码：
//put函数直接调用了putVal
public V put(K key, V value) {
return putVal(key, value, false);
}
//put函数和putIfAbsent函数的具体实现,
//onlyIfAbsent如果是true说明值存在不覆盖保留原有值,
//如果为false说明值存在直接覆盖,保留新的值
final V putVal(K key, V value, boolean onlyIfAbsent) {
//ConcurrentHashMap与HashMap不同不允许key和value为空,HashMap允许为空,key为空时hash为0
if (key == null || value == null) throw new NullPointerException();
//求出hash值
int hash = spread(key.hashCode());
//用于记录hash槽中链表长度,后面会用于判断是否链表过长需要转红黑树
int binCount = 0;
//循环,直到插入成功之后才跳出
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//如果table为空,先初始化table
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//定位到table中的位置,用tabAt函数volatile读取table[i]
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//如果对应槽位为null,未发生hash冲突,直接使用cas进行插入,插入成功直接跳出循环
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin(添加空的节点时候没用锁)
}
//如果头结点的hash=-1,则为ForwardingNode节点,说明在扩容,调用helpTransfer方法帮助扩容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//如果发生hash冲突,节点是普通链表节点或者是树节点,则使用synchronized同步,同步节点为头结点(f为头结点)
//remove/replace也会尝试锁住头结点,这样保证锁住数组槽的头结点能够阻塞其他基本的写操作
synchronized (f) {
//在检查一下,避免加锁的空隙中其他线程进行操作使头结点,类似于单例模式的双重校检锁
if (tabAt(tab, i) == f) {
//fh>0说明头节点是普通链表节点
if (fh >= 0) {
//因为是第一次处理所以计时为1
binCount = 1;
//遍历链表
for (Node<K,V> e = f;; ++binCount) {
K ek;
//找到相等节点,看是否需需要更新value值,通过onlyIfAbsent判断
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
//如果遍历链表没用发现对应节点,直接创建一个新的节点放到链表的末尾
//注意是尾插法
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
//如果是TreeBin节点
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
//调用putTreeVal利用红黑手的方法进行添加
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
//如果链表超过树化阈值,即链表长度太长,则调用treeBin将链表转换为红黑树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
//说明是更新老的节点,直接返回不需要调用addCount
if (oldVal != null)
return oldVal;
break;
}
}
}
//说明是添加新的节点,调用addCont函数计数器加1,可能引发扩容
addCount(1L, binCount);
return null;
}
put操作总结：
插入时首先判断key和value是否为null，如果是null则抛出空指针异常，若不是判断表是否初始化，如果尚未初始化，则调用initTable()函数进行初始化表，否则定位到相应的数组槽，如果数组槽为null说明没有发送hash冲突，利用CAS将节点挂上去，否则判断是否是ForwardingNode节点，若是调用helpTransfer帮助扩容，若均不是，则一定是普通节点或者TreeBing节点，利用synnchronized锁住头结点，如果是普通链表节点，使用尾插法插入，如果是树节点调用putTreeVal函数插入红黑树，最后利用binCount检查链表长度是否超过树化阈值，已经如果是插入新的节点计数器加1并且判断是否需要扩容。
put操作会用synchronized进行加锁，除非数组槽位是null，直接使用CAS操作避免加锁
当链表长度太长，超过树化阈值8，调用treeifyBing将链表转换为红黑树(必须保证hash桶长度超过64否则先扩容，超过64才会进行树化)。
treeifyBin代码
//当链表长度超过8时候转换为红黑树
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
//若表长度小于64不需要转换为红黑树,调用tryPrrsize扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
//扩容一倍
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
//对头结点加锁转换为红黑树
synchronized (b) {
//二次判断头结点,避免在加锁的间隙其他线程对头结点进行的了修改
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
//循环调整链表,创建树节点(其实就是将单向链表转换为双向链表,此时的TreeNode不具备红黑树特点,就是一个双向链表)
for (Node<K,V> e = b; e != null; e = e.next) {
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
//利用TreeBin<K,V>包装红黑树,并放入数组槽中,TreeBin的构造方法会将上面的TreeNode调整到满足红黑树特点
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
TreeBin的操作
get操作中调用了TreeBin的find函数，进行红黑树查找。
put操作则调用了TreeBin的putTreeVal函数进行红黑树插入。
下面是TreeBin的部分代码：
static final class TreeBin<K,V> extends Node<K,V> {
TreeNode<K,V> root;//红黑树的根节点
volatile TreeNode<K,V> first;//链表的头结点,TreeBin仍然保存了链表结构(双向链表)
volatile Thread waiter;//标记设置WAITER标识位的线程
volatile int lockState;//锁状态的标志位
// values for lockState
static final int WRITER = 1; // 写锁标志位
static final int WAITER = 2; // 等待写锁标志位
static final int READER = 4; // 读锁标志位
//省略部分代码
.....
/**
* 锁住根节点
*/
private final void lockRoot() {
//通过CAS更改锁的状态标志位,将其设置为WRITER并调用contendedLock()
if (!U.compareAndSwapInt(this, LOCKSTATE, 0, WRITER))
contendedLock(); // offload to separate method
}
/**
* 解锁,锁状态标志位置0
*/
private final void unlockRoot() {
lockState = 0;
}
/**
* 写线程会调用该方法.由于不需要考虑写-写情况,因此只需要考虑读锁阻塞线程获取写锁
*/
private final void contendedLock() {
boolean waiting = false;
for (int s;;) {
//若锁状态为waiter或为0则可以进入,否则条件不满足
//由于写-写不可能,所以锁状态没有读状态进入该if
// ~ 的意思是按位取反运算符翻转操作数的每一位，即0变成1，1变成0。
if (((s = lockState) & ~WAITER) == 0) {
//使用CAS替换锁状态为WRITER
if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) {
//如果当初线程注册过waiter状态则清楚
if (waiting)
waiter = null;
//返回获得写锁
return;
}
}
//锁状态为非waiter状态则进入,由于不可能是写-写,所以锁状态这时候若为读状态则进入
//即若锁状态有读状态时,且没用waiter状态会进入else id
else if ((s & WAITER) == 0) {
//尝试占据 WAITER 状态标志位
if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {
// 设置waiting标志位,设置为waiter线程
waiting = true;
waiter = Thread.currentThread();
}
}
//若waiting等于true则休眠自己
//锁状态为读状态,waiting标志位true,park当前线程
else if (waiting)
LockSupport.park(this);
}
}
//在红黑树中查找,相等的节点,有两种方式查找
//1:以链表形式查找
//2:以红黑树的方式进行查找
final Node<K,V> find(int h, Object k) {
if (k != null) {
for (Node<K,V> e = first; e != null; ) {
int s; K ek;
//当锁状态为等待或写的状态时,以链表形式进行查找节点,
//当为写状态的时候,读则采取遍历链表的方式,这样虽然时间复杂度提高,但是读写不阻塞
//当为等待状态是,不继续加写锁,能让被阻塞的线程尽快恢复运行,或者刚好让某个线程不阻塞
if (((s = lockState) & (WAITER|WRITER)) != 0) {
if (e.hash == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
e = e.next;
}
//否则设置读锁状态
else if (U.compareAndSwapInt(this, LOCKSTATE, s,
s + READER)) {
TreeNode<K,V> r, p;
try {
//利用红黑树来查找速度很快
p = ((r = root) == null ? null :
r.findTreeNode(h, k, null));
} finally {
Thread w;
//如果最后一个读线程,并且有写线程因为读锁而阻塞,要么通知他,告诉他可以尝试获取写锁了
if (U.getAndAddInt(this, LOCKSTATE, -READER) ==
(READER|WAITER) && (w = waiter) != null)
LockSupport.unpark(w);
}
return p;
}
}
}
return null;
}
/**
* 在TreeBin中插入新的TreeNode或者更新TreeNode
*/
final TreeNode<K,V> putTreeVal(int h, K k, V v) {
Class<?> kc = null;
boolean searched = false;
for (TreeNode<K,V> p = root;;) {
int dir, ph; K pk;
if (p == null) {
first = root = new TreeNode<K,V>(h, k, v, null, null);
break;
}
else if ((ph = p.hash) > h)
dir = -1;
else if (ph < h)
dir = 1;
else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
return p;
else if ((kc == null &&
(kc = comparableClassFor(k)) == null) ||
(dir = compareComparables(kc, k, pk)) == 0) {
if (!searched) {
TreeNode<K,V> q, ch;
searched = true;
if (((ch = p.left) != null &&
(q = ch.findTreeNode(h, k, kc)) != null) ||
((ch = p.right) != null &&
(q = ch.findTreeNode(h, k, kc)) != null))
return q;
}
dir = tieBreakOrder(k, pk);
}
TreeNode<K,V> xp = p;
if ((p = (dir <= 0) ? p.left : p.right) == null) {
TreeNode<K,V> x, f = first;
first = x = new TreeNode<K,V>(h, k, v, f, xp);
if (f != null)
f.prev = x;
if (dir <= 0)
xp.left = x;
else
xp.right = x;
// xp是新添加的节点的父节点，如果它是黑色的，新添加一个红色节点就能够保证x这部分的一部分路径关系不变,这种情况下不用锁根
if (!xp.red)
if (!xp.red)
x.red = true;
else {
//其他情况需要锁根，不锁根的话其他线程调用find等函数，会因为红黑树保持平衡而左旋右旋导致出错。
lockRoot();
try {
root = balanceInsertion(root, x);
} finally {
unlockRoot();
}
}
break;
}
}
assert checkInvariants(root);
return null;
}
....
}
红黑树节点TreeNode实际上好保存有链表的指针，因此也可以用链表的方式进行遍历读取操作，TreeBin自身维护了一个简单的读写锁，由于put/remove/replace方法插入删除替换时会用synchronized锁住根节点，因此不用考虑写-写竞争情况，只需要解决读-写情况。
这里区分一个概念：红黑树的读锁状态和写锁状态是互斥的，但是从ConcurrentHashMap角度来说，读写操作时间上是不互斥的(get操作是无锁状态)，当有线程持有红黑树的读锁时，写线程可能会阻塞，不过因为红黑树查找很快，写线程阻塞时间很短，而当有线程持有红黑树写锁时，读线程不会以红黑树方式进行读取操作，而是简单表方式进行读取，此时读操作和写操作可以并发执行。
上文中find函数和putTreeVal函数就是基本的读写操作，find函数不会发生阻塞(利用遍历链表而不是红黑树，虽然提高了搜索时间复杂度，但避免了阻塞)，而putTreeVal在插入时红黑树需要进行平衡调整下会调用lockRoot()，若此时有其他读操作，则会发生读写阻塞，若红黑树插入后不需要调整则根本不用调用lockRoot()
并发扩容
什么时候扩容？
- 在putVal函数中最后调用addCount函数，对计数器加1，改操作可能会引发扩容
- 如果新增节点之后，所在链表的元素个数达到了阈值8，则会调用treeifyBin方法把链表转换为红黑树，不过在结构转换之前，会对数组长度进行判断，如果长度小于64，则会执行tryPresize执行扩容。
如何扩容？ - 不管addCount还是tryPreSize，他们都有相同的一部分代码，来调用扩容操作。
tryPreSize的代码
//尝试调整表的大小以容纳给定数量的元素。
//size 元素数量（不需要完全准确）
private final void tryPresize(int size) {
//调整size的大小获取一个满足2^n次方条件的数组容量
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
//sizeCtl>=0说明是没有线程在进行扩容,sizeCtl的值为下一次扩容的阈值
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
//table为null的情况
if (tab == null || (n = tab.length) == 0) {
//新数组的length,sizeCtl和根据size计算的容量,那个大取哪个
n = (sc > c) ? sc : c;
//通过CAS更改内存中SIZECTL的值为-1,表示有一个线程正在进行扩容
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设置为下一次扩容的阈值
sc = n - (n >>> 2);
}
} finally {
//将下一次扩容的阈值赋值给sizeCtl
sizeCtl = sc;
}
}
}
//如果table不为null,扩容的阈值(sizeCtl>0时表示扩容的阈值)大于等于需要扩容大小c,或者当前table>2^30次方则不会扩容直接跳出循环
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
//进行扩容
else if (tab == table) {
//计算一个扩容戳,该扩容戳,保证了并发扩容是新表只会生成一次
int rs = resizeStamp(n);
//sc<0说明有别的线程正在进行扩容
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;
//帮助扩容,CAS操作时sc+1(按照对sizeCtl的注释看不应该是减1吗?不太懂)
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
//第一个执行transfer的线程
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
}
}
}
transfer方法
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//根据CPU数来分配任务?(不确定)
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
//第一个线程扩容时nextTab为null
if (nextTab == null) { // initiating
try {
//创建新表是旧表长度的2被
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
//将新表赋值给全局变量nextTable
nextTable = nextTab;
//transferIndex赋值为n,从旧表末尾开始将旧表数据赋值到新表
transferIndex = n;
}
//新表的长度
int nextn = nextTab.length;
//建立ForwardingNode<K,V>节点,当旧表数组槽为空或元素赋值完成后,旧表数组槽中放入该ForwardingNode<K,V>节点
//其他线程若读到该节点,会在新表中读,若发生扩容则会帮助扩容
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
//advance开始为true会进入下面的while循环申请一段数组槽中元素赋值的任务
//完成一个旧数组槽的赋值后也会设置为true
boolean advance = true;
//是否完成所有的赋值标志位
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//申请任务
while (advance) {
int nextIndex, nextBound;
//当前线程承担transfer任务仍未执行完毕,若已经finishing也会进入
if (--i >= bound || finishing)
advance = false;
//将transferIndx赋值到nextIndex中
//若transfeerIndex<=0,则transfer任务已执行完毕
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//申请任务,得到数组nextIndex -1 位置的复制任务
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
//这三种情况表示任务完成
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
//扩容结束,nextTable为null,table置为新表
nextTable = null;
table = nextTab;
//sizeCtl设置为旧表的1.5倍,也就是新表的0.75倍,即新的阈值
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//退出扩容操作
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
//在回去之前重新设置i为n，检查是否已经完全进行复制
i = n; // recheck before commit
}
}
//若当前槽为null,放入ForwardingNode<K,V>节点
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//若当前槽已经为ForwardingNode节点则表示已经处理过了
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
//否则锁住头结点
synchronized (f) {
if (tabAt(tab, i) == f) {
//设置两个链表与hashMap扩容时一样
//ln链表接在新表中的原位置也就是低位
//hn链表接在新表的原位置+旧表长,也就是高位
Node<K,V> ln, hn;
//如果是普通链表节点
//这里和1.8的hashMap扩容基本一致,旧表中的槽在新表中只有两种可能
//一种是仍在原来位置,另一种是原位置+旧表长
if (fh >= 0) {
//根据runBit是0还是1就能判断新表中的位置
//0放入原位置
//1放入新位置原位置+旧表长
int runBit = fh & n;
Node<K,V> lastRun = f;
//先遍历一遍链表,由于链表长度小于8,所以遍历很快结束
//根据后面代码可以看出用意,尽量重用Node链表尾部部分
//比如链表长度为2,新表为4
//若数组槽1中原链表为3->7->5,则runbit=0,lastRun=5
//若数组槽1中原链为5->3->7,则runbit=1，lastRun=3
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
//重用的是新位置是原位置的链
if (runBit == 0) {
ln = lastRun;
hn = null;
}
//重用的新位置在原位置+旧表长的链
else {
hn = lastRun;
ln = null;
}
//遍历重用链表的头结点,之前的链表
for (Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
//根据新表的原位置,头插法插入相应的链
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
//ln放在新表的原位置
setTabAt(nextTab, i, ln);
//hn放入新表的原位置+旧表长度
setTabAt(nextTab, i + n, hn);
//完成操作将就数组槽设置为ForwardingNode节点
setTabAt(tab, i, fwd);
advance = true;
}
else if (f instanceof TreeBin) {
//红黑树赋值方法几乎和链表一样,遍历链表
//形成高低位两个链表,
//根据两个新链表的长度来决定是否转换为数
//最后将两个链表放入新表中对应数组槽中
//设置旧表数组槽设置为ForwardingNode节点
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> lo = null, loTail = null;
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node<K,V> e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null, null);
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
//如果链表长度小于等于6取消树化
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin<K,V>(hi) : t;
//使用CAS更新nextTab的值
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
从旧表复制到新表过程与1.8HashMap的扩容过程几乎一样(多了重用尾部链表这一步看代码感觉链表扩容时候是头插法)，而红黑树的赋值和链表赋值几乎一样。
其他线程发生读操作或者迭代读碰到ForeardingNode时，将操作转发到扩容后的新的table数组上执行，如果写操作碰见则尝试帮助扩容。
总结
- ConcurrentHashMap，get()时，如果只有红黑树需要加锁的原因：
- 链表添加一个元素不会改变链表的内部结构
- 红黑树如果get时候恰好碰见put一个元素的线程会改变红黑树的数据结构，导致找不到对应元素，所以get操作，如果是从红黑树中查找会有存在锁的状态，如果是遍历链表则不用
- 为什么没有写-写锁的冲突
- 因为写操作都会锁住头结点，所以写-写是不存在的因为锁住头结点了不会同时发生写-写竞争
- 红黑树的读锁状态和写锁状态是互斥的，但是从ConcurrentHashMap角度来说，读写操作时间上是不互斥的(get操作是无锁状态)，当有线程持有红黑树的读锁时，写线程可能会阻塞，不过因为红黑树查找很快，写线程阻塞时间很短，而当有线程持有红黑树写锁时，读线程不会以红黑树方式进行读取操作，而是简单表方式进行读取，此时读操作和写操作可以并发执行。
find函数和putTreeVal函数就是基本的读写操作，find函数不会发生阻塞(利用遍历链表而不是红黑树，虽然提高了搜索时间复杂度，但避免了阻塞)。
网友评论