Java 7之多线程并发容器 - ConcurrentHashMap

与HashMap一样，ConcurrentHashMap也是一个基于散列的Map，但是它使用了锁分段的技术来提供更高的并发性和伸缩性。

锁分段就是进一步对一组独立的对象进行分解。例如，在ConcurrentHashMap的实现中使用了一个包含16个锁的数组，每个锁保护所有散列桶的1/16，其实第N个散列桶由第(N mod 16)个锁来保护。所以这个并发集合可以支持多达16个并发的写入器。

首先举个例子：

[java]
view plain
copy

print?





public class StripedMap {  
    // Synchronization policy: buckets
 guarded by locks[n%N_LOCKS]  
    private static final int N_LOCKS = 16;  // 并发锁的数量  
    private final Node[] buckets;           // 散列桶  
    private final Object[] locks;           // 锁数组  
  
    private static class Node {             // 链表中的节点  
        Node next;  
        Object key;  
        Object value;  
    }  
  
    public StripedMap(int numBuckets) {    // 构造函数  
        buckets = new Node[numBuckets];  
        locks = new Object[N_LOCKS];  
        for (int i = 0; i < N_LOCKS; i++)  
            locks[i] = new Object();  
    }  
  
    private final int hash(Object key) {    // 计算值的存储位置，相当于散列函数  
        return Math.abs(key.hashCode() % buckets.length);  
    }  
  
    public Object get(Object key) {  
        int hash = hash(key);  
        synchronized (locks[hash % N_LOCKS]) {                   // 计算出由哪个锁来保护这个散列桶  
            for (Node m = buckets[hash]; m != null; m = m.next)  // 遍历这个散列桶，找到需要的值  
                if (m.key.equals(key))  
                    return m.value;  
        }  
        return null;  
    }  
  
    public void clear() {  
        for (int i = 0; i < buckets.length; i++) {  
            synchronized (locks[i % N_LOCKS]) {                 // 将锁分段中的值清空  
                buckets[i] = null;  
            }  
        }  
    }  
}  

public class StripedMap {
// Synchronization policy: buckets
guarded by locks[n%N_LOCKS]
private static final int N_LOCKS = 16;  // 并发锁的数量
private final Node[] buckets;           // 散列桶
private final Object[] locks;           // 锁数组

private static class Node {             // 链表中的节点
Node next;
Object key;
Object value;
}

public StripedMap(int numBuckets) {    // 构造函数
buckets = new Node[numBuckets];
locks = new Object[N_LOCKS];
for (int i = 0; i < N_LOCKS; i++)
locks[i] = new Object();
}

private final int hash(Object key) {    // 计算值的存储位置，相当于散列函数
return Math.abs(key.hashCode() % buckets.length);
}

public Object get(Object key) {
int hash = hash(key);
synchronized (locks[hash % N_LOCKS]) {                   // 计算出由哪个锁来保护这个散列桶
for (Node m = buckets[hash]; m != null; m = m.next)  // 遍历这个散列桶，找到需要的值
if (m.key.equals(key))
return m.value;
}
return null;
}

public void clear() {
for (int i = 0
4000
; i < buckets.length; i++) {
synchronized (locks[i % N_LOCKS]) {                 // 将锁分段中的值清空
buckets[i] = null;
}
}
}
}

如上使用了锁分段技术简单实现了一个Map并发容器，但是与采用单个锁来实现独占访问相比，要获取多个锁来实现独占访问将更加困难并且开销更高。例如有些方法需要跨段，如size()和containsValue()，它们可能需要锁定整个表而而不仅仅是某个段，这需要按顺序锁定所有段，操作完毕后，又按顺序释放所有段的锁。这里“按顺序”是很重要的，否则极有可能出现死锁，在ConcurrentHashMap内部，段数组是final的，并且其成员变量实际上也是final的，但是，仅仅是将数组声明为final的并不保证数组成员也是final的，这需要实现上的保证。这可以确保不会出现死锁，因为获得锁的顺序是固定的。不变性是多线程编程占有很重要的地位。

