public class ConcurrentHashMap {
/*
hash表初始化或扩容时的一个控制位标识量。
负数代表正在进行初始化或扩容操作
-1代表正在初始化
-N 表示有N-1个线程正在进行扩容操作
正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小
*/
private transient volatile int sizeCtl;
// 以下两个是用来控制扩容的时候 单线程进入的变量
private static int RESIZE_STAMP_BITS = 16;
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
static final int MOVED = -1; // hash值是-1，表示这是一个forwardNode节点
static final int TREEBIN = -2; // hash值是-2 表示这时一个TreeBin节点
// 扩容时候临时用的数组
private transient volatile Node<K,V>[] nextTable;
//ConcurrentHashMap的元素个数＝baseCount＋SUM(counterCells)!!
//元素基础个数，通过CAS更新，当CAS失败则将要加的值加到counterCells数组
private transient volatile long baseCount;
//下一个线程领扩容任务时，分配的hash桶起始索引
private transient volatile int transferIndex;
//用数组来处理当CAS失败时，元素统计提高效率的方案。参见java.util.concurrent.atomic.LongAdder
private transient volatile CounterCell[] counterCells;
// 自旋锁 （锁定通过 CAS） 在调整大小和/或创建 CounterCells 时使用。 在CounterCell类更新value中会使用，功能类似显示锁和内置锁，性能更好
private transient volatile int cellsBusy;
static class Node<K,V> {
volatile V val; // 增加了volatile属性
volatile Node<K,V> next; // 增加了volatile属性
//不允许直接改变value的值
public final V setValue(V value) {
throw new UnsupportedOperationException();
}
// 增加了find方法辅助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))))
return e;
} while ((e = e.next) != null);
}
return null;
}
}
static final class TreeNode<K,V> extends Node<K,V> {
// 自带next属性，方便基于TreeBin的访问
TreeNode(int hash, K key, V val, Node<K,V> next,
TreeNode<K,V> parent) {
super(hash, key, val, next);
this.parent = parent;
}
}
// TreeNode的根节点
static final class TreeBin<K,V> {
...
}
// 连接2个hash桶的Node类，
static final class ForwardingNode<K,V> {
final Node<K,V>[] nextTable; // 指向下一个桶
}
// CAS无锁同步
U = sun.misc.Unsafe.getUnsafe();
// 原子查找tab数组下标i的元素
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方式修改值
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方法设置节点位置的值
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
// 仅仅在第一次插入哈希桶时执行初始化
while ((tab = table) == null || tab.length == 0) {
// sizeCtl小于0表示有其他线程正在进行初始化操作，把线程挂起。
// 对于table的初始化工作，只能有一个线程在进行。
// 这里yield后，别人初始化后while条件也不满足了，就退出initTable方法了
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
// 如果没人在做初始化，自己可以做初始化了
// 做之前CAS设置标志位-1，保证自己初始化的同时，其他人不会初始化
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
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); // 初始化数组后，将sizeCtl的值改为0.75*n
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
// 扩容时转移节点用
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//NCPU为CPU核心数，每个核心均分复制任务，如果均分小于16个
//那么以16为步长分给处理器：例如0-15号给处理器1，16-32号分给处理器2。处理器3就不用接任务了。
// 确定步长，多线程复制过程中防止出现混乱。每个线程分配步长长度的hash桶长度。最低不少于16。
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
// 构造新的数组，容量是原来的2倍
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 = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
// 创建一个Forwarding节点，指向扩容后的新数组
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 保证并发扩容安全，如果等于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循环体的作用就是在控制i-- 通过i--可以依次遍历原hash表中的节点
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;
//如果所有的节点都已经完成复制工作 就把nextTable赋值给table 清空临时对象nextTable
if (finishing) {
nextTable = null;
table = nextTab;
//扩容阈值设置为原来容量的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; // recheck before commit
}
}
// 如果值为null，说明一个桶已经结束了，把下一个桶放到forwarding节点里
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// 遇到forwarding节点，说明别的线程在处理了
// 设置advance=true，保证while继续往下遍历
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
//这保证了不会出现该桶正在resize又执行put操作的情况
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
int runBit = fh & n;
//以下的部分在完成的工作是构造两个链表 一个是原链表 另一个是原链表的反序排列
//锁定桶节点，执行复制操作。在复制到nexttab的过程中，未破坏原tab的链表顺序和结构，所以不影响原tab的检索。
Node<K,V> lastRun = f;
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
//这里尽量少的复制链表节点，从lastrun到链尾的这段链表段，无需复制节点，直接复用
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);
}
//在nextTable的i位置上插入一个链表
setTabAt(nextTab, i, ln);
//在nextTable的i+n的位置上插入另一个链表
setTabAt(nextTab, i + n, hn);
//在table的i位置上插入forwardNode节点 表示已经处理过该节点
setTabAt(tab, i, fwd);
//设置advance为true 返回到上面的while循环中 就可以执行i--操作
advance = true;
}
//对TreeBin对象进行处理 与上面的过程类似
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;
}
}
//如果扩容后已经不再需要tree的结构 反向转换为链表结构
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;
}
}
}
}
}
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
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;
//如果table为空的话，初始化table
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//根据hash值计算出在table里面的位置
//如果这个位置没有值 ，直接放进去，不需要加锁
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//这里使用了CAS，避免使用锁。如果CAS失败，说明该节点已经发生改变，
//可能被其他线程插入了，那么继续执行死循环，在链尾插入。
//(只有插入成功返回true，才会break）
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//当遇到表连接点时，帮助一起转移
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//这个地方设计非常的巧妙，内置锁synchronized锁住了f,因为f是指定特定的tab[i]的，
// 所以就锁住了整行链表,这个设计跟分段锁有异曲同工之妙，只是其他读取操作需要用cas来保证
synchronized (f) {
if (tabAt(tab, i) == f) {
//fh〉0 说明这个节点是一个链表的节点 不是树的节点
if (fh >= 0) {
binCount = 1;
//在这里遍历链表所有的结点
for (Node<K,V> e = f;; ++binCount) {
K ek;
//如果hash值和key值相同 则修改对应结点的value值
// 这里因为在synchronize里面，所以不需要cas
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;
//如果遍历到了最后一个结点，那么就证明新的节点需要插入 就把它插入在链表尾部
// 同理，也不用CAS
if ((e = e.next) == null) {
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) {
//如果链表长度已经达到临界值8 就需要把链表转换为树结构
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
}
网友评论