文章已同步發表于微信公眾號JasonGaoH，ConcurrentHashMap是如何保證線程安全的

之前分析過HashMap的一些實現細節，關于HashMap你需要知道的一些細節, 今天我們從源碼角度來看看ConcurrentHashMap是如何實現線程安全的，其實網上這類文章分析特別多，秉著”紙上得來終覺淺，絕知此事要躬行“的原則，我們嘗試自己去分析下，希望這樣對于ConcurrentHashMap有一個更深刻的理解。

為什么說HashMap線程不安全，而ConcurrentHashMap就線程安全

其實ConcurrentHashMap在Android開發中使用的場景并不多，但是ConcurrentHashMap為了支持多線程并發這些優秀的設計卻是最值得我們學習的地方，往往”ConcurrentHashMap是如何實現線程安全“這類問題卻是面試官比較喜歡問的問題。

首先，我們嘗試用代碼模擬下HashMap在多線程場景下會不安全，如果把這個場景替換成ConcurrentHashMap會不會有問題。

因為不同于其他的線程同步問題，想模擬出一種場景來表明HashMap是線程不安全的稍微有點麻煩，可能是hash散列有關，在數據量較小的情況下，計算出來的hashCode是不太容易產生碰撞的，網上很多文章都是嘗試從源碼角度來分析HashMap可能會導致的線程安全問題。

我們來看下下面這段代碼,我們構造10個線程，每個線程分別往map中put 1000個數據，為了保證每個數據的key不一樣，我們將i+ 線程名字來作為map 的key，這樣，如果所有的線程都累加完的話，我們預期的map的size應該是10 * 1000 = 10000。

