ConcurrentHashMap高并發(fā)性的實(shí)現(xiàn)機(jī)制

一、HashMap，即java.util.HashMap標(biāo)準(zhǔn)鏈地址法實(shí)現(xiàn)。這個(gè)不用多解析，下圖十分明了。（圖片來自網(wǎng)絡(luò)）二、Collections.synchronizedMap()函數(shù)返回的線程安全的HashMap這個(gè)的實(shí)現(xiàn)比較簡單。代碼中有：12privatefinalMap<K,V>m;

//BackingMapfinalObject

mutex;//Objectonwhichtosynchronize基本所有的方法都加上了synchronized(mutex)。但是這個(gè)HashMap不能隨便地進(jìn)行迭代，因?yàn)樗皇呛唵伟b了HashMap，而回看HashMap的實(shí)現(xiàn)，我們可以發(fā)現(xiàn)，對(duì)于沖突的key，形成一個(gè)鏈表，明顯如果有一個(gè)線程在歷遍HashMap，另一個(gè)線程在做刪除操作，則很有可能出錯(cuò)。因此，JDK中給出以下代碼：123456789Mapm=Collections.synchronizedMap(newHashMap());

...

Sets=m.keySet();

//Needn'tbeinsynchronizedblock

...

synchronized(m){

//Synchronizingonm,nots!

Iteratori=s.iterator();//Mustbeinsynchronizedblock

while(i.hasNext())

foo(i.next());

}三、ConcurrentHashMap對(duì)于HashMap，最主要的是以下四種的操作：1234publicVget(Objectkey)

publicVput(Kkey,Vvalue)

publicVremove(Objectkey)

迭代在多線程環(huán)境下，get，put，remove都是比較容易實(shí)現(xiàn)的（如果不考慮效率，簡單加鎖即可），迭代的操作才是真正的難點(diǎn)。從Collections.synchronizedMap()的迭代來看，它并不能做到對(duì)客戶代碼透明，有點(diǎn)蛋疼。下面簡單分析ConcurrentHashMap的實(shí)現(xiàn)，相當(dāng)精巧。默認(rèn)一個(gè)ConcurrentHashMap中有16個(gè)子HashMap，所以相當(dāng)于一個(gè)二級(jí)哈希。對(duì)于所有的操作都是先定位到子HashMap，再作相應(yīng)的操作。對(duì)于：publicVget(Objectkey)先得到key所在的table，再像HashMap一樣get

中間并不加鎖publicVput(Kkey,Vvalue)先得到所屬的table，加鎖

判斷table是否要擴(kuò)容

如果table要擴(kuò)容，則產(chǎn)生newTable

hash值相同的slot整體移到newTable

hash值不同的slot，把oldTable中的所有Entry都復(fù)制到newTable中

因?yàn)橛锌赡芷渌€程在歷遍這個(gè)table，所以不能把原來的鏈表拆斷publicVremove(Objectkey)remove操作，如下圖，要?jiǎng)h除Entry3，則先復(fù)制Entry1為Entry1*，Entry1*指向Entry4，再復(fù)制Entry2為Entry2*，Entry2*指向Entry1*，最終形成一個(gè)兩叉的鏈表。原本的Entry1，Entry2，Entry3會(huì)被GC自動(dòng)回收。迭代操作：ConcurrentHashMap的歷遍是從后向前歷遍的，因?yàn)槿绻辛硪粋€(gè)線程B在執(zhí)行clear操作時(shí)，會(huì)把table中的所有slot都置為null，這個(gè)操作是從前向后執(zhí)行

如果線程A在歷遍Map時(shí)也是從前向后，則有可能會(huì)出現(xiàn)追趕現(xiàn)象。以下代碼：123456HashMap<Integer,String>m1=newHashMap();

m1.put(1,"001");

m1.put(2,"002");

for(Entry<Integer,String>entry:m1.entrySet()){

System.out.println("key:"+entry.getKey());

}HashMap輸出的是key:1key:2