[java]
view plain
copy

print?





final Segment<K,V>[] segments;  // 段数组为final类型的  

final Segment<K,V>[] segments;  // 段数组为final类型的

实现代码如下：

[java]
view plain
copy

print?





static final class Segment<K,V> extends ReentrantLock implements Serializable {  
    transient volatile HashEntry<K,V>[] table;  
    // threshold阈，是哈希表在其容量自动增加之前可以达到多满的一种尺度。  
    transient int threshold;  
    // loadFactor是加载因子  
    final float loadFactor;  
  
    Segment(float lf, int threshold, HashEntry<K,V>[] tab) {  
        this.loadFactor = lf;  
        this.threshold = threshold;  
        this.table = tab;  
    }  
    // 省略部分代码  
}  

static final class Segment<K,V> extends ReentrantLock implements Serializable {
transient volatile HashEntry<K,V>[] table;
// threshold阈，是哈希表在其容量自动增加之前可以达到多满的一种尺度。
transient int threshold;
// loadFactor是加载因子
final float loadFactor;

Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
this.loadFactor = lf;
this.threshold = threshold;
this.table = tab;
}
// 省略部分代码
}

在每个Segment中通过HashEntry来表示链结构，类似于前面例子中的Node节点，主要的代码如下：

[java]
view plain
copy

print?





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;  
       }  
       // 省略部分代码  
   }  

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;
}
// 省略部分代码
}

可以看到除了value不是final的，其它值都是final的，这意味着不能从链表的中间或尾部添加或删除节点，因为这需要修改next 引用值，所有的节点的修改只能从头部开始。

首先来看一下ConcurrentHashMap中最主要的一个构造函数，如下：

[java]
view plain
copy

print?





public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {  
        // 参数有效性判断  
        if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)  
            throw new IllegalArgumentException();  
        // concurrencyLevel是用来计算segments的容量  
        if (concurrencyLevel > MAX_SEGMENTS)  
            concurrencyLevel = MAX_SEGMENTS;  
        int sshift = 0;  
        int ssize = 1;  
        // ssize是大于或等于concurrencyLevel的最小的2的N次方值  
        while (ssize < concurrencyLevel) {  
            ++sshift;  
            ssize <<= 1;  
        }  
        // 初始化segmentShift和segmentMask  
        this.segmentShift = 32 - sshift;  
        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; // segments中的HashEntry数组的长度  
        while (cap < c)  
            cap <<= 1;  
        // segments  
        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;  
    }  

public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {
// 参数有效性判断
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
// concurrencyLevel是用来计算segments的容量
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
int sshift = 0;
int ssize = 1;
// ssize是大于或等于concurrencyLevel的最小的2的N次方值
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
// 初始化segmentShift和segmentMask
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
// 哈希表的初始容量
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize; // 计算哈希表的实际容量
if (c * ssize < initialC
1081f
apacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY; // segments中的HashEntry数组的长度
while (cap < c)
cap <<= 1;
// segments
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;
}

1、获取元素

[java]
view plain
copy

print?





public V get(Object key) {  
    Segment<K,V> s; // manually integrate access methods to reduce overhead  
    HashEntry<K,V>[] tab;  
    int h = hash(key.hashCode());  
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;  
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&(tab = s.table) != null) {  
        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;  
}  

public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key.hashCode());
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&(tab = s.table) != null) {
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;
}

如上获取元素的操作是不带锁的，效率会提高。

2、添加元素

[java]
view plain
copy

print?





public V put(K key, V value) {  
       Segment<K,V> s;  
       if (value == null)  
           throw new NullPointerException();  
       int hash = hash(key.hashCode());  
       int j = (hash >>> segmentShift) & segmentMask;  
       if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck  
            (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment  
           s = ensureSegment(j);  
       return s.put(key, hash, value, false);  
   }  

public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key.hashCode());
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
s = ensureSegment(j);
return s.put(key, hash, value, false);
}

