本篇文章接上篇文章，介紹一些ConcurrentHashMap的細節，雖然不是特別重要，但也是干貨滿滿，
先列一個引數：
//0：表示counterCells無人使用
//因為counterCells是一個陣列，執行緒不安全的
//所以在操作counterCells之前，都要用CAS將cellsBusy改成1，表示有人在使用該陣列
private transient volatile int cellsBusy;
方法串列
- spread()：
- addCount的第一部分：
- 判斷是否有失敗的風險
- 判斷是否可以投機
- fullAddCount()：
- counterCells != null，找到的下標 == n
- 找到的下標 != null
- counterCells == null
- ThreadLocalRandom.getProbe()：
- ThreadLocalRandom.localInit()：
- ThreadLocalRandom.advanceProbe()：
- 面試題：concurrentHashMap的key和value為啥不能為null
spread()：
spread(int h)：
//跟hashmap有點區別，不僅高16參與了計算，并且還再次 & 計算
return (h ^ (h >>> 16)) & HASH_BITS;
前面運算這里不解釋，跟hashMap一樣，主要看后面的& HASH_BITS
HASH_BITS是0x7fffffff，換算下來是2147483649，int的最大值
作用是消除負數，這里順便看下負數的二進制
舉個例子：
-5的原碼是：10000000000000000000000000000101
反碼是：11111111111111111111111111111010
補碼是：11111111111111111111111111111011
負數的原碼：是對應的正數的二進制，但是符號位是1，
符號位是區分正負數用的，1是負數，0是整數(符號位就是最高位的意思，左邊第一位)
負數的反碼：是除符號位，其余的取反
負數的補碼：是反碼+1
HASH_BITS換算下來的二進制是：11111111111111111111111111111111（注意是31位，雖然Java中的int是32位，但是一般只有負數才會顯示32位，正數默認31位）
hash & HASH_BITS，不管hash是正還是負，結果都會轉成正數，因為HASH_BITS的符號位是0，& 下來的最高位肯定是0，也就是正數，
spread(int h) 的作用其實就是避免hash值是負數，大概是因為ConcurrentHashMap內置了MOVED、TREEBIN、RESERVED這3個hash，是負數，為了避免沖突吧，
addCount的第一部分：
//判斷是否有失敗的風險
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true;
//判斷是否可以投機
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
//死回圈把x記錄到counterCells，uncontended是判斷有沒有嘗試用CAS更新過
fullAddCount(x, uncontended);
return;
}
//到這表示CAS把值記到counterCells肯定成功了，因為上面的是 ||，CAS是最后一個判斷條件
//check == 0：表示插入的下標是空的，此時是頭節點，此坑位還有不少直接return
//check == 1：表示插入的是鏈表的第2個，此坑位還有不少，就不考慮擴容，直接return
//check == -1：表示的是洗掉，直接return
//這里跟hashMap有點不一樣，沒有直接判斷大于多少就擴容
if (check <= 1)
return;
//把counterCells的值累加到baseCount
s = sumCount();
}
先介紹一下counterCells，每次是CAS去增加baseCount的，這樣并發的話肯定會有執行緒更新失敗，這時候ConcurrentHashMap并沒有重試，而是把失敗的值記錄下來，就是記錄到counterCells里面，
這里可以分為2步：
- 判斷是否有失敗的風險
- 判斷是否可以
投機（開玩笑）
判斷是否有失敗的風險
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
(as = counterCells) != null：
很簡單，counterCells初始為null，CAS增加baseCount失敗就會把值記錄到counterCells里
counterCells != null說明出現并發修改baseCount，有執行緒失敗了
已經有執行緒失敗了，那么再CAS新增很有可能失敗，所有這里 != null 就沒有嘗試CAS，直接走下面的邏輯
為null才走CAS，盡可能的避免競爭
判斷是否可以投機
//進到這里，表示很有可能多執行緒在修改baseCount
//ThreadLocalRandom.getProbe()可以簡單的先理解成取當前執行緒的hash，后面再詳細分析
//非常Nice的判斷
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
//死回圈把x記錄到counterCells，uncontended是判斷有沒有嘗試用CAS更新過
fullAddCount(x, uncontended);
return;
}
as == null || (m = as.length - 1) < 0：
這個判斷很簡單，判斷counterCells是否有值，但是含義卻不簡單
因為上面已經判斷過 != null了，這里再判斷一下，為了減少競爭，繼續看下面的判斷
(a = as[ThreadLocalRandom.getProbe() & m]) == null：
ThreadLocalRandom.getProbe()最后說，先簡單理解成執行緒的hash
這里判斷的是根據當前執行緒推算出要插入的counterCells的下標是否為null
走到這里，counterCells!= null，再判斷插入的下標是否= null
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)：
因為if里面是 || 判斷，走到這，說明counterCells != null，要插入的counterCells的下標 =null
實在沒辦法了，必須得競爭了，此時才CAS，把值更新到counterCells 里面
回憶一下，整個判斷，是這樣的：
- counterCells == null，CAS更新baseCount
- counterCells != null（或者CAS失敗），再判斷一遍counterCells != null
- counterCells != null，再判斷要插入的下標!= null
- 要插入的下標!= null，才CAS把值更新到counterCells
可以看到，盡量的保證進入到fullAddCount這個死回圈的時候，沒有競爭，
實在避免不了，再進死回圈之前，再嘗試一遍CAS，非常nice，Doug Lea牛逼，
fullAddCount()：
繼續fullAddCount：
由于較為復雜，分成3部分看：
- counterCells != null，找到的下標 == null
- 找到的下標 != null
- counterCells == null
counterCells != null，找到的下標 == n
private final void fullAddCount(long x, boolean wasUncontended) {
int h;
//判斷當前執行緒的PROBE是否初始化過
if ((h = ThreadLocalRandom.getProbe()) == 0) {
//為0就先初始化
ThreadLocalRandom.localInit();
//獲取值
h = ThreadLocalRandom.getProbe();
//把標志位置為true，確保為true，也就是確保就是沒有CAS過
wasUncontended = true;
}
boolean collide = false; // True if last slot nonempty
for (;;) {
CounterCell[] as; CounterCell a; int n; long v;
//先判斷counterCells是否初始化過
if ((as = counterCells) != null && (n = as.length) > 0) {
//h是執行緒的隨機值，(n - 1) & h就是計算下標
if ((a = as[(n - 1) & h]) == null) {
//cellsBusy為0，表示沒執行緒在初始化或者擴容counterCells
if (cellsBusy == 0) {
//生成要插入的物件
CounterCell r = new CounterCell(x);
//把cellsBusy改成1，表示有執行緒再用
if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
boolean created = false;
try {
CounterCell[] rs; int m, j;
//把cellsBusy改成1，再判斷一下
//因為改成1之后，counterCells就鎖定了，此時的判斷就是準的
if ((rs = counterCells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
//賦值
rs[j] = r;
created = true;
}
} finally {
//最終將cellsBusy置為0
cellsBusy = 0;
}
if (created)
break;
continue;
}
}
collide = false;
}
總的來說，還是比較簡單的，就是一個普通陣列，用cellsBusy來保證執行緒安全，
不過有一個判斷比較巧妙，需要單獨說一下：
if ((h = ThreadLocalRandom.getProbe()) == 0) {
//記住這個變數
wasUncontended = true;
這里不僅僅是判斷PROBE是否初始化過，還有一個意義wasUncontended ：
其實這里判斷的是進方法之前，有沒有CAS過
== 0說明沒有初始化，說明進fullAddCount是因為counterCells為null
!= 0說明初始化過了，說明進fullAddCount之前已經走過getProbe()了，也就是很有可能CAS過
wasUncontended == true：說明沒有CAS
wasUncontended == false：說明CAS失敗了
找到的下標 != null
//走到這，說明計算的下標不為null
else if (!wasUncontended)
//進到里面，說明之前在addCount已經CAS過了，并且失敗了
//第一次只把標志位改為true，改個狀態出去，給個機會，第二次再來CAS，避免競爭
wasUncontended = true;
//走到這，2種可能
//1：是死回圈的第二次了
//2：是之前沒有CAS過
//不管哪種情況，必須要CAS了，要么是第一次，要么已經讓過一次了，只能競爭了，
else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
//成功就退出了
break;
//判斷一下counterCells是否已經被改掉了或者說陣列長度已經 >= CPU數量了
//counterCells 最大是NCPU
else if (counterCells != as || n >= NCPU)
collide = false;
else if (!collide)
//進到里面，說明需要擴容了
//改個狀態出去，給個機會
collide = true;
//走到這，沒辦法了，準備擴容吧
//嘗試CAS標志cellsBusy
else if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
try {
//雙重校驗，一個意思
if (counterCells == as) {
//擴容2倍
CounterCell[] rs = new CounterCell[n << 1];
//把舊值復制到新陣列
for (int i = 0; i < n; ++i)
rs[i] = as[i];
counterCells = rs;
}
} finally {
//最終將cellsBusy置為0
cellsBusy = 0;
}
//改為false，下次還要擴容又要2次判斷
collide = false;
//跳出，繼續嘗試把值新增到陣列
continue;
}
//每次不成功就算下新的hash，換個下標試試
h = ThreadLocalRandom.advanceProbe(h);
}
}
}
同樣不是很難，只是有一點確實值得借鑒，concurrentHashMap真的是極力避免競爭，上面說的給個機會，其實相當于空輪詢，競爭失敗之后，基本上都要空輪詢一次，第2次才會繼續競爭，
counterCells == null
//走到這，說明counterCells為null
//就嘗試把CELLSBUSY改完1
else if (cellsBusy == 0 && counterCells == as &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
boolean init = false;
try { // Initialize table
//雙重校驗，一個意思
if (counterCells == as) {
//初始化長度為2
CounterCell[] rs = new CounterCell[2];
//h是執行緒的隨機值，h & 1就是計算下標
rs[h & 1] = new CounterCell(x);
counterCells = rs;
//設定初始化狀態init為true
init = true;
}
} finally {
//最終將cellsBusy置為0
cellsBusy = 0;
}
if (init)
//初始化好了，且新增好了就跳出
break;
}
//走到這，說明已經有執行緒在初始化counterCells，就再嘗試一下baseCount + 1，不行就死回圈
else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
break;
這也不難，就是普通 陣列，執行緒安全的初始化，
拆開來，貌似都不難，但是值得借鑒的確實多，
ThreadLocalRandom.getProbe()：
UNSAFE.getInt(Thread.currentThread(), PROBE)
很簡單，就是獲取當前執行緒的某個變數值
看下PROBE：
private static final sun.misc.Unsafe UNSAFE;
private static final long PROBE;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> tk = Thread.class;
(tk.getDeclaredField("threadLocalRandomProbe"));
} catch (Exception e) {
throw new Error(e);
}
}
是獲取的threadLocalRandomProbe的初始地址的偏移量，簡單理解成獲取值，
看下threadLocalRandomProbe：
@sun.misc.Contended("tlr")
int threadLocalRandomProbe;
就是一個變數，那可以肯定，第一次ThreadLocalRandom.getProbe()肯定為0
ThreadLocalRandom.localInit()：
private static final int PROBE_INCREMENT = 0x9e3779b9;
private static final AtomicInteger probeGenerator = new AtomicInteger();
static final void localInit() {
//很簡單，獲取一個值0x9e3779b9，-1640531527
int p = probeGenerator.addAndGet(PROBE_INCREMENT);
int probe = (p == 0) ? 1 : p;
//賦值
UNSAFE.putInt(t, PROBE, probe);
}
說明一下：
threadLocalRandomProbe是Thread的變數
probeGenerator 、PROBE_INCREMENT是ThreadLocalRandom的，每調一次localInit，probeGenerator的值也在變，
看完整個PROBE的程序，理解成執行緒的hash是沒啥問題的，
ThreadLocalRandom.advanceProbe()：
static final int advanceProbe(int probe) {
probe ^= probe << 13; // xorshift
probe ^= probe >>> 17;
probe ^= probe << 5;
UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
return probe;
}
一通計算，PROBE肯定變了，每次換個值，也就是換個下標，
面試題：concurrentHashMap的key和value為啥不能為null
先來回憶一下hashMap:
static final int hash(Object key) {
int h;
return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}
concurrentHashMap：
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
可以看到，hashMap是允許的，concurrentHashMap不允許，
先說value不允許為null：
concurrentHashMap定位的是并發場景，如果允許value為null，那么就會造成一種情況，
A執行緒：get(key) 為null，這是A是不知道key的value為null，還是key不存在
如果是hashMap，可以用contains來判斷，而concurrentHashMap不行，
B執行緒：在Aget(key) 之后put(key,null)
此時Acontains就是true，那么就以為key當時是有值的，不知道是B中途插入的，
而hashMap定位的是單執行緒，contains為true，就表示為key當時是有值的，不考慮B中途插入的情況，
有人會說，別的value不為null，也會發生這種情況，
沒錯，但是get(key) 為null，contains為true，A能知道有人中途插入了，
先說key不允許為null：
跟value不允許為null：同樣的原因，containsKey的時候有二義性
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/233535.html
標籤：其他