ConcurrentHashMap輸出的是key:2key:1考慮到在使用HashMap在并發(fā)時(shí)會(huì)出現(xiàn)不正確行為，根據(jù)網(wǎng)上資料自己編寫了采用ConcurrentHashMap來完成靜態(tài)緩存的處理，目的是為了能夠用來處理高并發(fā)的線程安全類，如有問題請(qǐng)各位大俠指教：[java]\o"viewplain"viewplain\o"copy"copy\o"print"print\o"?"?package

com.zengms.cache;

import

java.util.Map;

import

java.util.concurrent.ConcurrentHashMap;

import

mons.logging.Log;

import

mons.logging.LogFactory;

public

class

MapCacheManager

{

private

final

static

Log

log

=

LogFactory.getLog(MapCacheManager.class);

private

volatile

long

updateTime

=

0L;//

更新緩存時(shí)記錄的時(shí)間

private

volatile

boolean

updateFlag

=

true;//

正在更新時(shí)的閥門，為false時(shí)表示當(dāng)前沒有更新緩存，為true時(shí)表示當(dāng)前正在更新緩存

private

volatile

static

MapCacheManager

mapCacheObject;//

緩存實(shí)例對(duì)象

private

static

Map<String,

String>

cacheMap

=

new

ConcurrentHashMap<String,

String>();//

緩存容器

private

MapCacheManager()

{

this.LoadCache();//

加載緩存

updateTime

=

System.currentTimeMillis();//

緩存更新時(shí)間

}

/**

*

采用單例模式獲取緩存對(duì)象實(shí)例

*

*

@return

*/

public

static

MapCacheManager

getInstance()

{

if

(null

==

mapCacheObject)

{

synchronized

(MapCacheManager.class)

{

if

(null

==

mapCacheObject)

{

mapCacheObject

=

new

MapCacheManager();

}

}

}

return

mapCacheObject;

}

/**

*

裝載緩存

*/

private

void

LoadCache()

{

this.updateFlag

=

true;//

正在更新

/**********

數(shù)據(jù)處理，將數(shù)據(jù)放入cacheMap緩存中

**begin

******/

cacheMap.put("key1",

"value1");

cacheMap.put("key2",

"value2");

cacheMap.put("key3",

"value3");

cacheMap.put("key4",

"value4");

cacheMap.put("key5",

"value5");

/**********

數(shù)據(jù)處理，將數(shù)據(jù)放入cacheMap緩存中

***end

*******/

this.updateFlag

=

false;//

更新已完成

}

/**

*

返回緩存對(duì)象

*

*

@return

*/

public

Map<String,

String>

getMapCache()

{

long

currentTime

=

System.currentTimeMillis();

if

(this.updateFlag)

{//

前緩存正在更新

("cache

is

Instance

.....");

return

null;

}

if

(this.IsTimeOut(currentTime))

{//

如果當(dāng)前緩存正在更新或者緩存超出時(shí)限，需重新加載

synchronized

(this)

{

this.ReLoadCache();

this.updateTime

=

currentTime;

}

}

return

this.cacheMap;

}

private

boolean

IsTimeOut(long

currentTime)

{

return

((currentTime

-

this.updateTime)

>

1000000);//

超過時(shí)限，超時(shí)

}

/**

*

獲取緩存項(xiàng)大小

*

@return

*/

private

int

getCacheSize()

{

return

cacheMap.size();

}

/**

*

獲取更新時(shí)間

*

@return

*/

private

long

getUpdateTime()

{

return

this.updateTime;

}

/**

*

獲取更新標(biāo)志

*

@return

*/

private

boolean

getUpdateFlag()

{

return

this.updateFlag;

}

/**

*

重新裝載

*/

private

void

ReLoadCache()

{

this.cacheMap.clear();

this.LoadCache();

}

}

測試代碼：[java]\o"viewplain"viewplain\o"copy"copy\o"print"print\o"?"?package

com.zengms.cache;

import

java.util.Iterator;

import

java.util.Map;

import

java.util.Set;

import

java.util.concurrent.ConcurrentHashMap;

public

class

CacheTest

{

public

static

void

main(String[]

args)

{

MapCacheManager

cache

=

MapCacheManager.getInstance();

Map<String,

String>

cacheMap

=

new

ConcurrentHashMap<String,

String>();

cacheMap

=

cache.getMapCache();

Set<String>

set

=

cacheMap.keySet();

Iterator<String>

it

=

set.iterator();

while(it.hasNext()){

String

key

=

it.next();

System.out.println(key+"="+cacheMap.get(key));

}

}

}

現(xiàn)階段的學(xué)習(xí)策略是理解和實(shí)踐這些知識(shí)點(diǎn)，并沒有深入分析其原理，但確實(shí)精讀了許多關(guān)于這個(gè)主題基礎(chǔ)性的資料讓我很受益（見參考資料）。哈希表基礎(chǔ)1.哈希表是基于數(shù)組的數(shù)據(jù)結(jié)構(gòu)2.通過對(duì)關(guān)鍵字的哈希運(yùn)算實(shí)現(xiàn)元素的快速定位3.哈希表的重點(diǎn)是哈?；；?fù)責(zé)把一個(gè)大范圍的數(shù)字轉(zhuǎn)化成一個(gè)小范圍的數(shù)字4.哈?；^程中會(huì)產(chǎn)生值沖突，這種情況有多種辦法可以解決（開放地址法、鏈地址法）4.1.開放地址法，通過在哈希表中尋找一個(gè)空位解決沖突問題，尋找空位的方法也有多種（線性探測、二次探測、再哈希）4.2.鏈地址法，通過在哈希表單元中加入鏈表的方式解決5.哈希表的重要缺點(diǎn)5.1.當(dāng)存儲(chǔ)數(shù)組基本被填滿時(shí)性能下降很高5.2.對(duì)存儲(chǔ)數(shù)組進(jìn)行擴(kuò)容會(huì)分別對(duì)已存儲(chǔ)的元素重新計(jì)算哈希的過程ConcurrentHashMap分段與鎖的學(xué)習(xí)一、結(jié)構(gòu)