接着调用了Segment类中的put方法将元素添加到链表中，如下：

[java]
view plain
copy

print?





final V put(K key, int hash, V value, boolean onlyIfAbsent) {  
           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);  
               for (HashEntry<K,V> e = first;;) {  
                   if (e != null) {  
                       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.setNext(first);  
                       else  
                           node = new HashEntry<K,V>(hash, key, value, first);  
                       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;  
       }  

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
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);
for (HashEntry<K,V> e = first;;) {
if (e != null) {
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.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
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;
}

3、删除元素

[java]
view plain
copy

print?





public V remove(Object key) {  
    int hash = hash(key);  
    // 根据hash值，找到key对应的Segment片段  
    Segment<K,V> s = segmentForHash(hash);  
    return s == null ? null : s.remove(key, hash, null);  
}  

public V remove(Object key) {
int hash = hash(key);
// 根据hash值，找到key对应的Segment片段
Segment<K,V> s = segmentForHash(hash);
return s == null ? null : s.remove(key, hash, null);
}

整个操作是先定位到段，然后委托给段的remove操作。当多个删除操作并发进行时，只要它们所在的段不相同，它们就可以同时进行。下面是Segment的remove方法实现：

[java]
view plain
copy

print?





final V remove(Object key, int hash, Object value) {  
           if (!tryLock())  
               scanAndLock(key, hash);  
           V oldValue = null;  
           try {  
               HashEntry<K,V>[] tab = table;          //  volatile类型的table赋值给一个局部的变量  
               int index = (tab.length - 1) & hash;  
               HashEntry<K,V> e = entryAt(tab, index);// 获取tab中index索引处的链表  
               HashEntry<K,V> pred = null;  
               while (e != null) {  
                   K k;  
                   HashEntry<K,V> next = e.next;  
                   if ((k = e.key) == key ||(e.hash == hash && key.equals(k))) {// 找到了对应的key  
                       V v = e.value;  
                       if (value == null || value == v || value.equals(v)) {  
                           if (pred == null)  
                               setEntryAt(tab, index, next);  
                           else  
                               pred.setNext(next);  
                           ++modCount;     // 修改结构的次数加1  
                           --count;        // 元素数量送去1  
                           oldValue = v;   // 记录原始值  
                       }  
                       break;  
                   }// end if  
                   pred = e;  
                   e = next;  
               }  
           } finally {  
               unlock();  
           }  
           return oldValue;  
       }  

final V remove(Object key, int hash, Object value) {
if (!tryLock())
scanAndLock(key, hash);
V oldValue = null;
try {
HashEntry<K,V>[] tab = table;          //  volatile类型的table赋值给一个局部的变量
int index = (tab.length - 1) & hash;
HashEntry<K,V> e = entryAt(tab, index);// 获取tab中index索引处的链表
HashEntry<K,V> pred = null;
while (e != null) {
K k;
HashEntry<K,V> next = e.next;
if ((k = e.key) == key ||(e.hash == hash && key.equals(k))) {// 找到了对应的key
V v = e.value;
if (value == null || value == v || value.equals(v)) {
if (pred == null)
setEntryAt(tab, index, next);
else
pred.setNext(next);
++modCount;     // 修改结构的次数加1
--count;        // 元素数量送去1
oldValue = v;   // 记录原始值
}
break;
}// end if
pred = e;
e = next;
}
} finally {
unlock();
}
return oldValue;
}

整个remove实现并不复杂，但是需要注意如下几点。

第一，当要删除的结点存在时，删除的最后一步操作要将count的值减一。这必须是最后一步操作，否则读取操作可能看不到之前对段所做的结构性修改。

第二，remove执行的开始就将table赋给一个局部变量tab，这是因为table是 volatile变量，读写volatile变量的开销很大。编译器也不能对volatile变量的读写做任何优化，直接多次访问非volatile实例变量没有多大影响，编译器会做相应优化。

4、迭代元素

转载自http://blog.csdn.net/mazhimazh/