Java8 源码阅读 - ConcurrentHashMap
2020-03-13 本文已影响0人
Mhhhhhhy
JDK提供的一个线程安全的HashMap，日常使用频率高，还是相当有阅读的价值；
特点
ConcurrentHashMap遵循着与java.util.map相同的功能规范，并且有包含与Hashtable每个方法对应的方法版本；尽管所有操作都是线程安全的，但是并不是所有操作都有加锁；
哈希表主要的设计目标是保持并发可读性，通常是get方法，还包括迭代器和相关方法，同时最大程度地减少更新锁竞争，次要目标是使空间消耗保持与java.util.HashMap相同或更好，并提高多线程在空表上的高初始插入率；
检索操作（比如get）一般不会阻塞其他线程，因此可能与更新操作put或者删除操作remove并行；
一些聚合状态的方法比如size、isEmpty、containsValue的结果通常只有在没有其他线程进行并发更新时才能正确的表示，所以这些方法的结果可能适合于监测或预估操作，但不适合于精确的控制程序的暂态状态；
哈希表的每个键值对都保存在一个节点中，大多数节点都是Node类型，另一部分是TreeNode类型；当哈希表中的元素有较多的hash冲突时，表会被动态扩展，每个键具有唯一的哈希码，但是可能会属于一个相同的槽(slot)，每个槽平均维护2个桶(bin)，当然在更新/删除时这个平均值可能会有较大的差异；总的来说，负载因子LOAD_FACTOR还是很好维护了在时间和空间之间做的权衡；
在动态扩展哈希表时是一个相对缓慢的操作，所以在可能的情况下，最好在构造函数中指定预估的大小来避免过多的扩展操作；
表的初始化被延迟到第一次添加元素时，大小为2的幂次方，表中每个bin通常包含一个Node列表，大多数情况下是空或者一两个节点；在将第一个节点插入空的bin时，只需要进行一次cas操作，这也是在ConcurrentHashMap中put操作时会遇到的大多数情况，这种情况下是不需要加锁的；
因为不想浪费过多的锁空间，所以每次都是向每个bin中列表的第一个Node进行加锁，锁定节点时，任何更新操作都必须首先确认是否仍然是锁定后的第一个节点，否则将重试加锁；对每个bin的首节点加锁的主要缺点是：对该bin节点列表的其他节点操作可能会被阻塞，但是这种概率并不是很高；
当bin中保存的节点数大于阀值时，会对该bin中的节点列表进行树化，树化后搜索的时间复杂度为O(logN)；
所有的方法的所有参数必须是非空的。
ConcurrentHashMap的结构和HashMap类似，也是由一个Node类型的数组储存着一个个桶(Bucket)，桶内储存的可能是一个链表的头节点，或者是一个红黑树的头节点；
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) {...}
...
public final int hashCode() { return key.hashCode() ^ val.hashCode(); }
public final String toString(){ return key + "=" + val; }
public final V setValue(V value) {
throw new UnsupportedOperationException();
}
public final boolean equals(Object o) {...}
Node<K,V> find(int h, Object k) {...}
}
节点Node的结构，用final来声明代表了每个Node的Key是不可以修改的，而可以修改的Value和Next指针则用volatile来保证可见性；
Java内存模型规定了每个线程都有自己的私有内存，私有内存中储存了该线程以读写的共享变量副本，为了避免线程私有内存中的副本和主存中变量不一致，需要使用volatile来保证该变量的值被修改后能立刻更新到主存，即修改后能被其他线程读取到最新的值
下面是在类中定义的变量
// 储存数据的数组，第一次插入时延迟进行初始化，大小始终是2的幂次方
transient volatile Node<K,V>[] table;
//扩容时用到的，用来替换旧表，仅在调整大小时非空
private transient volatile Node<K,V>[] nextTable;
//基本计数器值，在没有遇到线程冲突时会更新该值，和LongAddr中基本类似
private transient volatile long baseCount;
/**
* 控制表初始化和表扩容的阀值；
* 如果为负，则表将被初始化或正在调整其大小：
*-1表示初始化，-N 表示有constant(取决于表的大小) - N - 1个线程正在进行扩容操作
* 当table为null时，保留创建时要使用的初始表大小，或者默认为0。
* 例如长度为16的数组，sizeCtl就为16 * 0.75 = 12
*/
private transient volatile int sizeCtl;
/** 记录的是表扩容前的大小，在扩容时使用 */
private transient volatile int transferIndex;
/** 调整大小或创建CounterCell时使用的自旋锁（通过CAS锁定）。*/
private transient volatile int cellsBusy;
/** 保存CounterCell的数组，如果为非null，则大小为2的幂。和LongAddr中的Cells[]一样*/
private transient volatile CounterCell[] counterCells;
还定义了一些操作Table元素的方法
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);
}
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);
}
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
因为Java内存模型中，线程都有自己的私有内存，同时储存了table的副本，尽管table有volatile修饰，但是数组内部储存的对象并不能保证可见性，所以通过Unsafe.getObjectVolatile或Unsafe.putObjectVolatile来获取或更新指定内存数据，也就是强制从主存中获取属性值；
对于数组内部对象的改变是通过cas进行的，cas是cpu提供的原子指令，可以保证在多线程下对同一个数据操作的原子性；
类的内部除了Node节点之外还维护了几种数据结构；
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;
...
}
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;
// values for lockState
static final int WRITER = 1; // 持有写锁时的标志
static final int WAITER = 2; // 等待写锁时的标志
static final int READER = 4; // 设置读锁的增量值
...
}
TreeNode是红黑树结构的节点，若是桶中保存的红黑树结构，TreeBin节点则是该桶的首元素，该节点并不保存键值对，通过root节点指向了具体红黑树结构，同时还维护了lockState来记录当前线程的状态从而达到加锁的目的；
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) { ... }
}
ForwardingNod节点是个特殊的节点，该节点只会在table扩容时才会使用，hash值固定为-1，并且内部储存了新建table的引用；
//构造函数
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
this.sizeCtl = cap;
}
//求出与c最近的2的幂次方
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;
}
在ConcurrentHashMap的文档中，是推荐使用这个构造函数来指定其大小的，可以看到就算指定table大小也只是初始化sizeCtl值；
final V putVal(K key, V value, boolean onlyIfAbsent) {
//无法添加null
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode())
int binCount = 0;
for (Node<K,V>[] tab = table;;) { //循环直到添加成功后才退出
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0) //table为空
tab = initTable(); //初始化table
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//通过hash定位到的桶是否为空 约等于table[i] == null
//如果为空则尝试通过cas赋值 约等于table[i] = new Node
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
break; //更新成功 退出循环
}
// 走到这个判断表示该桶已被其他元素占用 即table[i] != null
else if ((fh = f.hash) == MOVED)
//如果判断成功 则表示当前桶的节点是`ForwardingNode`
//表示其他线程正在扩容 则帮助其迁移
tab = helpTransfer(tab, f);
else {
//table[i] != null 且不在扩容
V oldVal = null;
synchronized (f) { //锁第一个节点
if (tabAt(tab, i) == f) { //加锁成功后再次判断该节点是否被其他线程改变
if (fh >= 0) { //是Node节点
binCount = 1; //计数该桶中有几个节点
for (Node<K,V> e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value; //有相同的hash值，替换value
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {
//在节点末尾添加新的Node
pred.next = new Node<K,V>(hash, key, value, null);
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
//如果是红黑树结构，走红黑树添加的流程
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
//如果节点数大于阀值8，进行树化
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
//增加计数
addCount(1L, binCount);
return null;
}
ConcurrentHashMap对比HashMap来说，为了保证线程安全性，所以是不允许键值对为空的；大多数情况下，通过hash值定位到桶要么是空的，要么是只有1-2个元素，在桶为空的情况下只需要进行一个cas操作即可完成插入，在桶被占用时才需要对桶里的第一个元素进行加锁；
还有一个需要注意的地方，在map进行扩容操作时，是运行其他线程进行添加操作的，如果发现其他线程正在扩容，则会同步帮助它进行扩容从而加快rebuild的时间，这是对比jdk7实现的一个很大的优势；
通常来说大部分使用ConcurrentHashMap时一切的触发逻辑通过添加函数put触发的，所以说理解这个方法了做了什么才能有切入点继续研究其他的方法；
初始化table
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
//如果一个线程发现sizeCtl<0，意味着另外的线程执行CAS操作成功
//当前线程只需要让出cpu时间片
Thread.yield();
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
//cas成功 进入初始化逻辑
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
创建table采用懒加载的方式，通过上面的构造方法可以得知在进行构造的时候只是初始化sizeCtl，只有在第一次添加元素时才会真正的初始化table；当第一个进入到初始化逻辑的线程，会将sizeCtl置为-1，如果其他线程发现sizeCtl为-1时，会调用Thread.yield()尝试通知线程让出cpu时间片；等初始化完成后sizeCtl的含义则是表示table扩容的阀值；
计算Size
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
if ((as = counterCells) != null || //counterCells已被初始化
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {//或者cas失败 表示被其他线程修改了
CounterCell a; long v; int m;
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 || //if counterCells未被初始化 或者 as为空
(a = as[ThreadLocalRandom.getProbe() & m]) == null || //或者 通过hash定位到的槽是空的
!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { //槽被占了 尝试cas更新往槽上累加数据
//上面条件有任何一项不满足 进入和longaddr一样的流程
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount(); //聚合操作获取count
}
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {//当count>=sizectl && tab != null && tab.length < max
int rs = resizeStamp(n); //分配一个rs表示某个线程扩容操作的标志
if (sc < 0) {//<0表示有其他线程正在进行扩容
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt); //协助进行扩容
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
//进到这里后sizeCtl的值为(rs << RESIZE_STAMP_SHIFT) + 2) 通常是负数
//可以通过sizeCtl倒推出当前正在扩容的线程有几个
transfer(tab, null);
s = sumCount();
}
}
}
//根据表大小生成一个标记位 用来计算当前扩容线程数量的
static final int resizeStamp(int n) {
return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}
addCount方法主要做了两件事，一是添加元素后的计数操作，二是检查是否要进行扩容；
- 在多线程环境下如果对同一个size更新可能会导致数据不准确，所以这里用到了和LongAddr类一样的更新手段，主要的思想是采用分治法，遇到线程冲突时把数值一分为二，不同的线程更新不同的小单位，这样冲突的概率就变小了，最后再求和；
- 检查扩容时会遇到两种情况，一是当判断到table的大小大于扩容阀值，但是还没有任何线程进行扩容，会进入到
transfer方法中执行扩容操作；二是当判断到table的大小大于扩容阀值，发现已经有线程正在进行扩容处理，则会帮助进行扩容，也就是说，扩容操作也是并发的；
扩容
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
//stride值规定了每个线程每次循环的个数，最小是16
//表示次会移动table[table.length - 16, table.length - 1]的桶到新的目的地
stride = MIN_TRANSFER_STRIDE;
if (nextTab == null) {
//第一个扩容的线程创建一个新的table
try {
@SuppressWarnings("unchecked")
//大小为原来的两倍
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) {
//当然有可能会失败
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
//transferIndex记录着扩容前大小
transferIndex = n;
}
int nextn = nextTab.length;
//创建ForwardingNode节点，主要功能是占位，代表该bin已经转移过了
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false;
//是从后面的桶往前面遍历的 bound表示每次转移的边界值
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
//通常来说每次循环遍历大部分都是走到这个判断
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
//表示遍历完成
i = -1;
advance = false;
}
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 = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//sizectl值减一，表明该线程完成扩容操作
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
//当前线程结束扩容 但不是最后一个线程 则退出函数
return;
finishing = advance = true;
//因为转移完所有桶 重新循环一次检查
i = n;
}
}
else if ((f = tabAt(tab, i)) == null)
//当前桶是空的 标记为fwd表示已经检查过的
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
//当前桶已被检查过
advance = true;
else {
//转移桶 锁定第一个节点
synchronized (f) {
if (tabAt(tab, i) == f) { //同样判断下加锁后节点是否被改变
Node<K,V> ln, hn;
if (fh >= 0) {
int runBit = fh & n;
Node<K,V> lastRun = f;
//下面的操作就是将桶内链表为拆分成两个链表
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);
}
//将新链表转移到新table中
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
else if (f instanceof TreeBin) {
//如果是红黑树结构
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;
}
}
//如果长度不够则将其转换成链表形式
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;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
扩容操作是非常复杂和难理解的一个函数，对于这些方法可以不必深究每一行代码的具体作用是什么、为什么要这么处理，只需要有一个大概的认识能站在宏观上了解这个函数；
扩容期间是允许其他线程插入数据的，所以说扩容也做了多线程处理，每个参与扩容的线程都有一定的工作边界，范围最低长度为16，遍历是从最后一个桶开始向前遍历的，比如第一个扩容线程的第一次范围就是table[table.length - 16, table.length]；当转移过桶都会标记为ForwardingNode节点表示已经处理完成，当当前线程处理完自己的工作边界时，如果检查到同时有其他线程正在扩容，则会退出扩容操作，如果没有其他线程参与扩容，则会重新更新其工作边界再一次的遍历进行扩容，直到结束；
获取元素
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//通过hash定位到的桶不为空
if ((eh = e.hash) == h) {
//如果桶节点保存的就是该key
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
else if (eh < 0)
//是红黑树节点或者是ForwardingNode节点 调用各自的查找方法
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操作在查找桶的首节点和链表结构的节点时是没有加锁的，这两种情况也是在ConcurrentHashMap中最常见的情况；
Node<K,V> find(int h, Object k) {
//在nextTable中查找元素
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;
//eh为桶内的首元素hash值
if ((eh = e.hash) == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
if (eh < 0) {
//如果eh小于0 则表示可能是ForwardingNode或者TreeBin
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode<K,V>)e).nextTable;
//如果是ForwardingNode节点 则重新回到循环
//这里不用递归语法是无法判断递归的深度 避免StackOverflowError异常
continue outer;
}
else
//红黑树TreeBin节点 使用红黑树的查找方法
return e.find(h, k);
}
if ((e = e.next) == null)
return null;
}
}
}
这个是ForwardingNode节点的查找方法，ForwardingNode在扩容时会用到，如果在查找过程中遇到该节点，则表示有其他线程正在对table扩容，会跳转到新的table中查询；
树节点
ConcurrentHashMap有两个树节点TreeBin和TreeNode，是在红黑树的基础上增加了并非的控制，首先需要大概了解下红黑树；
红黑树的特性
- 每个节点只有红黑两种颜色；
- 根节点是黑色；
- 每个叶子节点（nil）是黑色；
- 每个红色结点的两个子结点一定都是黑色；
- 任意一结点到每个叶子结点的路径都包含数量相同的黑结点；
红黑树添加新元素x的调整过程
- 首先先根据x的key通过二分查找法找到被插入的位置然后添加x；
- 将新添加的元素x颜色置为红色，为什么要改变成红色呢？首先不管这颗红黑树只有1个元素还是有100个元素，他肯定是符合红黑树的特性不然就不叫红黑树了，那么肯定满足任意一结点到每个叶子结点的路径都包含数量相同的黑结点这个条件，如果我们添加的新节点是黑色，势必会造成树的不平衡，而添加红色的话则并不一定会破坏上面那条规则；
- 然后判断x的父节点是否是红色，如果是红色的话则不满足每个红色结点的两个子结点一定都是黑色这个条件，需要调整；如果是黑色的话则跳过步骤4；
- 判断x的父节点(y)是左子树还是右子树，开始分情况了
4.1 右子树
获取y的父节点的左子树yleft，判断yleft的颜色是红色还是黑色
- 红色：
将y和yleft的颜色替换成黑色，将y的父节点颜色替换成红色，然后将x指向y的父节点后回到步骤3；
- 黑色：
判断x是在y的左子树还是右子树-
左子树
以y为中心进行右旋，设置y的父节点为黑色，设置y的父节点的父节点(p)为红色，然后进行左旋，最后回到步骤3；
- 右子树
设置y的颜色为黑色，设置y的父节点的父节点为红色，然后进行左旋，最后回到步骤3；
4.2 左子树
获取y的父节点的右子树yright，判断yright的颜色是红色还是黑色
-
- 红色：
设置y为黑色，设置yright为黑色，设置y的父节点为红色，回到步骤3； - 黑色：
判断x是在y的左子树还是右子树-
左子树
设置y的颜色为黑色，设置y的父节点为红色，以y的父节点为中心进行右旋，然后回到步骤3；
- 右子树
以y为中心进行左旋，然后再以y的父节点进行右旋，和上面反过来；
-
- 最后将根节点的颜色替换成黑色，结束调整；
红黑树查询
treeMap的查询是通过二分法进行查找的，时间复杂度为O(logn)；
TreeBin
TreeNode<K,V> root;
volatile TreeNode<K,V> first;
volatile Thread waiter;
volatile int lockState;
// values for lockState
static final int WRITER = 1; // 持有写锁
static final int WAITER = 2; // 等待写锁
static final int READER = 4; // 设置读锁
再回到源码中，TreeBin作为红黑树的首节点，是不保存实际的键值对的，有一个root节点指向红黑树的首节点，同时也有一个lockState来保持锁信息；
TreeBin(TreeNode<K,V> b) {
super(TREEBIN, null, null, null); //TREEBIN的hash为-2
this.first = b;
TreeNode<K,V> r = null;
for (TreeNode<K,V> x = b, next; x != null; x = next) {
next = (TreeNode<K,V>)x.next;
x.left = x.right = null;
if (r == null) { //首节点
x.parent = null;
x.red = false; //红黑树根节点是黑色的
r = x;
}
else {
K k = x.key;
int h = x.hash;
Class<?> kc = null;
for (TreeNode<K,V> p = r;;) {
int dir, ph;
K pk = p.key; //
if ((ph = p.hash) > h)
dir = -1;
else if (ph < h)
dir = 1;
else if ((kc == null &&
(kc = comparableClassFor(k)) == null) || // 如果Node<K, V>的k没有实现Comparable<K>接口
(dir = compareComparables(kc, k, pk)) == 0) // Node<K, V>的k实现Comparable<K>接口，但是TreeNode节点x和TreeNode节点r的K类型不是和kc一致的
dir = tieBreakOrder(k, pk); //不具有可比性 就使用系统的api判断
TreeNode<K,V> xp = p;
// 调整节点
if ((p = (dir <= 0) ? p.left : p.right) == null) {
x.parent = xp;
if (dir <= 0)
xp.left = x;
else
xp.right = x;
r = balanceInsertion(r, x); //调整平衡节点
break;
}
}
}
}
this.root = r;
}
TreeBin的初始化方法，仔细一看和HashMap的树化操作treeify基本一样，具体就不详细说了；
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;
if (!xp.red)
// 新增的节点x如果是红色节点 则不需要调整整颗树
// 因为添加红色节点是不会影响平衡的
x.red = true;
else {
lockRoot(); // 否则加锁调整节点
try {
root = balanceInsertion(root, x);
} finally {
unlockRoot();
}
}
break;
}
}
assert checkInvariants(root);
return null;
}
这里只关心红黑树的插入操作部分中的并发控制，可以看到将新的元素插入到合适位置后调整数据前进行了加锁操作；
private final void lockRoot() {
if (!U.compareAndSwapInt(this, LOCKSTATE, 0, WRITER))// cas上写锁
//如果cas失败 表示锁被占用
contendedLock();
}
private final void unlockRoot() { lockState = 0; }
private final void contendedLock() {
boolean waiting = false;
for (int s;;) {//一开始s为0
if (((s = lockState) & ~WAITER) == 0) {
// WAITER 二进制为10 ~WAITER十进制为-3 s&-3 == 0 表示 s的二进制10（2）或者为0（0） 即lockState为WAITER或者是0
if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) {//再次尝试加写锁
//如果在WAITER或者是未持有任何锁的情况下 可能会加锁成功
if (waiting)
waiter = null;
return;
}
}
else if ((s & WAITER) == 0) {
// s&10 == 0 表示 s=01或00或者11111....1101（即表示-3, 也就是~WAITER）
// 即lockState为WRITER或者是未持有任何锁的情况
if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {
// 01 | 10 = 11, 00 | 10 = 10 (WAITER), -3 | 2 = -1（二进制为1111....1111）
// cas成功 进入等到状态
waiting = true;
waiter = Thread.currentThread();//记录当前线程
}
}
else if (waiting)
LockSupport.park(this);//阻塞线程 等待锁释放
}
}
lockRoot方法在红黑树加锁或者移除节点时会用到，锁住的是树的root节点，然后注意该方法的并非控制，只有当当前树root节点没有添加任何锁的时候，才会加锁成功，否则进入contendedLock，该方法最主要的功能就是使线程自旋重复添加写锁直到成功，当遇到root节点被其他线程添加写锁的时候，该线程会进入阻塞状态；
final Node<K,V> find(int h, Object k) {
if (k != null) {
for (Node<K,V> e = first; e != null; ) {
int s; K ek;
// s & 11 != 0 表示 s不可能是0 代表lockState已经记录了状态 可能是写锁也可能是正在等待写锁
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;
}
树节点查找方法中有两个特点，
- 如果遇到其他线程对树的root节点添加了写锁或者正在有线程等待添加写锁，则不会对其添加读锁，查找模式会从原来的树查找变成链表查找，因为维护红黑树的结构可能正在被调整，而属于
Node结构的next字段早在添加调整树结构前就赋值完毕，就是下图红框那里，这么做的原因就是保证在写的同时可以读取数据；
- 在没有遇到写锁时，会添加一个读锁，然后再通过树的查找方式查找，每当有个线程添加读锁时会对
LOCKSTATE自增4，最后等所有读锁释放时会坚持是否有写线程被阻塞，有如果将其释放，这就是树结构并发控制的核心；
结尾
这个类看的过程中历时大概1个多月，断断续续的，内容很多也很难理解，暂时就打算写到这里，后续如果有时间再继续研究下其他方法；对于看源码不同的人有不同的理解和目的，个人觉得并不需要每一行都能知其然知其所以然，固然细究每个细节且能够坚持下来是可以很大程度的提高程序员的水平，但是在时间精力有限的情况下并不是每个人都有这样的条件，这时候可以选择站在比较宏观的角度看代码，即是掌握一个度并能够理解其思想，这也是考验人水平的一个方面；