二、定位分段這塊對(duì)Key的哈希值進(jìn)行移位處理，首先給定的Key在哪一段，然后從具體段中定位Hash值對(duì)應(yīng)具體值對(duì)象。

三、鎖ConcurrentHashMap沒有將每個(gè)方法都在同一個(gè)鎖上同步并使得每次只能有一個(gè)線程訪問容器，而是使用一種粒度更細(xì)的加鎖機(jī)制實(shí)現(xiàn)更大程度的共享。這種細(xì)粒度的加鎖機(jī)制體現(xiàn)在ConcurrentHashMap劃分的Segment數(shù)組，Segment數(shù)組上各Segment元素代表了粒度更細(xì)的鎖，從結(jié)構(gòu)圖中可以看到Segment繼承自ReentrantLock可重入鎖。ConcurrentHashMap這種基于分組Segment并加鎖的策略可在高并發(fā)的環(huán)境下獲得更高的吞吐量。ConcurrentHashMap實(shí)現(xiàn)并發(fā)的基礎(chǔ)操作都通過sun.misc.Unsafe完成。

四、sun.misc.UnsafeUnsafe類相關(guān)于一個(gè)工具類，低層調(diào)用native方法，提供了基礎(chǔ)CompareAndSet(CAS)支持，通過CAS可以在不加鎖并發(fā)情況下實(shí)現(xiàn)數(shù)字或引用的細(xì)粒度更新。ConcurrentHashMap的鎖分離技術(shù)博客分類：源碼學(xué)習(xí)