import?java.util.HashMap;import?java.util.Map;import?java.util.concurrent.ConcurrentHashMap;public?class?HashMapTest?{?public?static?void?main(String[]?args)?{??Map?map?=?new?HashMap();??//?????Map?map?=?new?ConcurrentHashMap();??for?(int?i?=?0;?i??1)???Thread.yield();????System.out.println(map.size());?}}class?MyThread?extends?Thread?{????public?Map?map;????public?String?name;????public?MyThread(Map?map,?String?name)?{??????this.map?=?map;??????this.name?=?name;????}????public?void?run()?{?????for(int?i?=0;i<1000;i++)?{??????map.put(i?+?name,?i?+?name);?????}????}??}

使用HashMap，程序運行，結果如下：

9930

那我們如果把這里的HashMap換成ConcurrentHashMap來試試看看效果如何，輸出結果如下：

10000

我們發現不管運行幾次，HashMap的size都是小于10000的，而ConcurrentHashMap的size都是10000。從這個角度也證明了ConcurrentHashMap是線程安全的，而HashMap則是線程不安全的。 HashMap在多線程put的時候，當產生hash碰撞的時候，會導致丟失數據，因為要put的兩個值hash相同，如果這個對于hash桶的位置個數小于8，那么應該是以鏈表的形式存儲，由于沒有做通過，后面put的元素可能會直接覆蓋之前那個線程put的數據，這樣就導致了數據丟失。

其實列舉上面這個例子只是為了從一個角度來展示下為什么說HashMap線程不安全，而ConcurrentHashMap則是線程安全的，鑒于HashMap線程安全例子比較難列舉出來，所以才通過打印size這個角度來模擬了下。

這篇文章深入解讀HashMap線程安全性問題就詳細介紹了HashMap可能會出現線程安全問題。 文章主要講了兩個可能會出現線程不安全地方，一個是多線程的put可能導致元素的丟失，另一個是put和get并發時，可能導致get為null，但是也僅是在源碼層面分析了下，因為這種場景想要完全用代碼展示出來是稍微有點麻煩的。

接下來我們來看看ConcurrentHashMap是如何做到線程安全的。

JDK8的ConcurrentHashMap文檔提煉

• ConcurrentHashMap支持檢索的完全并發和更新的高預期并發性,這里的說法很有意思檢索支持完全并發，更新則支持高預期并發性，因為它的檢索操作是沒有加鎖的，實際上檢索也沒有必要加鎖。
• 實際上ConcurrentHashMap和Hashtable在不考慮實現細節來說，這兩者完全是可以互相操作的,Hashtable在get，put，remove等這些方法中全部加入了synchronized，這樣的問題是能夠實現線程安全，但是缺點是性能太差，幾乎所有的操作都加鎖的，但是ConcurrentHashMap的檢測操作卻是沒有加鎖的。
• ConcurrentHashMap檢索操作(包括get)通常不會阻塞，因此可能與更新操作(包括put和remove)重疊。
• ConcurrentHashMap跟Hashtable類似但不同于HashMap，它不可以存放空值，key和value都不可以為null。

印象中一直以為ConcurrentHashMap是基于Segment分段鎖來實現的，之前沒仔細看過源碼，一直有這么個錯誤的認識。ConcurrentHashMap是基于Segment分段鎖來實現的，這句話也不能說不對，加個前提條件就是正確的了，ConcurrentHashMap從JDK1.5開始隨java.util.concurrent包一起引入JDK中，在JDK8以前，ConcurrentHashMap都是基于Segment分段鎖來實現的，在JDK8以后，就換成synchronized和CAS這套實現機制了。

JDK1.8中的ConcurrentHashMap中仍然存在Segment這個類，而這個類的聲明則是為了兼容之前的版本序列化而存在的。

???/**?????*?Stripped-down?version?of?helper?class?used?in?previous?version,?????*?declared?for?the?sake?of?serialization?compatibility.?????*/????static?class?Segment?extends?ReentrantLock?implements?Serializable?{????????private?static?final?long?serialVersionUID?=?2249069246763182397L;????????final?float?loadFactor;????????Segment(float?lf)?{?this.loadFactor?=?lf;?}????}

JDK1.8中的ConcurrentHashMap不再使用Segment分段鎖，而是以table數組的頭結點作為synchronized的鎖。和JDK1.8中的HashMap類似，對于hashCode相同的時候，在Node節點的數量少于8個時，這時的Node存儲結構是鏈表形式，時間復雜度為O(N)，當Node節點的個數超過8個時，則會轉換為紅黑樹，此時訪問的時間復雜度為O(long(N))。

?/**?????*?The?array?of?bins.?Lazily?initialized?upon?first?insertion.?????*?Size?is?always?a?power?of?two.?Accessed?directly?by?iterators.?????*/????transient?volatile?Node[]?table;

數據結構圖如下所示：

其實ConcurrentHashMap保證線程安全主要有三個地方。

一、使用volatile保證當Node中的值變化時對于其他線程是可見的

二、使用table數組的頭結點作為synchronized的鎖來保證寫操作的安全

三、當頭結點為null時，使用CAS操作來保證數據能正確的寫入。

使用volatile

可以看到，Node中的val和next都被volatile關鍵字修飾。

volatile的happens-before規則：對一個volatile變量的寫一定可見(happens-before)于隨后對它的讀。

也就是說，我們改動val的值或者next的值對于其他線程是可見的，因為volatile關鍵字，會在讀指令前插入讀屏障，可以讓高速緩存中的數據失效，重新從主內存加載數據。

static?class?Node?implements?Map.Entry?{????????final?int?hash;????????final?K?key;????????volatile?V?val;????????volatile?Node?next;??}??...

另外，ConcurrentHashMap提供類似tabAt來讀取Table數組中的元素，這里是以volatile讀的方式讀取table數組中的元素，主要通過Unsafe這個類來實現的，保證其他線程改變了這個數組中的值的情況下，在當前線程get的時候能拿到。

?static?final??Node?tabAt(Node[]?tab,?int?i)?{????????return?(Node)U.getObjectVolatile(tab,?((long)i?<

而與之對應的，是setTabAt,這里是以volatile寫的方式往數組寫入元素，這樣能保證修改后能對其他線程可見。

?static?final??void?setTabAt(Node[]?tab,?int?i,?Node?v)?{????????U.putObjectVolatile(tab,?((long)i?<

我們來看下ConcurrentHashMap的putVal方法：

??/**?Implementation?for?put?and?putIfAbsent?*/????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[]?tab?=?table;;)?{????????????Node?f;?int?n,?i,?fh;????????????if?(tab?==?null?||?(n?=?tab.length)?==?0)????????????????tab?=?initTable();????????????//當頭結點為null,則通過casTabAt方式寫入????????????else?if?((f?=?tabAt(tab,?i?=?(n?-?1)?&?hash))?==?null)?{????????????????if?(casTabAt(tab,?i,?null,?????????????????????????????new?Node(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;????????????????//頭結點不為null，使用synchronized加鎖????????????????synchronized?(f)?{????????????????????if?(tabAt(tab,?i)?==?f)?{????????????????????????if?(fh?>=?0)?{????????????????????????????//此時hash桶是鏈表結構????????????????????????????binCount?=?1;????????????????????????????for?(Node?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?pred?=?e;????????????????????????????????if?((e?=?e.next)?==?null)?{????????????????????????????????????pred.next?=?new?Node(hash,?key,??????????????????????????????????????????????????????????????value,?null);????????????????????????????????????break;????????????????????????????????}????????????????????????????}????????????????????????}????????????????????????else?if?(f?instanceof?TreeBin)?{????????????????????????????//此時是紅黑樹????????????????????????????Node?p;????????????????????????????binCount?=?2;????????????????????????????if?((p?=?((TreeBin)f).putTreeVal(hash,?key,???????????????????????????????????????????????????????????value))?!=?null)?{????????????????????????????????oldVal?=?p.val;????????????????????????????????if?(!onlyIfAbsent)????????????????????????????????????p.val?=?value;????????????????????????????}????????????????????????}????????????????????????else?if?(f?instanceof?ReservationNode)????????????????????????????throw?new?IllegalStateException("Recursive?update");????????????????????}????????????????}????????????????if?(binCount?!=?0)?{????????????????????//當鏈表結構大于等于8，則將鏈表轉換為紅黑樹????????????????????if?(binCount?>=?TREEIFY_THRESHOLD)????????????????????????treeifyBin(tab,?i);????????????????????if?(oldVal?!=?null)??????????????????return?oldVal;????????????????????break;????????????????}????????????}????????}????????addCount(1L,?binCount);????????return?null;????}

在putVal方法重要的地方都加了注釋，可以幫助理解，現在我們一步一步來看putVal方法。

使用CAS

當有一個新的值需要put到ConcurrentHashMap中時，首先會遍歷ConcurrentHashMap的table數組，然后根據key的hashCode來定位到需要將這個value放到數組的哪個位置。

tabAt(tab, i = (n - 1) & hash))就是定位到這個數組的位置，如果當前這個位置的Node為null，則通過CAS方式的方法寫入。所謂的CAS，即即compareAndSwap，執行CAS操作的時候，將內存位置的值與預期原值比較，如果相匹配，那么處理器會自動將該位置值更新為新值，否則，處理器不做任何操作。

這里就是調用casTabAt方法來實現的。

?????static?final??boolean?casTabAt(Node[]?tab,?int?i,????????????????????????????????????????Node?c,?Node?v)?{????????return?U.compareAndSwapObject(tab,?((long)i?<

casTabAt同樣是通過調用Unsafe類來實現的，調用Unsafe的compareAndSwapObject來實現，其實如果仔細去追蹤這條線路，會發現其實最終調用的是cmpxchg這個CPU指令來實現的，這是一個CPU的原子指令，能保證數據的一致性問題。

使用synchronized

當頭結點不為null時，則使用該頭結點加鎖，這樣就能多線程去put hashCode相同的時候不會出現數據丟失的問題。synchronized是互斥鎖，有且只有一個線程能夠拿到這個鎖，從而保證了put操作是線程安全的。

下面是ConcurrentHashMap的put操作的示意圖，圖片來自于ConcurrentHashMap源碼分析(JDK8)get/put/remove方法分析。

參考文章

從ConcurrentHashMap的演進看Java多線程核心技術

ConcurrentHashMap源碼分析(JDK8)get/put/remove方法分析

總結

以上是生活随笔為你收集整理的concurrenthashmap_ConcurrentHashMap是如何保证线程安全的的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。