基本属性
private static final int MAXIMUM_CAPACITY = 1 << 30; //最大容量
private static final int DEFAULT_CAPACITY = 16; //默认容量
private static final int DEFAULT_CONCURRENCY_LEVEL = 16; //默认并发级别
private static final float LOAD_FACTOR = 0.75f; //默认负载因子
static final int TREEIFY_THRESHOLD = 8; //树化阈值
static final int UNTREEIFY_THRESHOLD = 6; //转为链表最小阈值
static final int MIN_TREEIFY_CAPACITY = 64; //最小树化容量
构造函数
public ConcurrentHashMap() {
}
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0) //非法数据
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? //initialCapacity >=1/2 * MAXIMUM_CAPACITY 直接给值MAXIMUM_CAPACITY
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
this.sizeCtl = cap;
}
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this.sizeCtl = DEFAULT_CAPACITY;
putAll(m);
}
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1); // 并发等级设置为1
}
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
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);
this.sizeCtl = cap;
}
put（）方法
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode()); //和hashmap类似
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable(); //初始化
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //获取i位置的对象 线程B准备访问
if (casTabAt(tab, i, null, //线程A正在访问 自旋锁 A的casTabAt操作具有volatile写相同内存语义 A执行该操作后 B就可以直接通过tabAt方法立刻看到table[i]中的变化
new Node<K,V>(hash, key, value, null))) // cas失败 最可能就是其他进程将其赋值成功导致cas失败
break; // no lock when adding to empty bin 此处控制逻辑为：线程一执行casTabAt(tab,i,null,node1),此时tab[i]等于预期值null,因此会插入node1。随后线程二执行casTabAt(tba,i,null,node2),此时tab[i]不等于预期值null，插入失败。然后线程二会回到for循环开始处，重新获取tab[i]作为预期值，重复上述逻辑。
} //以上for循环+CAS操作就是无锁算法（unlook free）的经典实现
else if ((fh = f.hash) == MOVED) //moved==-1 表示当前map在进行扩容
tab = helpTransfer(tab, f); //调用helpTransfer来协助扩容 而后再更新值
else { //hash冲突 当前位置有结点 判断是什么类型的结点 是红黑树还是链表
V oldVal = null;
synchronized (f) { //f为对应bucket中的对象 hash桶不为空 对tab[i]的头节点进行加锁
if (tabAt(tab, i) == f) { //加锁再检查
if (fh >= 0) {
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;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) { //节点不存在，尾插法进链表
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
} //红黑树根节点
else if (f instanceof TreeBin) { //1.8的hashmap是TreeNode节点
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)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount); //put进去一个元素之后 进行+1操作，相当于siz++ 作用：1、对 table 的长度加一。无论是通过修改 baseCount，还是通过使用 CounterCell。当 CounterCell 被初始化了，就优先使用他，不再使用 baseCount。
return null; // 2、检查是否需要扩容，或者是否正在扩容。如果需要扩容，就调用扩容方法，如果正在扩容，就帮助其扩容。
}
代码逻辑流程分析
- 判断tab是否需要初始化（关键点 sizeCtl的值），使用 initTable（）进行初始化， 若不为null 则进入 第2步
-
sizeCtl默认为0 多个线程可以进行竞争对数组进行初始化 -
sizeCtl== -1时 表示有线程正在进行初始化 其他线程执行Thread.yield() 将时间片结束 放弃这次cpu的竞争 -
sizeCtl== n - (n >>> 2) 为其阈值 此时阈值初始化赋值完成
-
- 判断对应索引位置是否为空 若为null 则采用cas方式 创建一个新对象 若不为null 则进入 第3步
- 加锁 ，加锁再检查，判断结点类型（链表还是红黑树）采用对应的采用对应put加入结构中
- 判断
binCount的值，是否需要进行树化 - 调用
addCount()方法进行数量统计，检查是否需要进行扩容，若需要进行扩容，进行实际扩容;
initTable() 初始化方法
private final Node<K,V>[] initTable() { //实例化ConcurrentHashMap时倘若声明了table的容量，初始化时会根据参数调整table的大小 确保始终是2的幂次方
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0) //sizeCtl是直接从内存中获取的值 如果为-1 则表示有线程CAS成功了 将当前线程执行 Thread.yield()方法 当前线程让出时间片 由执行状态转换为就绪状态
Thread.yield(); // lost initialization race; just spin 先放弃这一次的cpu 然后再次和其他线程竞争cpu
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { //将SIZECTL和sc比较 相等则将索引为SIZECTL的位置中的值赋值为-1 相当于加锁 只有一个线程能改成-1
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY; //16
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2); //0.75n n*(1-1/4)
}
} finally {
sizeCtl = sc; //此时sc相当于阈值
}
break;
}
}
return tab;
}
addCount()分析
1.保证要加入map的容量能够
++1。
2.检查是否需要进行扩容
流程分析
1、判断counterCells是否需要进行初始化；若counterCells不为null，使用cas的方式将baseCount+1。
1）若baseCount竞争失败，使用随机值ThreadLocalRandom.getProbe() & m计算CounterCell中对应索引位置，使用cas对cell里面的value进行++1
2）value也竞争失败，此时uncontended==false则调用fullAddCount（）保证+1操作一定能实现。
2、检查是否需要扩容
3、当前容量大于阈值，table不为null，且小于最大容量则进行调用transfer()方法扩容、转移。
private final void addCount(long x, int check) { //x == -1 remove方法
CounterCell[] as; long b, s;
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { //用cas的方法去修改baseCount
CounterCell a; long v; int m; //使用cas竞争baseCount失败（使用cas竞争后续的value+1） 就进入if 或者as不空 as计数盒子为空则说明此时还未有并发发生
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 || //计数盒子as为空 或者 长度小于2 或者 索引对应节点为空
(a = as[ThreadLocalRandom.getProbe() & m]) == null || //ThreadLocalRandom.getProbe() & m 当前CountCell数组里面对应的下标
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { //uncontended 表示竞争value是否成功，true flase
fullAddCount(x, uncontended); //初始化CountCell 数组 多线程修改AddCount时，竞争失败的线程会执行fullAddCount，把x的值插入到uncontended中
return;
}
if (check <= 1) //check = binCount 为链表长度
return;
s = sumCount(); //统计现有节点个数 s == size ==baseCount+all cells
} //检查是否需要扩容
if (check >= 0) { //实际扩容 binCount的值最小也是0 所以每次都必会检查 除非进行了覆盖操作
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null && //sc此时为阈值 容量达到阈值了且table表不为空 且 table长度小于最大长度（可以扩容）
(n = tab.length) < MAXIMUM_CAPACITY) { //进行第一次扩容之后 可能立刻进行第二次扩容
int rs = resizeStamp(n); //根据数组length得到一个标识 rs的值只和n值有关 其他运算部分都是固定的 不同n得到的值不同
if (sc < 0) {//如果正在进行扩容 sc=-1或者-N
// 如果 sc 的低 16 位不等于 标识符（校验异常 sizeCtl 变化 了）
// 如果 sc == 标识符 + 1 （扩容结束了，不再有线程进行扩容）（默认第一个线程设置 sc ==rs 左移 16 位 + 2，当第一个线程结束扩容了，就会将 sc 减一。这个时候，sc 就等于 rs + 1）
// 如果 sc == 标识符 + 65535（帮助线程数已经达到最大）
// 如果 nextTable == null（结束扩容了）
// 如果 transferIndex <= 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)) // 如果可以帮助扩容，那么将 sc 加 1. 表示多了一个线程在帮助扩容
transfer(tab, nt); //将旧数组中的元素 转移到新数组中
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2)) //如果不在扩容，将 sc 更新：标识符左移 16 位 然后 + 2. 也就是变成一个负数。高 16 位是标识符，低 16 位初始是 2
transfer(tab, null); //更新后sc值变成负数，开始扩容
s = sumCount(); //对新数组进行容量计算
}
}
}
fullAddCount（）
保证一定能够对容量进行+1。
对CounterCell数组进行初始化或者扩容。
流程分析
1、使用cellsBusy做为标志位，表示当前cell是否有线程正在使用。
2、计算对应索引位置是否为null，若当前没有元素，则对使用cas方式改变cellsBusy 的值，将当前位置放入新元素，结束将cellsBusy 改变回0，否则执行其他的条件语句。
3、再次尝试对value进行竞争，若成功则跳出循环
4、若第一次hash找到对应位置时 cas失败，进行第二次循环，得到新的hash值，若第二次循环 cas value仍然失败 则可认为当前容量太小，线程竞争激烈，counterCells 扩容为原来2倍。
5、再次竞争baseCount，若竞争仍然失败继续进行循环。
总结来说，就是类似于让线程反复寻找萝卜坑。
private final void fullAddCount(long x, boolean wasUncontended) { // 竞争失败 wasUncontended == false
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;
if ((as = counterCells) != null && (n = as.length) > 0) {
if ((a = as[(n - 1) & h]) == null) { //计算索引位置是否为null 初始counterCells容量为2 里面含有一个结点不为空 值此时为0 或者 1
if (cellsBusy == 0) { // Try to attach new Cell cellsBusy判断其他功能是否正在被使用
CounterCell r = new CounterCell(x); // Optimistic create
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) {
rs[j] = r; //若此时对应索引位置仍然为null 则加入到cell数组中
created = true;
}
} finally {
cellsBusy = 0;
}
if (created)
break;
continue; // Slot is now non-empty
}
}
collide = false;
}//else语句 只有该线程所对应的位置有一个CountCell对象才会进入
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash 其逻辑为：wasUncontended先是true 然后改成false 然后重新算一个新的hash值
else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
break;
else if (counterCells != as || n >= NCPU) //限制扩容的条件：数组已经改变||数组长度已经大于cpu核心的数量
collide = false; // At max size or stale
else if (!collide) //要使collide == true 和两次循环有关 第一次false 第二次是true
collide = true; //下面的else if 要执行必须要collide ==false 实际上是用来控制下面的条件分支
else if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
try { //数组countCells[]扩容的时机: 两次对cell cas都失败则进行扩容
if (counterCells == as) {// Expand table unless stale 第一次hash找到对应位置时 cas失败，进行第二次循环，得到新的hash值，若第二次循环 cas value仍然失败 则可认为当前容量太小，线程竞争激烈，扩容为原来2倍
CounterCell[] rs = new CounterCell[n << 1]; //相当于collide == true 扩容为2倍 转移元素
for (int i = 0; i < n; ++i) //扩容本质是 因为太多次cas都不能成功 则进行扩容为2倍 用空间换时间以提高效率
rs[i] = as[i];
counterCells = rs;
}
} finally {
cellsBusy = 0;
}
collide = false; //扩容完成将其变回false
continue; // Retry with expanded table
}
h = ThreadLocalRandom.advanceProbe(h); //生成一个新的hash值
}
else if (cellsBusy == 0 && counterCells == as &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { //cellsBusy == 0 表示没有线程在使用
boolean init = false;
try { // Initialize table
if (counterCells == as) { //若其他线程已经初始化好counterCells 则as值改变 则为false
CounterCell[] rs = new CounterCell[2]; //初始化生成的数组 容量为2
rs[h & 1] = new CounterCell(x); //将CounterCell里面的value改成1
counterCells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break; //初始化成功 就直接退出当前循环
}
else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x)) //若其他竞争线程竞争counterCells失败，则再次竞争BASECOUNT
break; // Fall back on using base
}
}
sumCount()
final long sumCount() { //size ==baseCount+all cells
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; //对baseCount和as.value求和计算容器中总共有多个元素
}
}
return sum;
}
transfer()分析
1、进行实际的扩容。
2、进行元素的转移：
1）链表类型:采用lastRun的方式（类似于蜘蛛纸牌）
2）红黑树：采用高低位链表的方式（和hashmap相同）
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 = MIN_TRANSFER_STRIDE; // subdivide range 最小步长为：16
if (nextTab == null) { // initiating
try {
@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 = nextTab;
transferIndex = n; //transferIndex等于就数组的长度
}
int nextn = nextTab.length;
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
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;
if (--i >= bound || finishing) //表示还在该范围内 不用向前走了
advance = false;
else if ((nextIndex = transferIndex) <= 0) { //nextIndex是下一个步长中的第一个元素
i = -1; //transferIndex 实际为要转移区域的最右边的下一个下标 如果是下标是3 transferIndex刚好就是3+1 = 4
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound; // bound和i表示当前要转移的元素的索引范围 [bound,i] 不断进行i--
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
nextTable = null;
table = nextTab; //只有完成了才能获取新的table
sizeCtl = (n << 1) - (n >>> 1);
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) // 当前相等则表示等于初始sizeCtl值了 表示没有线程在扩容了
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd); //将fwd赋给i advance==true
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
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);
}
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;
}
}
}
}
}
}
