主要內(nèi)容：
- ConcurrentHashMap數(shù)據(jù)結(jié)構(gòu)
- 繼承關(guān)系、關(guān)鍵屬性、構(gòu)造函數(shù)
- 插入元素
ConcurrentHashMap概述(基于jdk1.7)
ConcurrentHashMap是線程安全的集合，與Hashtable相比ConcurrentHashMap效率高。并發(fā)環(huán)境下Hashtable效率低下的原因是訪問(wèn)Hashtable的多個(gè)線程都競(jìng)爭(zhēng)同一把鎖，當(dāng)然訪問(wèn)慢了。那我們很容易想到如果有多把鎖，每把鎖作用于集合的一部分?jǐn)?shù)據(jù)，那么訪問(wèn)多線程訪問(wèn)不同數(shù)據(jù)段的數(shù)據(jù)時(shí)就不會(huì)發(fā)生鎖競(jìng)爭(zhēng)的情況，那么效率當(dāng)然就提高了。這其實(shí)就是ConcurrentHashMap的鎖分離技術(shù)。
鎖分離技術(shù)：將數(shù)據(jù)分成一段一段的存儲(chǔ)，每一段數(shù)據(jù)都有一把鎖，當(dāng)一個(gè)線程訪問(wèn)某段數(shù)據(jù)時(shí)，其他段數(shù)據(jù)可以被其他線程鎖訪問(wèn)。
ConcurrentHashMap數(shù)據(jù)結(jié)構(gòu)
ConcurrentHashMap數(shù)據(jù)結(jié)構(gòu).png
底層基于分段的數(shù)組和鏈表實(shí)現(xiàn)，由
Segment段類(lèi)型的數(shù)組結(jié)構(gòu)和HashEntry類(lèi)型的數(shù)據(jù)結(jié)構(gòu)存儲(chǔ)數(shù)據(jù)，將Segement數(shù)組聲明為final。
- Segment：繼承了ReentrantLock，可重入鎖。Segment中包含了HashEntry類(lèi)型的數(shù)組。
static final class Segment<K,V> extends ReentrantLock
implements Serializable {
transient volatile HashEntry<K,V>[] table;//HashEntry類(lèi)型的數(shù)組
transient int count;//Segment中保存的元素?cái)?shù)量
transient int modCount;//段修改的次數(shù)
transient int threshold;//閾值，超過(guò)閾值后數(shù)組會(huì)進(jìn)行擴(kuò)容
final float loadFactor;//負(fù)載因子
}
-
HashEntry：鏈表上的節(jié)點(diǎn)，定義了鍵key、值value、哈希值hash以及指向下一個(gè)節(jié)點(diǎn)的指針next。這里注意到哈希值、鍵值是
final型的，保證讀操作時(shí)不需要加鎖。值、下一節(jié)點(diǎn)指針是volatile型，保證讀操作時(shí)可以讀到最新的值。
static final class HashEntry<K,V> {
final int hash;
final K key;
volatile V value;
volatile HashEntry<K,V> next;
HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
this.hash = hash;
this.key = key;
this.value = value;
this.next = next;
}
final void setNext(HashEntry<K,V> n) {
UNSAFE.putOrderedObject(this, nextOffset, n);
}
// Unsafe mechanics
static final sun.misc.Unsafe UNSAFE;
static final long nextOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class k = HashEntry.class;
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
源碼分析
繼承關(guān)系
Concurrent繼承關(guān)系
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable {
- 繼承AbstractMap抽象類(lèi)，實(shí)現(xiàn)Map接口
- 實(shí)現(xiàn)java.io.Serialization接口，支持序列化
- 實(shí)現(xiàn)ConcurrentMap接口
初始化
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
//ssize為大于等于concurrencyLevel的最小的2的冪次方
int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
/**
* 用于定位Segment，假設(shè)ssize為2的n次方，我們可以發(fā)現(xiàn)sshift就是n。
* segmentShift為32-n，segmentMask為2^n-1，二進(jìn)制表示時(shí)各位數(shù)都是1
*/
this.segmentShift = 32 - sshift;//偏移量，定位參與hash運(yùn)算的位數(shù)
this.segmentMask = ssize - 1;//掩碼
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;//segment中HashEntry數(shù)組的長(zhǎng)度
while (cap < c)
cap <<= 1;
//創(chuàng)建segments和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;
}
//使用指定初始化容量、負(fù)載因子和默認(rèn)并發(fā)等級(jí)16，創(chuàng)建空ConcurrentHashMap
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, DEFAULT_CONCURRENCY_LEVEL);
}
//使用指定初始化容量、默認(rèn)負(fù)載因子0.75和默認(rèn)并發(fā)等級(jí)16，創(chuàng)建空ConcurrentHashMap
public ConcurrentHashMap(int initialCapacity) {
this(initialCapacity, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
//使用默認(rèn)初始化容量16、負(fù)載因子和默認(rèn)并發(fā)等級(jí)16，創(chuàng)建空ConcurrentHashMap
public ConcurrentHashMap() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this(Math.max((int) (m.size() / DEFAULT_LOAD_FACTOR) + 1,
DEFAULT_INITIAL_CAPACITY),
DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
putAll(m);
}
發(fā)現(xiàn)構(gòu)造函數(shù)最終調(diào)用的都是public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)這個(gè)方法，參數(shù)分別是DEFAULT_INITIAL_CAPACITY初始容量、DEFAULT_LOAD_FACTOR默認(rèn)負(fù)載因子以及DEFAULT_CONCURRENCY_LEVEL默認(rèn)并發(fā)級(jí)別，表示將ConcurrentHashMap分為幾個(gè)段。
插入
ConcurrentHashMap用分段Segment來(lái)保存數(shù)據(jù)，所以插入和獲取數(shù)據(jù)時(shí)都需要先定位到對(duì)應(yīng)的段上。
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)//value值不能為空
throw new NullPointerException();
int hash = hash(key);//求出key的hash值
/**
* segmentShift是32-sshift，將計(jì)算的hash值無(wú)符號(hào)右移segmentShift位，其實(shí)就是取hash值的高sshift位
* 然后與segmentMask進(jìn)行按位與操作
**/
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject
(segments, (j << SSHIFT) + SBASE)) == null) //找不到Segment新建一個(gè)
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
定位到對(duì)應(yīng)的段上后就需要調(diào)用segment的put方法。
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
/**
* tryLock方法嘗試去獲得鎖
* 獲取鎖失敗的話(huà)，調(diào)用scanAndLockForPut獲得鎖
**/
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;//當(dāng)前段的HashEntry類(lèi)型的數(shù)組
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);//定位到對(duì)應(yīng)的桶
for (HashEntry<K,V> e = first;;) {//遍歷HashEntry鏈表
if (e != null) {
K k;
/**
* 要插入的鍵值對(duì)已經(jīng)存在于HashEntry數(shù)組中
* onlyIfAbsent為true，表示只能插入不存在的鍵值對(duì)直接退出
* onlyIfAbsent為false，用新值替代舊值
**/
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 {//要插入的鍵值對(duì)不存在HashEntry數(shù)組中
if (node != null)//在scanAndLockForPut獲得鎖的過(guò)程中新建了對(duì)應(yīng)的節(jié)點(diǎn)
node.setNext(first);//將first節(jié)點(diǎn)設(shè)置成node的下一個(gè)節(jié)點(diǎn)
else
node = new HashEntry<K,V>(hash, key, value, first);//新建HashEntry鏈表
/**
* 如果插入鍵值對(duì)后，超過(guò)閾值，則進(jìn)行擴(kuò)容；
* 否則直接插入鍵值對(duì)
**/
int c = count + 1;
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();//釋放鎖
}
return oldValue;
}
put方法里首先需要先獲得Segment的互斥鎖，如果沒(méi)有獲得鎖的話(huà)會(huì)調(diào)用scanAndLockForPut這個(gè)自旋方法，介紹下這個(gè)方法具體做了什么。
//
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);//根據(jù)hash值獲取對(duì)應(yīng)的HashEntry鏈表
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1; //自旋計(jì)數(shù)器
while (!tryLock()) {
HashEntry<K,V> f; // to recheck first below
if (retries < 0) {
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) {
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
未完。。。。
