記一個(gè) ConcurrentHashMap 使用不當(dāng)導(dǎo)致的并發(fā)事故
一、引言
我們都知道ConcurrentHashMap可以保證鍵值對(duì)并發(fā)插入安全，因?yàn)槠鋕ey值唯一性的原因，所以hutool對(duì)其進(jìn)行了進(jìn)一步的封裝實(shí)現(xiàn)了一個(gè)ConcurrentHashSet，代碼如下，即判斷put后是否返回null，若是null則說(shuō)明是第一次插入，反之就是存在重復(fù)元素，返回已存在的元素值。從而保證并發(fā)插入元素線程安全且唯一。
//hutool的ConcurrentHashSet通過(guò)判斷返回null得知之前是否插入過(guò)重復(fù)元素
@Override
public boolean add(E e) {
return map.put(e, PRESENT) == null;
}但是如果對(duì)于這些映射容器的鍵使用不當(dāng)就可能導(dǎo)致唯一鍵值對(duì)多次插入的情況，所以本文將基于筆者前段時(shí)間遇到的經(jīng)典的例子為切入點(diǎn)，深入剖析該問(wèn)題的原因和解決思路。
二、詳解ConcurrentHashMap并發(fā)重復(fù)插入問(wèn)題
1. 需求說(shuō)明
我們現(xiàn)在有這樣一個(gè)需求，大體是通過(guò)數(shù)據(jù)庫(kù)獲取要處理的任務(wù)并按照如下步驟執(zhí)行：
- 從數(shù)據(jù)庫(kù)讀取未完成(status為0)的任務(wù)，將其采用并發(fā)容器(ConcurrentHashSet)存放，key為這個(gè)任務(wù)對(duì)象
- 工作線程處理，并在內(nèi)存中將其設(shè)置為1
- 定時(shí)任務(wù)線程從容器中讀取這些任務(wù)并移除
- 將已完成任務(wù)狀態(tài)寫(xiě)回庫(kù)中
2. 落地代碼
對(duì)應(yīng)任務(wù)表的實(shí)體類封裝如下，我們的加載到ConcurrentHashSet會(huì)被多個(gè)線程并發(fā)的調(diào)度處理，處理過(guò)程中會(huì)并發(fā)更新?tīng)顟B(tài)。
@Data
publicclass Task {
privateint id;
/**
* 任務(wù)名稱
*/
private String taskName;
/**
* 0.未開(kāi)始
* 1.進(jìn)行中
* 2.已完成
*/
privateint status;
}對(duì)應(yīng)的實(shí)現(xiàn)代碼如下，可以看到從數(shù)據(jù)庫(kù)讀取未開(kāi)始的任務(wù)，線程1將其更新為處理完成后更新為處理中，線程2處理完成后更新為已完成：
public static void main(String[] args) throws InterruptedException {
ConcurrentHashSet<Task> set = new ConcurrentHashSet<>();
CountDownLatch countDownLatch = new CountDownLatch(2);
//假設(shè)從數(shù)據(jù)庫(kù)讀取一個(gè)task
Task task = new Task();
task.setId(1);
task.setTaskName("任務(wù)1");
task.setStatus(0);
set.add(task);
//模擬多線程并發(fā)更新
//線程1更新為處理中
new Thread(() -> {
log.info("線程1處理中....");
task.setStatus(1);
set.add(task);
countDownLatch.countDown();
}, "t1").start();
//線程2更新為已完成
new Thread(() -> {
log.info("線程2處理中....");
task.setStatus(2);
set.add(task);
countDownLatch.countDown();
}, "t2").start();
countDownLatch.await();
log.info("set size:{}", set.size());
}輸出結(jié)果如下，可以看到明明同一個(gè)對(duì)象，結(jié)果插入了3次：
00:44:32.637 [main] INFO com.sharkChili.webTemplate.Main - set size:3調(diào)試查看set內(nèi)部，3個(gè)元素都指向我們的唯一的任務(wù)-1。
3. 事故原因
我們都知道JDK8版本無(wú)論是HashMap還是ConcurrentHashMap底層采用數(shù)組+鏈表/紅黑樹(shù)，元素進(jìn)行插入前都需要進(jìn)行hash運(yùn)算定位數(shù)組索引，然后使用equal和hashCode比較的過(guò)程元素是否存在。 很明顯，我們上文并發(fā)操作元素時(shí)修改了status字典，導(dǎo)致每次得出的hashCode結(jié)果值改變了，進(jìn)而導(dǎo)致同一個(gè)元素因?yàn)椴煌膆ashCode插入到不同的位置，出現(xiàn)去重失?。?/p>
對(duì)應(yīng)筆者也給出ConcurrentHashMap的put方法底層實(shí)現(xiàn)：
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) thrownew NullPointerException();
//計(jì)算key的hash值，因?yàn)槲覀儎?dòng)態(tài)修改了status導(dǎo)致hash值不同
int hash = spread(key.hashCode());
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();
//因?yàn)閔ash值不同每次定位到的i位置不同，最終存到不同的位置
elseif ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
}
.....
}4. 解決方案
很明顯出現(xiàn)這個(gè)問(wèn)題的原因就是因?yàn)椴l(fā)操作修改的status影響了hashcode計(jì)算結(jié)果，進(jìn)而導(dǎo)致并發(fā)操作變得無(wú)效，因?yàn)閕d是全局唯一的，所以直接重寫(xiě)hashCode和equals方法，讓Task對(duì)象的計(jì)算和比對(duì)都通過(guò)id進(jìn)行：
@Data
publicclass Task {
//......略
@Override
public boolean equals(Object o) {
if (this == o) returntrue;
if (o == null || getClass() != o.getClass()) returnfalse;
Task task = (Task) o;
return id == task.id;
}
@Override
public int hashCode() {
return Objects.hash(id);
}
}三、小結(jié)
總的來(lái)說(shuō)，對(duì)于這類涉及并發(fā)操作的重構(gòu)，建議梳理清晰的數(shù)據(jù)流向并結(jié)合源碼工作流程加以推斷分析，最終明確問(wèn)題風(fēng)險(xiǎn)點(diǎn)直接進(jìn)行邏輯修復(fù)并及時(shí)提測(cè)。
