ConcurrentHashMap原始碼分析

jdk 1.8的實現已經拋棄了segment分段鎖機制，利用cas+synchronized來保證併發更新的安全，底層採用陣列+鍊錶+紅黑樹的儲存結構。

如果f的hash值為-1，說明當前f是forwardingnode節點，意味有其它執行緒正在擴容，則一起進行擴容操作。

tab =helptransfer(tab, f);
else
nodepred = e;
if ((e = e.next) == null) }}
else if (f instanceof treebin) }}
}if (bincount != 0) }}
addcount(1l, bincount);
return null;
}

1 獲取table中對應索引的元素f。

採用unsafe.getobjectvolatile來獲取，也許有人質疑，直接table[index]不可以麼，為什麼要這麼複雜？

在j**a記憶體模型中，我們已經知道每個執行緒都有乙個工作記憶體，裡面儲存著table的副本，雖然table是volatile修飾的，但不能保證執行緒每次都拿到table中的最新元素，unsafe.getobjectvolatile可以直接獲取指定記憶體的資料，保證了每次拿到資料都是最新的。

2 如果f為null，說明table中這個位置第一次插入元素，利用unsafe.compareandswapobject方法插入node節點。

3 在節點f上進行同步，節點插入之前，再次利用tabat(tab, i) == f判斷，防止被其它執行緒修改。

jdk8中，採用多執行緒擴容。整個擴容過程，通過cas設定sizectl，transferindex等變數協調多個執行緒進行併發擴容。

/**擴容執行緒每次最少要遷移16個hash桶

*/private

static final int min_transfer_stride = 16

;擴容索引，表示已經分配給擴容執行緒的table陣列索引位置。主要用來協調多個執行緒，併發安全地

獲取遷移任務（hash桶）。

1 在擴容之前，transferindex 在陣列的最右邊。此時有乙個執行緒發現已經到達擴容閾值，準備開始擴容。

2 擴容執行緒，在遷移資料之前，首先要將transferindex右移（以cas的方式修改transferindex=transferindex-stride(要遷移hash桶的個數)），獲取遷移任務。每個擴容執行緒都會通過for迴圈+cas的方式設定transferindex，因此可以確保多執行緒擴容的併發安全。

換個角度，我們可以將待遷移的table陣列，看成乙個任務佇列，transferindex看成任務佇列的頭指標。而擴容執行緒，就是這個佇列的消費者。擴容執行緒通過cas設定transferindex索引的過程，就是消費者從任務佇列中獲取任務的過程。為了效能考慮，我們當然不會每次只獲取乙個任務（hash桶），因此concurrenthashmap規定，每次至少要獲取16個遷移任務（遷移16個hash桶，min_transfer_stride = 16）

cas設定transferindex的原始碼如下：

private final void transfer(node tab, node nexttab)  
catch (throwable ex) 
nexttable =nexttab;
transferindex =n;
}int nextn =nexttab.length;
forwardingnode
fwd = new forwardingnode(nexttab);
boolean advance = true
;        boolean finishing = false; //
to ensure sweep before committing nexttab
for (int i = 0, bound = 0
;;) 
else
if(u.compareandswapint
(this
, transferindex, nextindex,
nextbound = (nextindex > stride ?nextindex - stride : 0
))) 
}if (i < 0 || i >= n || i + n >=nextn) 
if (u.compareandswapint(this, sizectl, sc = sizectl, sc - 1
))             }
else
if ((f = tabat(tab, i)) == null
)                advance = castabat(tab, i, null
, fwd);
else
if ((fh = f.hash) ==moved)
advance = true; //
already processed
else
}if (runbit == 0
)                             
else
for (nodep = f; p != lastrun; p =p.next) 
settabat(nexttab, i, ln);//在新的佇列中還在原來的位置
settabat(nexttab, i +n, hn);//2進製的值左移一位，所以為n+1
settabat(tab, i, fwd);//修改原始佇列中的值hash=-1
advance = true
;                        }
else
if(f instanceof treebin) 
else
}ln = (lc <= untreeify_threshold) ?untreeify(lo) :
(hc != 0) ? new treebin(lo) : t;
hn = (hc <= untreeify_threshold) ?untreeify(hi) :
(lc != 0) ? new treebin(hi) : t;
settabat(nexttab, i, ln);
settabat(nexttab, i +n, hn);
settabat(tab, i, fwd);
advance = true
;                        }}}}}}

ConcurrentHashMap原始碼分析

ConcurrentHashMap原始碼分析

ConcurrentHashMap原始碼詳解

concurrentHashMap原始碼分析

ConcurrentHashMap原始碼分析

ConcurrentHashMap原始碼分析

ConcurrentHashMap原始碼詳解

concurrentHashMap原始碼分析

相關推薦