JUC并发—8.并发安全集合二
大纲
1.JDK 1.7的HashMap的死循环与数据丢失
2.ConcurrentHashMap的并发安全
3.ConcurrentHashMap的设计介绍
4.ConcurrentHashMap的put操作流程
5.ConcurrentHashMap的Node数组初始化
6.ConcurrentHashMap对Hash冲突的处理
7.ConcurrentHashMap的并发扩容机制
8.ConcurrentHashMap的分段锁统计元素数据
9.ConcurrentHashMap的查询操作是否涉及锁
10.ConcurrentHashMap中红黑树的使用
7.ConcurrentHashMap的并发扩容机制
(1)ConcurrentHashMap扩容的前置操作
(2)ConcurrentHashMap并发扩容的机制
(3)ConcurrentHashMap并发扩容的流程
(1)ConcurrentHashMap扩容的前置操作
ConcurrentHashMap的tryPresize()方法用于处理数组扩容前的前置操作，该方法主要分为四部分。
第一部分：
首先通过tableSizeFor()方法计算传入size的最小的2的幂次方。
第二部分：
然后判断Node数组是否已初始化，如果还没初始化则要先进行初始化。初始化时会计算扩容阈值为数组大小的0.75倍 + 将扩容阈值赋值给sizeCtl。
第三部分：
如果Node数组已经初始化，则判断是否需要进行扩容。如果Node数组已经被其他线程完成扩容，则当前线程退出循环，无需扩容。如果Node数组已达到最大容量，则无法再进行扩容，也需退出循环。
第四部分：
调用transfer()方法开始执行扩容操作。如果sizeCtl < 0，说明此时已经有其他线程在执行扩容了。如果sizeCtl >= 0，说明此时没有其他线程进行扩容。当前线程都会先通过CAS成功设置sizeCtl后，再调用transfer()方法来扩容。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
//Returns a power of two table size for the given desired capacity.
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;
}
//Tries to presize table to accommodate the given number of elements.
private final void tryPresize(int size) {
//一.首先通过tableSizeFor()方法计算传入size的最小的2的幂次方
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>> 1) + 1);
int sc;
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
//二.判断Node数组是否已经初始化，如果还没初始化，需要先进行初始化
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
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 = n - (n >>> 2);//扩容阈值为数组大小的0.75倍
}
} finally {
sizeCtl = sc;//将扩容阈值赋值给sizeCtl
}
}
}
//三.如果Node数组已经初始化，则判断是否需要进行扩容
else if (c <= sc || n >= MAXIMUM_CAPACITY) {
//c <= sc，说明Node数组已经被其他线程完成扩容了，不需要再进行扩容
//n >= MAXIMUM_CAPACITY，说明Node数组已达到最大容量，无法再进行扩容
break;
}
//四.调用transfer()方法开始执行扩容操作
else if (tab == table) {
int rs = resizeStamp(n);
//如果sc < 0，说明此时已经有其他线程在执行扩容了
//于是当前线程可以先通过CAS成功设置sizeCtl的值后，再调用transfer()方法协助扩容
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;
}
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nt);
}
}
//如果sc >= 0，说明此时没有其他线程进行扩容
//于是当前线程也是先通过CAS成功设置sizeCtl的值后，再调用transfer()方法进行扩容
else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) {
transfer(tab, null);
}
}
}
}
...
}(2)ConcurrentHashMap并发扩容的机制
一.ConcurrentHashMap中的扩容设计
二.多线程并发扩容的原理
一.ConcurrentHashMap中的扩容设计
扩容就是创建一个2倍原大小的数组，然后把原数组的数据迁移到新数组中。但多线程环境下的扩容，需要考虑其他线程会同时往数组添加元素的情况。如果简单地对扩容过程增加一把同步锁，保证扩容过程不存在其他线程操作，那么就会对性能的损耗特别大，特别是数据量比较大时，阻塞的线程会很多。
首先使用CAS来实现计算每个线程的迁移区间。然后使用synchronized把锁粒度控制到每个数组元素上。如果数组有16个元素就有16把锁，如果数组有32个元素就有32把锁。接着如果线程A在进行数组扩容时，线程B要修改数组的某个元素f。那么就让修改元素的线程加入迁移，从而实现多线程并发扩容来提高效率。等数组扩容完成后，线程B才继续去修改元素f。最后通过高低位迁移逻辑计算出高位链和低位链，大大减少了数据迁移次数。
二.多线程并发扩容的原理
当存在多个线程并发扩容及数据迁移时，默认会给每个线程分配一个区间。这个区间的默认长度是16，每个线程会负责自己区间内的数据迁移工作。如果只有两个线程对长度为64的数组迁移数据，则每个线程要做2次迁移，迁移过程会依赖transferIndex来更新每个线程的迁移区间。
(3)ConcurrentHashMap并发扩容的流程
ConcurrentHashMap的transfer()方法用于处理数组扩容时的流程细节，该方法主要分为五部分：
第一部分：创建扩容后的数组
第二部分：计算当前线程的数据迁移区间
第三部分：更新扩容标记advance
第四部分：开始数据迁移和扩容
第五部分：完成迁移后的判断
第一部分：创建扩容后的数组
这部分代码主要做两件事情：
一.计算每个线程处理的迁移区间长度，默认是16。
二.初始化一个新的数组nt，赋值给方法入参nextTab和全局变量nextTable。该数组的长度是原数组的2倍，并且设置transferIndex的值为为原数组大小。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
//The next table to use; non-null only while resizing.
private transient volatile Node<K,V>[] nextTable;
...
//Moves and/or copies the nodes in each bin to new table.
//tab是原数组，nextTab是扩容后的数组
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//计算每个线程处理的迁移区间长度，默认是16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) {
stride = MIN_TRANSFER_STRIDE;//subdivide range
}
//初始化一个新的数组nt，赋值给方法入参nextTab和全局变量nextTable，该数组的长度是原数组的2倍
//并且设置transferIndex的值为为原数组大小
if (nextTab == null) {//initiating
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];//扩容为2n
nextTab = nt;//将创建的扩容数组赋值给nextTab
} catch (Throwable ex) {//try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;//将创建的扩容数组赋值给nextTable
transferIndex = n;//设置transferIndex为原来的数组大小
}
...
}
}第二部分：计算当前线程的数据迁移区间
下面的while循环会计算每个执行到此处的线程需要负责的数据迁移区间。假设当前数组长度是32，需要扩容到64。那么此时transferIndex = 32，nextn = 64，n = 32。
当前线程第一次for循环：nextIndex被transferIndex赋值为32，之后CAS修改transferIndex。CAS修改成功后，nextBound = 32 - 16 = 16，transferIndex = 16。所以bound = 16，i = 31，当前线程负责的迁移区间为[bound, i] = [16, 31]。
当前线程第二次for循环，或者有其他线程进来第一次for循环：由于此时transferIndex = 16，所以nextIndex会被transferIndex赋值为16。之后CAS修改transferIndex为0，修改成功后，nextBound = 16 - 16 = 0。所以bound = 0，i = 15，此时线程负责的迁移区间为[bound, i] = [0, 15]。
需要注意的是：每次循环都会通过if (--i >= bound || finishing)判断区间是否已迁移完成。如果已完成，则会继续进入while循环中的CAS，获取新的迁移区间。
数组从高位往低位进行迁移，比如第一次for循环，处理的区间是[16, 31]。那么就会从位置为31开始往前进行遍历，对每个数组元素进行数据迁移。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
//The next table to use; non-null only while resizing.
private transient volatile Node<K,V>[] nextTable;
...
//Moves and/or copies the nodes in each bin to new table.
//tab是原数组，nextTab是扩容后的数组
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
...
int nextn = nextTab.length;//扩容后的数组长度
//继承自Node的ForwardingNode表示一个正在被迁移的Node
//当原数组中某位置的数据完成迁移后，会对该位置设置一个ForwardingNode，表示该位置已经处理过了
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
//advance字段用来判断是否还有待处理的数据迁移工作，也就是扩容标记
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
//当前线程负责的迁移区间是[bound, i]
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//while循环会计算每个执行到此处的线程需要负责的数据迁移区间
while (advance) {
//假设当前数组长度是32，需要扩容到64；
//那么此时transferIndex = 32，nextn = 64，n = 32；
//刚开始循环时i = 0，nextIndex被transferIndex赋值为32
int nextIndex, nextBound;
if (--i >= bound || finishing) {
//一开始i = bound = 0，所以不会进入这里，而是进入U.compareAndSwapInt()的条件中
//但后来bound = 16, i = 31后，就会进入这里，退出循环
//此后，每次--i，当i = bound = 16时，就又会进入U.compareAndSwapInt()的条件中，重新获取数据迁移区间
advance = false;
} else if ((nextIndex = transferIndex) <= 0) {
//判断当前线程是否已经分配到了新的迁移区间
i = -1;
advance = false;
} else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) {
//如果CAS设置transferIndex从nextIndex=32变为nextIndex-16=16成功后，
//那么advance设置为false，退出while循环
//当前线程再次进入while循环或者其他线程也进入while循环，就会从transferIndex = 16开始，计算要负责的迁移区间
bound = nextBound;//第一次for循环nextBound = 16
i = nextIndex - 1;//第一次for循环i = 31
advance = false;
}
}
...
}
...
}
//A node inserted at head of bins during transfer operations.
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;
}
...
}
}第三部分：更新扩容标记advance
如果位置i的数组元素Node为空，说明该Node对象不需要迁移。所以通过casTabAt()方法修改原数组在位置i的元素为fwd对象，这样其他线程在进行put()操作的时候就可以发现当前数组正在扩容。
如果位置i的数组元素Node的hash值为MOVED，那么说明该Node对象已经被迁移了。所以设置扩容标记位advance为true，等下次for循环时进入while循环--i。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
//The next table to use; non-null only while resizing.
private transient volatile Node<K,V>[] nextTable;
//The array of bins. Lazily initialized upon first insertion.
//Size is always a power of two. Accessed directly by iterators.
transient volatile Node<K,V>[] table;
...
//Moves and/or copies the nodes in each bin to new table.
//tab是原数组，nextTab是扩容后的数组
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
...
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
...
//当前线程负责的迁移区间是[bound, i]
for (int i = 0, bound = 0;;) {
...
} else if ((f = tabAt(tab, i)) == null) {
//第三部分开始：更新扩容标记advance，这样其他线程在put()的时候就可以发现当前数组正在扩容
advance = casTabAt(tab, i, null, fwd);
} else if ((fh = f.hash) == MOVED) {
//设置扩容标记位advance为true，等下次for循环时进入while循环--i
advance = true; // already processed
//第三部分结束
} else {
...
}
...
}
//虽然table变量使用了volatile修饰，但这只保证了table引用对于所有线程的可见性，还不能保证table数组中的元素的修改对于所有线程是可见的
//因此需要通过Unsafe类的getObjectVolatile()来保证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);
}
}第四部分：开始数据迁移和扩容
首先对当前要迁移的Node结点f添加同步锁synchronized，避免多线程竞争。
如果结点f的哈希值大于0，则表示Node结点f为链表或普通结点，那么此时就需要按照链表或普通结点的方式来进行数据迁移。
如果结点f属于TreeBin类型，则表示结点f为红黑树，那么此时就要按红黑树的规则进行数据迁移。
需要注意的是，数据迁移之后可能会存在红黑树转化为链表的情况，当链表长度小于等于6时，红黑树就会转化为链表。
接着使用高位链和低位链的计算方法构造高位链和低位链，遍历链表的每一个结点，计算p.hash & n的值。如果值为0，表示需要迁移，属于高位链；否则不需要迁移，属于低位链。
比如在数组长度为16的一个链表中，hash值为：4, 20, 52, 68, 84, 100。经过hash & (n - 1)得到的下标位置都是4，接着数组长度需要扩容到32。于是经过hash & (n - 1)计算，发现20, 52, 84对应的下标变成了20。这就意味着，这个链表中hash值为20, 52, 84的结点需要迁移到位置20。
最后把低位链设置到扩容后的数组的位置i，把高位链设置到位置i + n。此时当前线程已处理完位置为i的数据迁移，于是设置advance为true，让后续的for循环可以进入while循环来实现对i的递减继续迁移数据。
第五部分：完成迁移后的判断
如果数据迁移完成了，则把扩容后的数组赋值给table。如果还没完成数据迁移，则通过CAS修改并发扩容的线程数。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
//The next table to use; non-null only while resizing.
private transient volatile Node<K,V>[] nextTable;
...
//Moves and/or copies the nodes in each bin to new table.
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
//第一部分开始：创建扩容后的数组
int n = tab.length, stride;//n就是原数组大小
//计算每个线程处理的迁移区间长度，默认是16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) {
stride = MIN_TRANSFER_STRIDE;//subdivide range
}
//初始化一个新的数组nt，赋值给方法入参nextTab和全局变量nextTable，该数组的长度是原数组的2倍
//并且设置transferIndex的值为为原数组大小
if (nextTab == null) {//initiating
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];//扩容为2n
nextTab = nt;//将创建的扩容数组赋值给nextTab
} catch (Throwable ex) {//try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;//将创建的扩容数组赋值给nextTable
transferIndex = n;//设置transferIndex为原来的数组大小
}
//第一部分结束
//第二部分开始：计算当前线程的数据迁移区间
int nextn = nextTab.length;//扩容后的数组长度
//继承自Node的ForwardingNode表示一个正在被迁移的Node
//当原数组中某位置的数据完成迁移后，会对该位置设置一个ForwardingNode，表示该位置已经处理过了
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
//advance字段用来判断是否还有待处理的数据迁移工作
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
//当前线程负责的迁移区间是[bound, i]
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//while循环会计算每个执行到此处的线程需要负责的数据迁移区间
while (advance) {
//假设当前数组长度是32，需要扩容到64；
//那么此时transferIndex = 32，nextn = 64，n = 32；
//刚开始循环时i = 0，nextIndex被transferIndex赋值为32
int nextIndex, nextBound;
if (--i >= bound || finishing) {
//一开始i = bound = 0，所以不会进入这里，而是进入U.compareAndSwapInt()的条件中
//但后来bound = 16, i = 31后，就会进入这里，退出循环
//此后，每次--i，当i = bound = 16时，就又会进入U.compareAndSwapInt()的条件中，重新获取数据迁移区间
advance = false;
} else if ((nextIndex = transferIndex) <= 0) {
//判断当前线程是否已经分配到了新的迁移区间
i = -1;
advance = false;
} else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) {
//如果CAS设置transferIndex从nextIndex=32变为nextIndex-16=16成功后，
//那么advance设置为false，退出while循环
//当前线程再次进入while循环或者其他线程也进入while循环，就会从transferIndex = 16开始，计算要负责的迁移区间
bound = nextBound;//第一次for循环nextBound = 16
i = nextIndex - 1;//第一次for循环i = 31
advance = false;
}
}
//第二部分结束
if (i < 0 || i >= n || i + n >= nextn) {
//第五部分开始：完成迁移后的判断
int sc;
//如果数据迁移完成了，则把扩容后的数组赋值给table
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//如果还没完成数据迁移，则通过CAS修改并发扩容的线程数
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) {
return;
}
finishing = advance = true;
i = n; // recheck before commit
}
//第五部分结束
} else if ((f = tabAt(tab, i)) == null) {
//第三部分开始：更新扩容标记advance，这样其他线程在put()的时候就可以发现当前数组正在扩容
advance = casTabAt(tab, i, null, fwd);
} else if ((fh = f.hash) == MOVED) {
//设置扩容标记位advance为true，等下次for循环时进入while循环--i
advance = true; // already processed
//第三部分结束
} else {
//第四部分开始：开始数据迁移和扩容
synchronized (f) {//首先对当前要迁移的Node结点f添加同步锁synchronized，避免多线程竞争
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
//如果fh >= 0，则表示Node结点f为链表或普通结点，此时需要按照链表或普通结点的方式来进行数据迁移
if (fh >= 0) {
int runBit = fh & n;
Node<K,V> lastRun = f;
//for循环遍历链表，计算出当前链表最后一个需要迁移或者不需要迁移的结点位置
//遍历链表的每一个结点，计算p.hash & n，如果值为0，表示需要迁移，否则不需要迁移
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作为参数，以ln为基础构造低位链，不需要迁移
ln = new Node<K,V>(ph, pk, pv, ln);
} else {
//将hn作为参数，以hn为基础构造高位链，需要迁移
hn = new Node<K,V>(ph, pk, pv, hn);
}
}
//把低位链设置到扩容后的数组的位置i
setTabAt(nextTab, i, ln);
//把高位链设置到扩容后的数组的位置i + n
setTabAt(nextTab, i + n, hn);
//当原数组中某位置的数据完成迁移后，会对该位置设置一个ForwardingNode，表示该位置已经处理过了
setTabAt(tab, i, fwd);
//当前线程已处理完位置为i的数据的迁移，于是设置advance为true，让后续的for循环继续进入while循环来实现对i的递减
advance = true;
} else if (f instanceof TreeBin) {
//如果f instanceof TreeBin，则表示结点f为红黑树，需要按照红黑树的规则进行数据迁移
//需要注意的是，数据迁移之后可能会存在红黑树转化为链表的情况，当链表长度小于等于6时，红黑树就会转化为链表
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;
}
}
}
//第四部分结束
}
}
}
...
}8.ConcurrentHashMap的分段锁统计元素数据
(1)ConcurrentHashMap维护数组元素个数思路
(2)ConcurrentHashMap维护数组元素个数流程
(3)维护数组元素个数的addCount()方法
(4)维护数组元素个数的fullAddCount()方法
(5)获取数组元素个数的size()方法
(1)ConcurrentHashMap维护数组元素个数思路
当调用完put()方法后，ConcurrentHashMap必须增加当前元素的个数，以方便在size()方法中获得存储的元素个数。
在常规的集合中，只需要一个全局int类型的字段保存元素个数即可。每次添加一个元素，就对这个size变量 + 1。
在ConcurrentHashMap中，需要保证对该变量修改的并发安全。如果使用同步锁synchronized，那么性能开销比较大，不合适。所以ConcurrentHashMap使用了自旋 + 分段锁来维护元素个数。
(2)ConcurrentHashMap维护数组元素个数流程
ConcurrentHashMap采用了两种方式来保存元素的个数。当线程竞争不激烈时，直接使用baseCount + 1来增加元素个数。当线程竞争比较激烈时，构建一个CounterCell数组，默认长度为2。然后随机选择一个CounterCell，针对该CounterCell中的value进行保存。
增加元素个数的流程如下：
(3)维护数组元素个数的addCount()方法
addCount()方法的作用主要包括两部分：
一.累加ConcurrentHashMap中的元素个数
二.通过check >= 0判断是否需要进行数组扩容
其中增加数组元素个数的核心逻辑是：
首先通过CAS修改全局成员变量baseCount来进行累加。注意会先判断(as = counterCells) != null，再尝试对baseCount进行累加。这是因为如果一个集合发生过并发，那么后续发生并发的可能性会更大。如果CAS累加baseCount失败，则尝试使用CounterCell来进行累加。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
//Base counter value, used mainly when there is no contention,
but also as a fallback during table initialization races. Updated via CAS.
private transient volatile long baseCount;
//Table of counter cells. When non-null, size is a power of 2.
private transient volatile CounterCell[] counterCells;
private static final long BASECOUNT;
static {
try {
U = sun.misc.Unsafe.getUnsafe();
...
BASECOUNT = U.objectFieldOffset(k.getDeclaredField("baseCount"));
...
} catch (Exception e) {
throw new Error(e);
}
}
//Maps the specified key to the specified value in this table.
//Neither the key nor the value can be null.
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
...
//调用addCount()方法统计Node数组元素的个数
addCount(1L, binCount);
return null;
}
//Adds to count, and if table is too small and not already resizing, initiates transfer.
//If already resizing, helps perform transfer if work is available.
//Rechecks occupancy after a transfer to see if another resize is already needed because resizings are lagging additions.
//x是要增加的数组元素个数
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
//首先通过CAS修改全局成员变量baseCount来进行累加
//注意：这里先判断(as = counterCells) != null，再尝试对baseCount进行CAS累加
//这是因为如果一个集合发生过并发，那么后续发生并发的可能性会更大，这种思想在并发编程中很常见
if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
//增加数组元素个数
CounterCell a; long v; int m;
boolean uncontended = true;
//如果CAS修改baseCount失败，则尝试使用CounterCell来进行累加
//1.as == null，说明CounterCell数组还没初始化
//2.(m = as.length - 1) < 0，说明CounterCell数组还没初始化
//3.(a = as[ThreadLocalRandom.getProbe() & m]) == null，说明CounterCell数组已经创建了，
//但是Hash定位到的数组位置没有对象实例，说明这个数字还存在没有CounterCell实例对象的情况
//4.如果U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)返回false，说明存在多线程竞争
if (as == null || (m = as.length - 1) < 0
|| (a = as[ThreadLocalRandom.getProbe() & m]) == null
|| !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
//调用fullAddCount()方法实现数组元素个数的累加
fullAddCount(x, uncontended);
return;
}
if (check <= 1) {
return;
}
//sumCount()方法返回总的元素个数，也就是CounterCell数组的元素个数和baseCount两者的和
s = sumCount();
}
if (check >= 0) {
//处理数组扩容
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 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)) {
transfer(tab, null);
}
s = sumCount();
}
}
}
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a;
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null) {
sum += a.value;
}
}
}
return sum;
}
...
}(4)维护数组元素个数的fullAddCount()方法
fullAddCount()方法的作用主要包括三部分：初始化CounterCell数组、增加数组元素个数、对CounterCell数组扩容。
注意：为了定位当前线程添加的数组元素个数应落到CounterCell数组哪个元素，会使用ThreadLocalRandom确定当前线程对应的hash值，由该hash值和CounterCell数组大小进行类似于取模的位与运算来决定。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
//Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
private transient volatile int cellsBusy;
//Table of counter cells. When non-null, size is a power of 2.
private transient volatile CounterCell[] counterCells;
//x是要增加的数组元素个数
private final void fullAddCount(long x, boolean wasUncontended) {
//通过ThreadLocalRandom来确定当前线程对应的hash值
int h;
if ((h = ThreadLocalRandom.getProbe()) == 0) {
ThreadLocalRandom.localInit();//force initialization
h = ThreadLocalRandom.getProbe();
wasUncontended = true;
}
boolean collide = false;// True if last slot nonempty
for (;;) {
CounterCell[] as; CounterCell a; int n; long v;
//(as = counterCells) != null && (n = as.length) > 0，表示counterCells数组已经完成初始化
if ((as = counterCells) != null && (n = as.length) > 0) {
//第二部分：增加数组元素个数，分两种情况
if ((a = as[(n - 1) & h]) == null) {
//情况一：(a = as[(n - 1) & h]) == null，表示将当前线程定位到counterCells数组的某位置的元素为null
//此时直接把当前要增加的元素个数x保存到新创建的CounterCell对象，然后将该对象赋值到CounterCell数组的该位置即可
if (cellsBusy == 0) {//Try to attach new Cell
//先创建一个CounterCell对象，并把x保存进去
CounterCell r = new CounterCell(x);//Optimistic create
//U.compareAndSwapInt(this, CELLSBUSY, 0, 1)返回true，表示当前线程占有了锁
if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
boolean created = false;
try {//Recheck under lock
CounterCell[] rs; int m, j;
if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) {
//把新构建的保存了元素个数x的CounterCell对象保存到rs[j]的位置
rs[j] = r;
created = true;
}
} finally {
cellsBusy = 0;
}
if (created) {
break;
}
continue;//Slot is now non-empty
}
}
collide = false;
} else if (!wasUncontended) {//CAS already known to fail
wasUncontended = true;//Continue after rehash
} else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)) {
//情况二：如果将当前线程定位到counterCells数组的某位置的元素不为null，
//那么直接通过U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)操作，对counterCells数组的指定位置进行累加
break;
} else if (counterCells != as || n >= NCPU) {
collide = false;//At max size or stale
} else if (!collide) {
collide = true;
} else if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
//第三部分：counterCells数组扩容
//需要先通过cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)，抢占锁
try {
if (counterCells == as) {// Expand table unless stale
//在原有的基础上扩容一倍
CounterCell[] rs = new CounterCell[n << 1];
//通过for循环进行数据迁移
for (int i = 0; i < n; ++i) {
rs[i] = as[i];
}
//把扩容后的对象赋值给counterCells
counterCells = rs;
}
} finally {
//恢复标识
cellsBusy = 0;
}
collide = false;
continue;//继续下一次自旋
}
h = ThreadLocalRandom.advanceProbe(h);
} else if (cellsBusy == 0 && counterCells == as && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
//第一部分：初始化CounterCell数组
//cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)，通过cellsBusy字段来抢占锁，通过CAS修改该字段值为1表示抢到锁
boolean init = false;
try {//Initialize table
if (counterCells == as) {
//构造一个元素个数为2的CounterCell数组
CounterCell[] rs = new CounterCell[2];
//把要增加的数组元素个数x，保存到CounterCell数组的某个元素中
rs[h & 1] = new CounterCell(x);
//把初始化的CounterCell数组赋值给全局对象counterCells
counterCells = rs;
init = true;
}
} finally {
//恢复标识
cellsBusy = 0;
}
if (init) {
break;
}
} else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x)) {
break;//Fall back on using base
}
}
}
...
}(5)获取数组元素个数的size()方法
sumCount()方法会先得到baseCount的值，保存到sum字段中。然后遍历CounterCell数组，把每个value进行累加。
注意：size()方法在计算总的元素个数时并没有加锁，所以size()方法返回的元素个数不一定代表此时此刻总数量。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
//Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
private transient volatile int cellsBusy;
//Table of counter cells. When non-null, size is a power of 2.
private transient volatile CounterCell[] counterCells;
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 : (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int)n);
}
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a;
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null) {
sum += a.value;
}
}
}
return sum;
}
...
}9.ConcurrentHashMap的查询操作是否涉及锁
(1)put操作会加锁
(2)size操作不会加锁
(3)get操作也不会加锁
(1)put操作会加锁
首先尝试通过CAS设置Node数组对应位置的Node元素。如果该位置的Node元素非空，或者CAS设置失败，则说明发生了哈希冲突。此时就会使用synchronized关键字对该数组元素加锁来处理链表或者红黑树。
其实JUC还可以继续优化，先用CAS尝试修改哈希冲突下的链表或红黑树。如果CAS修改失败，再通过使用synchronized对该数组元素加锁来处理。
(2)size操作不会加锁
size()方法在计算总的元素个数时并没有加锁，所以size()方法返回的元素个数不一定代表此时此刻数组元素的总数量。
(3)get操作也不会加锁
get()方法也使用了CAS操作，通过Unsafe类让数组中的元素具有可见性。保证线程根据tabAt()方法获取数组的某个位置的元素时，能获取最新的值。所以get不加锁，但通过volatile读数组，可以保证读到数组元素的最新值。
虽然table变量使用了volatile，但这只保证了table引用对所有线程的可见性，还不能保证table数组中的元素的修改对于所有线程是可见的。因此才通过Unsafe类的getObjectVolatile()来保证table数组中元素的可见性。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
//The array of bins. Lazily initialized upon first insertion.
//Size is always a power of two. Accessed directly by iterators.
transient volatile Node<K,V>[] table;
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) {
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek))) {
return e.val;
}
} else if (eh < 0) {
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;
}
//获取Node数组在位置i的元素，通过Unsafe类让数组中的元素具有可见性
//虽然table变量使用了volatile修饰，但这只保证了table引用对于所有线程的可见性，还不能保证table数组中的元素的修改对于所有线程是可见的
//因此需要通过Unsafe类的getObjectVolatile()来保证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);
}
...
}10.ConcurrentHashMap中红黑树的使用
(1)treeifyBin()方法的逻辑
(2)TreeBin的成员变量和方法
(3)TreeBin在构造方法中将链表转为红黑树
(4)TreeBin在插入元素时实现红黑树的自平衡
(1)treeifyBin()方法的逻辑
put操作中当发现链表元素>=8时会调用treeifyBin()方法将链表转为红黑树。首先通过tabAt()方法从Node数组中获取位置为index的元素并赋值给变量b，然后使用synchronized对Node数组中位置为index的元素b进行加锁，接着通过for循环遍历Node数组中位置为index的元素b这个链表，并且根据链表中每个结点的数据封装成一个TreeNode对象来组成新链表，最后把新链表的头结点作为参数传给TreeBin构造方法来完成红黑树的构建。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
//Replaces all linked nodes in bin at given index unless table is too small, in which case resizes instead.
//将Node数组tab中位置为index的元素，从链表转化为红黑树
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
if ((n = tab.length) < MIN_TREEIFY_CAPACITY) {
tryPresize(n << 1);//数组扩容
} else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {//b就是链表，先用synchronized对b加锁，保证并发安全
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;//hd是新链表的头结点，tl是新链表的尾结点
//将链表b赋值给e，然后遍历通过e.next赋值回给e来遍历链表
for (Node<K,V> e = b; e != null; e = e.next) {
//根据Node结点e来封装一个TreeNode对象
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;
}
//将构造好的新链表的头结点hd作为参数，创建一个TreeBin对象
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
...
}(2)TreeBin的成员变量和方法
ConcurrentHashMap中红黑树用继承自Node的TreeNode来表示。TreeBin则主要提供了红黑树的一系列功能的实现，并且实现了读写锁。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
//Nodes for use in 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;
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;
}
...
}
//TreeNodes used at the heads of bins.
//TreeBins do not hold user keys or values, but instead point to list of TreeNodes and their root.
//They also maintain a parasitic read-write lock forcing writers (who hold bin lock)
//to wait for readers (who do not) to complete before tree restructuring operations.
static final class TreeBin<K,V> extends Node<K,V> {
TreeNode<K,V> root;//红黑树根结点
volatile TreeNode<K,V> first;//链表头结点，由构造函数传入
volatile Thread waiter;//保存最近一个抢占写锁的线程(如果有值，则说明lockState是读锁状态)
volatile int lockState;//表示锁的状态
// values for lockState
static final int WRITER = 1;//写锁状态
static final int WAITER = 2;//等待获取写锁状态
static final int READER = 4;//读锁状态
...
//构造函数，将以b为头结点的链表转换为红黑树
//Creates bin with initial set of nodes headed by b.
TreeBin(TreeNode<K,V> b) {
...
}
//对红黑树的根结点加写锁
//Acquires write lock for tree restructuring.
private final void lockRoot() {
if (!U.compareAndSwapInt(this, LOCKSTATE, 0, WRITER)) {
contendedLock(); // offload to separate method
}
}
//释放写锁
//Releases write lock for tree restructuring.
private final void unlockRoot() {
lockState = 0;
}
//根据key获取指定的结点
//Returns matching node or null if none.
//Tries to search using tree comparisons from root, but continues linear search when lock not available.
final Node<K,V> find(int h, Object k) {
...
}
...
}
...
}(3)TreeBin在构造方法中将链表转为红黑树
treeifyBin()方法在对链表进行转化时，会先构建一个双向链表，然后将该双向链表传入TreeBin的构造方法。
TreeBin的构造方法会通过如下处理将该双向链表转化为红黑树：
一.如果红黑树为空，则初始化红黑树的根结点
二.如果红黑树不为空，则按平衡二叉树逻辑插入
三.通过balanceInsertion()方法进行自平衡
TreeBin的构造方法可以分为三部分：
第一部分：初始化红黑树
遍历链表b，将链表b的头结点设置为红黑树的根结点，接着设置红黑树根结点的左右子结点为null，以及设置红黑树根结点为黑色。
第二部分：将链表中的结点添加到初始化好的红黑树
首先计算dir的值。如果dir = -1，表示红黑树中被插入结点的hash值大于新添加结点x的hash值。如果dir = 1，表示红黑树中被插入结点的hash值小于新添加结点x的hash值。然后根据dir的值来决定新添加结点x是被插入结点的左结点还是右结点，最后调用TreeBin的balanceInsertion()方法对红黑树进行自平衡处理。
第三部分：对红黑树进行自平衡
调用TreeBin的balanceInsertion()方法对红黑树进行自平衡处理。
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
static final class TreeBin<K,V> extends Node<K,V> {
TreeNode<K,V> root;//红黑树根结点
volatile TreeNode<K,V> first;//链表头结点，由构造函数传入
volatile Thread waiter;//保存最近一个抢占写锁的线程(如果有值，则说明lockState是读锁状态)
volatile int lockState;//表示锁的状态
//values for lockState
static final int WRITER = 1;//写锁状态
static final int WAITER = 2;//等待获取写锁状态
static final int READER = 4;//读锁状态
...
//构造函数，将以b为头结点的链表转换为红黑树
TreeBin(TreeNode<K,V> b) {
//第一部分开始：初始化红黑树
super(TREEBIN, null, null, null);
this.first = b;
//r表示红黑树的根结点
TreeNode<K,V> r = null;
//遍历链表b，x将作为新添加的红黑树结点
for (TreeNode<K,V> x = b, next; x != null; x = next) {
next = (TreeNode<K,V>)x.next;
//把新添加的红黑树结点x的左右子结点设置为null
x.left = x.right = null;
//r表示红黑树的根结点，r == null表示红黑树为空，将x结点设置为红黑树的根结点
if (r == null) {
x.parent = null;
//把红黑树的根结点设置为黑色
x.red = false;
r = x;
//第一部分结束
} else {
//第二部分开始：将链表中的结点添加到初始化好的红黑树中
//x是新添加的红黑树结点
K k = x.key;
int h = x.hash;
Class<?> kc = null;
//p是红黑树中被插入的结点
for (TreeNode<K,V> p = r;;) {
int dir, ph;
K pk = p.key;
//首先计算dir的值
//dir = -1，表示红黑树中被插入结点的hash值大于新添加结点x的hash值
//dir = 1，表示红黑树中被插入结点的hash值小于新添加结点x的hash值
if ((ph = p.hash) > h) {
dir = -1;
} else if (ph < h) {
dir = 1;
} else if ((kc == null && (kc = comparableClassFor(k)) == null) || (dir = compareComparables(kc, k, pk)) == 0) {
dir = tieBreakOrder(k, pk);
TreeNode<K,V> xp = p;
}
//然后根据dir的值来决定新添加的结点x是左结点还是右结点
if ((p = (dir <= 0) ? p.left : p.right) == null) {
x.parent = xp;
if (dir <= 0) {
xp.left = x;
} else {
xp.right = x;
}
//第二部分结束
//第三部分开始：红黑树进行自平衡
//r代表一棵红黑树，x代表往红黑树r添加的结点
r = balanceInsertion(r, x);
//第三部分结束
break;
}
}
}
}
this.root = r;
assert checkInvariants(root);
}
...
}
...
}(4)TreeBin在插入元素时实现红黑树的自平衡
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {
...
static final class TreeBin<K,V> extends Node<K,V> {
TreeNode<K,V> root;//红黑树根结点
volatile TreeNode<K,V> first;//链表头结点，由构造函数传入
volatile Thread waiter;//保存最近一个抢占写锁的线程(如果有值，则说明lockState是读锁状态)
volatile int lockState;//表示锁的状态
// values for lockState
static final int WRITER = 1;//写锁状态
static final int WAITER = 2;//等待获取写锁状态
static final int READER = 4;//读锁状态
...
//root代表一棵红黑树，x代表往红黑树r添加的结点
static <K,V> TreeNode<K,V> balanceInsertion(TreeNode<K,V> root, TreeNode<K,V> x) {
//所有往红黑树添加的结点默认为红色
x.red = true;
//自旋，xp表示添加结点x的父结点，xpp表示添加结点x的爷结点，xppl表示爷结点的左结点，xppr表示爷结点的右结点
for (TreeNode<K,V> xp, xpp, xppl, xppr;;) {
if ((xp = x.parent) == null) {//此处判断条件表示：x结点的父结点为空
//由于只有根结点的父结点才会为空，所以此时x结点为根结点，于是设置根结点x为黑色
x.red = false;
return x;
} else if (!xp.red || (xpp = xp.parent) == null) {//此处判断条件表示：表示x结点的父结点为黑色，或者x结点的爷结点为空
//那么直接返回红黑树root，不需要处理
return root;
}
//代码执行到这里，说明x结点的父结点为红色
if (xp == (xppl = xpp.left)) {//此处判断条件表示：表示x结点的父结点xp是其爷结点xpp的左子结点xppl
if ((xppr = xpp.right) != null && xppr.red) {//此处判断条件表示：x结点的叔结点存在且为红色
//那么直接修改父结点和叔结点的颜色为黑色
xppr.red = false;
xp.red = false;
xpp.red = true;
x = xpp;
} else {//此处判断条件表示：如果x结点的叔结点不存在，或者叔结点存在且为黑色
if (x == xp.right) {//如果x结点是父结点的右子结点，则按x结点的父结点进行左旋
root = rotateLeft(root, x = xp);//将x结点的父结点赋值给x结点
xpp = (xp = x.parent) == null ? null : xp.parent;
}
if (xp != null) {
xp.red = false;
if (xpp != null) {
xpp.red = true;
root = rotateRight(root, xpp);
}
}
}
} else {//表示x结点的父结点是其爷结点的右子结点
if (xppl != null && xppl.red) {
xppl.red = false;
xp.red = false;
xpp.red = true;
x = xpp;
} else {
if (x == xp.left) {
root = rotateRight(root, x = xp);
xpp = (xp = x.parent) == null ? null : xp.parent;
}
if (xp != null) {
xp.red = false;
if (xpp != null) {
xpp.red = true;
root = rotateLeft(root, xpp);
}
}
}
}
}
}
static <K,V> TreeNode<K,V> rotateLeft(TreeNode<K,V> root, TreeNode<K,V> p) {
TreeNode<K,V> r, pp, rl;
if (p != null && (r = p.right) != null) {
if ((rl = p.right = r.left) != null) {
rl.parent = p;
}
if ((pp = r.parent = p.parent) == null) {
(root = r).red = false;
} else if (pp.left == p) {
pp.left = r;
} else {
pp.right = r;
}
r.left = p;
p.parent = r;
}
return root;
}
static <K,V> TreeNode<K,V> rotateRight(TreeNode<K,V> root, TreeNode<K,V> p) {
TreeNode<K,V> l, pp, lr;
if (p != null && (l = p.left) != null) {
if ((lr = p.left = l.right) != null) {
lr.parent = p;
}
if ((pp = l.parent = p.parent) == null) {
(root = l).red = false;
} else if (pp.right == p) {
pp.right = l;
} else {
pp.left = l;
}
l.right = p;
p.parent = l;
}
return root;
}
...
}
}