java.util.concurrent.ConcurrentHashMap

一、简介

ConcurrentHashMap是Map的一种并发实现，在该类中元素的read操作都是无锁了，而write操作需要被同步。这非常适合于读操作远大于写操作的情况。在实现过程中，ConcurrentHashMap将所有元素分成了若干个segment，每个segment是独立的，在一个segment上加锁并不影响其他segment的操作。segment本身是一个hashtable，对于一个加入ConcurrentHashMap的<key, value>对，key的hash值中的高位被用来索引segment，而低位用于segment中的索引。

二、segment实现

segment是ConcurrentHashMap存储元素的基本段，它本身是一个hashtable的实现，read操作时无锁的，write需要同步，定义如下：

public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable {

/**
*  key, hash, next都是不可改的
*  value值可被重写
*/
static final class HashEntry<K,V> {
final K key;
final int hash;
volatile V value;
final HashEntry<K,V> next;

...
}

static final class Segment<K,V> extends ReentrantLock implements Serializable {

transient volatile int count;
transient volatile HashEntry[] table;
// 当segment中元素个数达到threshold时，需要rehash
transient int threshold;
}

...
}

segment的read操作：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

HashEntry<K,V> getFirst(int hash) {
HashEntry[] tab = table;
return (HashEntry<K,V>) tab[hash & (tab.length - 1)];
}

V get(Object key, int hash) { // 该操作是无锁的
if (count != 0) { // read-volatile
HashEntry<K,V> 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;
}

...

由于HashEntry当中的key和next都是final的，所以segment之上的操作不可能影响HashEntry列表之间相对的顺序，而value是可变的，当第一次读值失败时，尝试加锁读。

segment的replace操作：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

/**
* replace操作是就地替换，HashEntry的value是非final的
*/
boolean replace(K key, int hash, V oldValue, V newValue) {
lock();  // replace操作是同步的
try {
// 得到该hash值对应的entry列表
HashEntry<K,V> e = getFirst(hash);
while (e != null && (e.hash != hash || !key.equals(e.key)))
e = e.next;

boolean replaced = false;
if (e != null && oldValue.equals(e.value)) { // 替换
replaced = true;
e.value = newValue;
}
return replaced;
} finally {
unlock();
}
}

...
}

segmet的put操作：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

V put(K key, int hash, V value, boolean onlyIfAbsent) {
lock(); // put是同步的
try {
int c = count;
if (c++ > threshold) // ensure capacity
rehash();
HashEntry[] tab = table;
int index = hash & (tab.length - 1);
HashEntry<K,V> first = (HashEntry<K,V>) tab[index];
HashEntry<K,V> 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;
// HashEntry的next是只读的，新加入的entry只能放在头部
tab[index] = new HashEntry<K,V>(key, hash, first, value);
count = c; // write-volatile
}
return oldValue;
} finally {
unlock();
}
}

...
}

segment的remove操作一种copy on write 的方法，保留被删元素之后的列表，copy被删元素之前的hashEntry：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

V remove(Object key, int hash, Object value) {
lock();
try {
int c = count - 1;
HashEntry[] tab = table;
int index = hash & (tab.length - 1);
HashEntry<K,V> first = (HashEntry<K,V>)tab[index];
HashEntry<K,V> e = first;
while (e != null && (e.hash != hash || !key.equals(e.key)))
e = e.next;

V oldValue = null;
if (e != null) {
V v = e.value;
if (value == null || value.equals(v)) { // copy on write
oldValue = v;
++modCount;
// e之后的列表可以保留，只需要重新创建e之前的HashEntry即可
HashEntry<K,V> newFirst = e.next;
// copy on write e之前的HashEntry
for (HashEntry<K,V> p = first; p != e; p = p.next)
newFirst = new HashEntry<K,V>(p.key, p.hash,
newFirst, p.value);
tab[index] = newFirst;
count = c; // write-volatile
}
}
return oldValue;
} finally {
unlock();
}
}

...
}

segment的rehash操作实现比较特别，为了保证rehash过程中copy的元素尽可能少，segment在rehash时Entry入口的个数是以2的倍数增长，这可以保证一个entry在rehash之后要么在原来的列表中，要么在下一个列表中：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

void rehash() {
// 局部变量引用table
HashEntry[] oldTable = table;
int oldCapacity = oldTable.length;
if (oldCapacity >= MAXIMUM_CAPACITY)
return;

// 右移1位相当于乘以2
HashEntry[] newTable = new HashEntry[oldCapacity << 1];
threshold = (int)(newTable.length * loadFactor);
int sizeMask = newTable.length - 1;
for (int i = 0; i < oldCapacity ; i++) {
// 第i个entry列表
HashEntry<K,V> e = (HashEntry<K,V>)oldTable[i];

if (e != null) {
HashEntry<K,V> next = e.next;
// 在新table上的索引
int idx = e.hash & sizeMask;

if (next == null)
newTable[idx] = e;
else {
// 寻找该entry列表末端，rehash之后idx相同的元素
// 这些元素不需要被copy
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;
}
}
// 将lastRun之后的整个列表挂到新位置上
newTable[lastIdx] = lastRun;

// Clone all remaining nodes
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
int k = p.hash & sizeMask;
HashEntry<K,V> n = (HashEntry<K,V>)newTable[k];
newTable[k] = new HashEntry<K,V>(p.key, p.hash,
n, p.value);
}
}
}
}
table = newTable;
}

...
}

三、ConcurrentHashMap方法实现

ConcurrentHashMap在Segment的基础上，通过首先将<key, value>对hash到一个segment，再由segment实现对entry的管理。

ConcurrentHashMap的get实现：

public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable {

final Segment<K,V> segmentFor(int hash) {
return (Segment<K,V>) segments[(hash >>> segmentShift) & segmentMask];
}

public V get(Object key) {
int hash = hash(key); // throws NullPointerException if key null
return segmentFor(hash).get(key, hash);
}

...
}

ConcurrentHashMap的put和get方法：

public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable {

public V put(K key, V value) {
if (value == null)
throw new NullPointerException();
int hash = hash(key);
return segmentFor(hash).put(key, hash, value, false);
}

public V remove(Object key) {
int hash = hash(key);
return segmentFor(hash).remove(key, hash, null);
}

...
}