concurrenthashmap是一個(gè)非常好的map實(shí)現(xiàn)，在高并發(fā)操作的場景下會(huì)有非常好的效率。實(shí)現(xiàn)的目的主要是為了避免同步操作時(shí)對(duì)整個(gè)map對(duì)象進(jìn)行鎖定從而提高并發(fā)訪問能力。

ConcurrentHashMap類中包含兩個(gè)靜態(tài)內(nèi)部類HashEntry和Segment。HashEntry用來封裝映射表的鍵/值對(duì)；Segment用來充當(dāng)鎖的角色，每個(gè)Segment對(duì)象守護(hù)整個(gè)散列映射表的若干個(gè)桶。每個(gè)桶是由若干個(gè)HashEntry對(duì)象鏈接起來的鏈表。一個(gè)ConcurrentHashMap實(shí)例中包含由若干個(gè)Segment對(duì)象組成的數(shù)組。

Java代碼

\o"收藏這段代碼"static

final

class

HashEntry<K,V>

{

final

K

key;

//

聲明

key

為

final

型

final

int

hash;

//

聲明

hash

值為

final

型

volatile

V

value;

//

聲明

value

為

volatile

型

final

HashEntry<K,V>

next;

//

聲明

next

為

final

型

HashEntry(K

key,

int

hash,

HashEntry<K,V>

next,

V

value)

{

this.key

=

key;

this.hash

=

hash;

this.next

=

next;

this.value

=

value;

}

}

Java代碼

\o"收藏這段代碼"static

final

class

Segment<K,V>

extends

ReentrantLock

implements

Serializable

{

transient

volatile

int

count;

//在本

segment

范圍內(nèi)，包含的

HashEntry

元素的個(gè)數(shù)

//volatile

型

transient

int

modCount;

//table

被更新的次數(shù)

transient

int

threshold;

//默認(rèn)容量

final

float

loadFactor;

//裝載因子

/**

*

table

是由

HashEntry

對(duì)象組成的數(shù)組

*

如果散列時(shí)發(fā)生碰撞，碰撞的

HashEntry

對(duì)象就以鏈表的形式鏈接成一個(gè)鏈表

*

table

數(shù)組的數(shù)組成員代表散列映射表的一個(gè)桶

*/

transient

volatile

HashEntry<K,V>[]

table;

/**

*

根據(jù)

key

的散列值，找到

table

中對(duì)應(yīng)的那個(gè)桶（table

數(shù)組的某個(gè)數(shù)組成員）

*

把散列值與

table

數(shù)組長度減

1

的值相“與”，得到散列值對(duì)應(yīng)的

table

數(shù)組的下標(biāo)

*

然后返回

table

數(shù)組中此下標(biāo)對(duì)應(yīng)的

HashEntry

元素

*

即這個(gè)段中鏈表的第一個(gè)元素

*/

HashEntry<K,V>

getFirst(int

hash)

{

HashEntry<K,V>[]

tab

=

table;

return

tab[hash

&

(tab.length

-

1)];

}

Segment(int

initialCapacity,

float

lf)

{

loadFactor

=

lf;

setTable(HashEntry.<K,V>newArray(initialCapacity));

}

/**

*

設(shè)置

table

引用到這個(gè)新生成的

HashEntry

數(shù)組

*

只能在持有鎖或構(gòu)造函數(shù)中調(diào)用本方法

*/

void

setTable(HashEntry<K,V>[]

newTable)

{

threshold

=

(int)(newTable.length

*

loadFactor);

table

=

newTable;

}

}

注意Segment繼承了ReentrantLock鎖

左邊便是Hashtable的實(shí)現(xiàn)方式---鎖整個(gè)hash表；而右邊則是ConcurrentHashMap的實(shí)現(xiàn)方式---鎖桶（或段）。ConcurrentHashMap將hash表分為16個(gè)桶（默認(rèn)值），諸如get,put,remove等常用操作只鎖當(dāng)前需要用到的桶。試想，原來只能一個(gè)線程進(jìn)入，現(xiàn)在卻能同時(shí)16個(gè)寫線程進(jìn)入（寫線程才需要鎖定，而讀線程幾乎不受限制，之后會(huì)提到），并發(fā)性的提升是顯而易見的。

