本篇文章主要介紹一下JDK1.7中的ConcurrentHashMap的一些代碼結構。ConcurrentHashMap顧名思義就是線程安全的HashMap，相對于HashMap來說，可以保證線程安全的問題；對于HashTable來說可以提高存取的效率。
基本結構
從上述類圖可以看出，ConcurrentHashMap除了Map通用的get()、put()等方法外，還有一個重要的屬性Segment數組。每個Segment又是一個ReentrantLock，其内部又有一個HashEntry數組，HashEntry是最終key-value的存儲單元。可以看出，HashEntry是一個連結清單結構。進而可以大緻的得到如下結構圖：
初始化
ConcurrentHashMap構造方法的3個入參分别是：initialCapacity、loadFactor、concurrencyLevel，用于初始化segment數組、segmentShift、segmentMask等資料。構造方法如下：
public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {
// 1、參數校驗
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
// 2、ssize是segments數組的大小，這裡取的是大于concurrencyLevel的2^n的一個值
int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
// 3、sshift就是上面描述中的n值，預設情況下concurrencyLevel等于16，sshift就等于4。是以預設情況下segmentShift的值就是28，這個值會參與hash運算。
// segmentMask是hash運算的掩碼，預設情況下等于16-1=15，類似于網絡中的子網路遮罩，segmentMask的二進制最後幾位都是1，最大值是末尾16個1（65535）。
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
// 4、下面的代碼主要是用來初始化segment，其中cap是segment裡面的HashEntry數組的長度。它取的是大于等于c（Map容量/ssize）的2^N的一個值。
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
// 5、建立segments和segments[0]（這裡面隻初始化了一個segments數組中的第0個元素）
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
- initialCapacity：ConcurrentHashMap的初始化容量，預設值是16（DEFAULT_INITIAL_CAPACITY）；
- loadFactor：負載因子，預設值是0.75f，當ConcurrentHashMap中的HashEntry數量到達 loadFactor * 目前容量 時，就會進行rehash()進行擴容；
- concurrencyLevel：并發的級别，也就是Segment數組的大小（ConcurrentHashMap是對每個Segment加鎖，是以得名并發級别），預設值是16（DEFAULT_CONCURRENCY_LEVEL）。這裡需要注意的是concurrencyLevel的大小都是2^n，concurrencyLevel會取大于傳入值的鄰近的一個值。如：傳入15，則concurrencyLevel=16（2^4）。
下面介紹一下構造函數中的初始化流程：
- 參數校驗。
- 設定ssize的值，ssize是segments數組的大小，這裡取的是大于concurrencyLevel的2^n的一個值。
- 設定segmentShift和segmentMask的值，sshift就是上面描述中的n值，預設情況下concurrencyLevel等于16，sshift就等于4。是以預設情況下segmentShift的值就是28，這個值會參與hash運算。segmentMask是hash運算的掩碼，預設情況下等于16-1=15，類似于網絡中的子網路遮罩，segmentMask的二進制最後幾位都是1，最大值是末尾16個1（65535）。
- 初始化segment，其中cap是segment裡面的HashEntry數組的長度。它取的是大于等于c（Map容量/ssize）的2^N的一個值。
- 建立segments和segments[0]（這裡面隻初始化了一個segments數組中的第0個元素）。
定位Segment
ConcurrentHashMap使用分段鎖Segment來保護不同的資料，下面簡單的描述一下定位的流程。
第一步是擷取key的hash值，代碼如下：
private int hash(Object k) {
int h = hashSeed;
if ((0 != h) && (k instanceof String)) {
return sun.misc.Hashing.stringHash32((String) k);
}
h ^= k.hashCode();
// 将key的hash值打開到segment分段中去
// 使用了Wang/Jenkins hash的變體算法
h += (h << 15) ^ 0xffffcd7d;
h ^= (h >>> 10);
h += (h << 3);
h ^= (h >>> 6);
h += (h << 2) + (h << 14);
return h ^ (h >>> 16);
}
第二步，通過上面的雜湊演算法得到的值，然後進行一下位移操作，取高位數值。
int index = (h >>> segmentShift) & segmentMask;
預設情況下segmentShift為28，segmentMask為15（低位有1111），進而可以得到h的高四位的值。
常用方法
get()
get操作是先定位到segment，然後再到segment中去擷取對應的value值。代碼如下：
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key);
// 根據Segment的索引（(h >>> segmentShift) & segmentMask）算出在Segment[]上的偏移量
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
// 若Segment存在，則繼續查找table[]的索引位置；
// 根據table的索引（(tab.length - 1) & h）算出在table[]上的偏移量，循環連結清單找出結果
for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
K k;
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
通過上面代碼可以看出：
- 首先，根據Segment的索引（(h >>> segmentShift) & segmentMask）算出在Segment[]上的偏移量。然後根據偏移量，調用UNSAFE.getObjectVolatile()判斷其是否存在。
- 若Segment存在，則繼續查找table[]的索引位置；根據table的索引（(tab.length - 1) & h）算出在table[]上的偏移量，循環連結清單找出結果。
從上面步驟可以看出，整個過程中讀取并未加鎖。是以在讀取的過程中，如果出現HashEntry的變更，則無法得到真正的結果。這一點是ConcurrentHashMap在弱一緻性上的展現。如果要求強一緻性，就必須要加鎖控制。
put()
對于put()操作，前面的定位Segment的操作都是和put()相同的。找到Segment以後，然後對整個Segment加鎖，然後再進行後續的操作。下面詳細介紹一下：
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
// 定位Segment，并判斷其是否存在
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);// 如果不存在，則建立；這是因為在構造ConcurrentHashMap的時候，隻建立了segment[0]這個元素
return s.put(key, hash, value, false);// 送出給Segment去處理
}
可以看出，首先定位Segment，并判斷其是否存在；如果不存在，則建立（這是因為在構造ConcurrentHashMap的時候，隻建立了segment[0]這個元素）；然後再送出給Segment去處理。下面看一下建立Segment的過程：
private Segment<K,V> ensureSegment(int k) {
final Segment<K,V>[] ss = this.segments;
// 定位Segment裡面table的位置
long u = (k << SSHIFT) + SBASE; // raw offset
Segment<K,V> seg;
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
Segment<K,V> proto = ss[0]; // 使用segment[0]作為原型建立，這裡面免去了一些計算
int cap = proto.table.length; // 2^n的一個值
float lf = proto.loadFactor; // 預設0.75f
int threshold = (int)(cap * lf);
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // 準備建立前，再去檢查是否已經建立
// 使用CAS建立，直至成功
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
}
上述首先定位出了table在segment中的位置；然後使用segment[0]作為原型建立元素（這裡面免去了一些計算）；然後使用CAS建立，直至成功。
下面看一下具體的put方法，由Segment實作：
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
// 嘗試加鎖，首先嘗試tryLock()，多次失敗以後使用lock()；同時會查找HashEntry，如果沒有找到，建立并傳回一個
HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
// 循環定位連結清單中的HashEntry位置，然後執行變更
for (HashEntry<K,V> e = first;;) {
if (e != null) {// 查找key，若找到直接修改value值
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {// 到達連結清單尾部
if (node != null)// node如果不等于null，說明之前已經預熱完成，可以直接插入
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
// 檢查是否到達threshold，到達後則進行rehash()
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
從上面Segment的put操作可以看出：
- 首先，segment嘗試tryLock()，多次失敗以後使用lock()，同時會查找HashEntry，如果沒有找到，建立并傳回一個（預熱操作）。
- 然後，循環定位連結清單中的HashEntry位置。如果查找到key，若找到直接修改value值，并退出。如果找不到則在連結清單頭部新增一個HashEntry節點。
- 在新增節點的之後，會檢查是否會達到threshold，如果到達則進行rehash()擴容。
關于rehash()操作，後面單獨介紹。下面介紹一下scanAndLockForPut()的實作。
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1; // 定位node時為負值
while (!tryLock()) {// 這裡首先嘗試使用tryLock()，達到最大重試次數MAX_SCAN_RETRIES後，轉為lock()的阻塞操作
HashEntry<K,V> f; // to recheck first below
if (retries < 0) {// 定位node的位置，若找不到則建立一個HashEntry
if (e == null) {
if (node == null) // speculatively create node
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) {// 達到MAX_SCAN_RETRIES時，進行阻塞加鎖
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // 如果加鎖過程中，node有新增，則重新周遊連結清單，（這裡可以解釋對于連結清單的插入位置總是head的問題了）
retries = -1;
}
}
return node;
}
從上述代碼可以看出，執行步驟大概如下：
- 這裡首先嘗試使用tryLock()，達到最大重試次數MAX_SCAN_RETRIES後，轉為lock()的阻塞操作；
- 定位node位置的時候，如果找不到則建立一個HashEntry；
- 如果加鎖過程中，node有新增，則重新周遊連結清單，（這裡可以解釋對于連結清單的插入位置總是head的問題了）。
rehash操作
rehash也就是擴容操作，擴容之後的容量是之前的兩倍，是以擴容之後的newCapacity也是2^n的一個值。
private void rehash(HashEntry<K,V> node) {
/*
* 将table中每個節點重新配置設定到新的table中去。因為使用的是 *2的方式擴容,
* 每個元素在table中的索引要麼為i（不變）,要麼是i+oldCapacity。
* 如：擴容前容量是16，目前HashEntry在table[]中的索引為3，則新的索引可能為3或者19。
* 在節點拷貝的過程中，有一些節點的next節點是不用調整的，就直接利用了。
* 據統計，在預設的threshold值時, 擴容隻需要1/6的節點需要拷貝。
* 那些被替換掉的節點，在沒有任何線程引用的時候，将會被GC回收。
* Entry accesses use plain array indexing because they are followed by volatile table write.
*/
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1; // 容量*2操作
threshold = (int)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
int sizeMask = newCapacity - 1;
// 周遊擴容前節點
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
int idx = e.hash & sizeMask;
if (next == null) // 若連結清單為單節點
newTable[idx] = e;
else { // 重複利用一些擴容後，next不變的節點，這些節點在原先連結清單的尾部
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// 對于next變化的節點重新計算hash（連結清單前面部分節點），然後重新插入
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
int nodeIndex = node.hash & sizeMask; // 将需要put的新node插入
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
從上面的代碼流程可以看出，其擴容步驟大緻如下：
- 首先計算出newCapacity的容量；
- 然後循環table[]，重新配置設定每條連結清單上面的元素。因為使用的是 *2的方式擴容，每個元素在table中的索引要麼為i（不變）,要麼是i+oldCapacity。如：擴容前容量是16，目前HashEntry在table[]中的索引為3，則新的索引可能為3或者19。
- 拷貝過程中，如果為單連結清單則直接指派；在節點拷貝的過程中，有一些節點的next節點是不用調整的（連結清單後端部分片段），就直接利用了；對于前端部分的片段，則重新hash，然後插入到對應的連結清單中。
- 最後再将需要put進來的node，在擴容後的結構中插入。
size()
ConcurrentHashMap的size()操作需要統計所有的Segment中的HashEntry數量，最大為Integer.MAX_VALUE。因為在統計個過程中，有可能出現多線程修改的問題。即便如此，ConcurrentHashMap首先會用無鎖嘗試3次，如果統計失敗，再加鎖統計。代碼如下：
public int size() {
// 首先嘗試3次無鎖的統計，如果失敗，再進入加鎖統計
final Segment<K,V>[] segments = this.segments;
int size;
boolean overflow; // 當大小超過 32 bits 時為true
long sum; // sum of modCounts
long last = 0L; // previous sum
int retries = -1; // first iteration isn't retry
try {
for (;;) {
if (retries++ == RETRIES_BEFORE_LOCK) {// 加鎖統計
for (int j = 0; j < segments.length; ++j)
ensureSegment(j).lock(); // force creation
}
sum = 0L;
size = 0;
overflow = false;
// 嘗試3次無鎖統計，這裡面通過統計前後的modCount值的和 變化，這個值在每個Segment中，每一次變更操作都會遞增，類似于Segment的版本号
for (int j = 0; j < segments.length; ++j) {
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null) {
sum += seg.modCount;
int c = seg.count;
if (c < 0 || (size += c) < 0)
overflow = true;
}
}
if (sum == last)
break;
last = sum;
}
} finally {
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
return overflow ? Integer.MAX_VALUE : size;
}
containsValue()
containsValue()方法的思想和size()的基本相同，下面貼出代碼，相信檢視備注即可。
public boolean containsValue(Object value) {
// Same idea as size()
if (value == null)
throw new NullPointerException();
final Segment<K,V>[] segments = this.segments;
boolean found = false;
long last = 0;
int retries = -1;
try {
outer: for (;;) {
if (retries++ == RETRIES_BEFORE_LOCK) {// 加鎖統計
for (int j = 0; j < segments.length; ++j)
ensureSegment(j).lock(); // force creation
}
long hashSum = 0L;
int sum = 0;
// 嘗試3次無鎖統計，這裡面通過統計前後的modCount值的和 變化，這個值在每個Segment中，每一次變更操作都會遞增，類似于Segment的版本号
for (int j = 0; j < segments.length; ++j) {
HashEntry<K,V>[] tab;
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null && (tab = seg.table) != null) {
for (int i = 0 ; i < tab.length; i++) {
HashEntry<K,V> e;
for (e = entryAt(tab, i); e != null; e = e.next) {
V v = e.value;
if (v != null && value.equals(v)) {
found = true;
break outer;
}
}
}
sum += seg.modCount;
}
}
if (retries > 0 && sum == last)
break;
last = sum;
}
} finally {
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
return found;
}
參考：http://www.infoq.com/cn/articles/ConcurrentHashMap