更令人驚訝的是ConcurrentHashMap的讀取并發(fā)，因?yàn)樵谧x取的大多數(shù)時(shí)候都沒有用到鎖定，所以讀取操作幾乎是完全的并發(fā)操作，而寫操作鎖定的粒度又非常細(xì)，比起之前又更加快速（這一點(diǎn)在桶更多時(shí)表現(xiàn)得更明顯些）。只有在求size等操作時(shí)才需要鎖定整個(gè)表。而在迭代時(shí)，ConcurrentHashMap使用了不同于傳統(tǒng)集合的快速失敗迭代器（見之前的文章《JAVAAPI備忘---集合》）的另一種迭代方式，我們稱為弱一致迭代器。在這種迭代方式中，當(dāng)iterator被創(chuàng)建后集合再發(fā)生改變就不再是拋出ConcurrentModificationException，取而代之的是在改變時(shí)new新的數(shù)據(jù)從而不影響原有的數(shù)據(jù)，iterator完成后再將頭指針替換為新的數(shù)據(jù)，這樣iterator線程可以使用原來老的數(shù)據(jù)，而寫線程也可以并發(fā)的完成改變，更重要的，這保證了多個(gè)線程并發(fā)執(zhí)行的連續(xù)性和擴(kuò)展性，是性能提升的關(guān)鍵。

接下來，讓我們看看ConcurrentHashMap中的幾個(gè)重要方法，心里知道了實(shí)現(xiàn)機(jī)制后，使用起來就更加有底氣。

ConcurrentHashMap中主要實(shí)體類就是三個(gè)：ConcurrentHashMap（整個(gè)Hash表）,Segment（桶），HashEntry（節(jié)點(diǎn)），對(duì)應(yīng)上面的圖可以看出之間的關(guān)系。

get方法（請(qǐng)注意，這里分析的方法都是針對(duì)桶的，因?yàn)镃oncurrentHashMap的最大改進(jìn)就是將粒度細(xì)化到了桶上），首先判斷了當(dāng)前桶的數(shù)據(jù)個(gè)數(shù)是否為0，為0自然不可能get到什么，只有返回null，這樣做避免了不必要的搜索，也用最小的代價(jià)避免出錯(cuò)。然后得到頭節(jié)點(diǎn)（方法將在下面涉及）之后就是根據(jù)hash和key逐個(gè)判斷是否是指定的值，如果是并且值非空就說明找到了，直接返回；程序非常簡單，但有一個(gè)令人困惑的地方，這句returnreadValueUnderLock(e)到底是用來干什么的呢？研究它的代碼，在鎖定之后返回一個(gè)值。但這里已經(jīng)有一句Vv=e.value得到了節(jié)點(diǎn)的值，這句returnreadValueUnderLock(e)是否多此一舉？事實(shí)上，這里完全是為了并發(fā)考慮的，這里當(dāng)v為空時(shí)，可能是一個(gè)線程正在改變節(jié)點(diǎn)，而之前的get操作都未進(jìn)行鎖定，根據(jù)bernstein條件，讀后寫或?qū)懞笞x都會(huì)引起數(shù)據(jù)的不一致，所以這里要對(duì)這個(gè)e重新上鎖再讀一遍，以保證得到的是正確值，這里不得不佩服DougLee思維的嚴(yán)密性。整個(gè)get操作只有很少的情況會(huì)鎖定，相對(duì)于之前的Hashtable，并發(fā)是不可避免的啊！

get操作不需要鎖。第一步是訪問count變量，這是一個(gè)volatile變量，由于所有的修改操作在進(jìn)行結(jié)構(gòu)修改時(shí)都會(huì)在最后一步寫count變量，通過這種機(jī)制保證get操作能夠得到幾乎最新的結(jié)構(gòu)更新。對(duì)于非結(jié)構(gòu)更新，也就是結(jié)點(diǎn)值的改變，由于HashEntry的value變量是volatile的，也能保證讀取到最新的值。接下來就是對(duì)hash鏈進(jìn)行遍歷找到要獲取的結(jié)點(diǎn)，如果沒有找到，直接訪回null。對(duì)hash鏈進(jìn)行遍歷不需要加鎖的原因在于鏈指針next是final的。但是頭指針卻不是final的，這是通過getFirst(hash)方法返回，也就是存在table數(shù)組中的值。這使得getFirst(hash)可能返回過時(shí)的頭結(jié)點(diǎn)，例如，當(dāng)執(zhí)行g(shù)et方法時(shí)，剛執(zhí)行完getFirst(hash)之后，另一個(gè)線程執(zhí)行了刪除操作并更新頭結(jié)點(diǎn)，這就導(dǎo)致get方法中返回的頭結(jié)點(diǎn)不是最新的。這是可以允許，通過對(duì)count變量的協(xié)調(diào)機(jī)制，get能讀取到幾乎最新的數(shù)據(jù)，雖然可能不是最新的。要得到最新的數(shù)據(jù)，只有采用完全的同步。Java代碼

V

get(Object

key,

int

hash)

{

if

(count

!=

0)

{

//

read-volatile

HashEntry

e

=

getFirst(hash);

while

(e

!=

null)

{

if

(e.hash

==

hash

&&

key.equals(e.key))

{

V

v

=

e.value;

if

(v

!=

null)

return

v;

return

readValueUnderLock(e);

//

recheck

}

e

=

e.next;

}

}

return

null;

}

V

readValueUnderLock(HashEntry

e)

{

lock();

try

{

return

e.value;

}

finally

{

unlock();

}

}

put操作一上來就鎖定了整個(gè)segment，這當(dāng)然是為了并發(fā)的安全，修改數(shù)據(jù)是不能并發(fā)進(jìn)行的，必須得有個(gè)判斷是否超限的語句以確保容量不足時(shí)能夠rehash，而比較難懂的是這句intindex=hash&(tab.length-1)，原來segment里面才是真正的hashtable，即每個(gè)segment是一個(gè)傳統(tǒng)意義上的hashtable,如上圖，從兩者的結(jié)構(gòu)就可以看出區(qū)別，這里就是找出需要的entry在table的哪一個(gè)位置，之后得到的entry就是這個(gè)鏈的第一個(gè)節(jié)點(diǎn)，如果e!=null，說明找到了，這是就要替換節(jié)點(diǎn)的值（onlyIfAbsent==false），否則，我們需要new一個(gè)entry，它的后繼是first，而讓tab[index]指向它，什么意思呢？實(shí)際上就是將這個(gè)新entry插入到鏈頭，剩下的就非常容易理解了。

Java代碼

V

put(K

key,

int

hash,

V

value,

boolean

onlyIfAbsent)

{

lock();

try

{

int

c

=

count;

if

(c++

>

threshold)

//

ensure

capacity

rehash();

HashEntry[]

tab

=

table;

int

index

=

hash

&

(tab.length

-

1);

HashEntry

first

=

(HashEntry)

tab[index];

HashEntry

e

=

first;

while

(e

!=

null

&&

(e.hash

!=

hash

||

!key.equals(e.key)))

e

=

e.next;

V

oldValue;

if

(e

!=

null)

{

oldValue

=

e.value;

if

(!onlyIfAbsent)

e.value

=

value;

}

else

{

oldValue

=

null;

++modCount;

tab[index]

=

new

HashEntry(key,

hash,

first,

value);

count

=

c;

//

write-volatile

}

return

oldValue;

}

finally

{

unlock();

}

}

remove操作非常類似put，但要注意一點(diǎn)區(qū)別，中間那個(gè)for循環(huán)是做什么用的呢？（*