Java并发编程之ConcurrentHashMap

ConcurrentHashMap
ConcurrentHashMap是一个线程安全的Hash Table，它的主要功能是提供了一组和HashTable功能相同但是线程安全的方法。ConcurrentHashMap可以做到读取数据不加锁，并且其内部的结构可以让其在进行写操作的时候能够将锁的粒度保持地尽量地小，不用对整个ConcurrentHashMap加锁。

ConcurrentHashMap的内部结构
ConcurrentHashMap为了提高本身的并发能力，在内部采用了一个叫做Segment的结构，一个Segment其实就是一个类Hash Table的结构，Segment内部维护了一个链表数组，我们用下面这一幅图来看下ConcurrentHashMap的内部结构：

从上面的结构我们可以了解到，ConcurrentHashMap定位一个元素的过程需要进行两次Hash操作，第一次Hash定位到Segment，第二次Hash定位到元素所在的链表的头部，因此，这一种结构的带来的副作用是Hash的过程要比普通的HashMap要长，但是带来的好处是写操作的时候可以只对元素所在的Segment进行加锁即可，不会影响到其他的Segment，这样，在最理想的情况下，ConcurrentHashMap可以最高同时支持Segment数量大小的写操作（刚好这些写操作都非常平均地分布在所有的Segment上），所以，通过这一种结构，ConcurrentHashMap的并发能力可以大大的提高。

Segment
我们再来具体了解一下Segment的数据结构：

1
static final class Segment<K,V> extends ReentrantLock implements Serializable {
2
    transient volatile int count;
3
    transient int modCount;
4
    transient int threshold;
5
    transient volatile HashEntry<K,V>[] table;
6
    final float loadFactor;
7
}
详细解释一下Segment里面的成员变量的意义：

count：Segment中元素的数量
modCount：对table的大小造成影响的操作的数量（比如put或者remove操作）
threshold：阈值，Segment里面元素的数量超过这个值依旧就会对Segment进行扩容
table：链表数组，数组中的每一个元素代表了一个链表的头部
loadFactor：负载因子，用于确定threshold
HashEntry
Segment中的元素是以HashEntry的形式存放在链表数组中的，看一下HashEntry的结构：

1
static final class HashEntry<K,V> {
2
    final K key;
3
    final int hash;
4
    volatile V value;
5
    final HashEntry<K,V> next;
6
}
可以看到HashEntry的一个特点，除了value以外，其他的几个变量都是final的，这样做是为了防止链表结构被破坏，出现ConcurrentModification的情况。

ConcurrentHashMap的初始化
下面我们来结合源代码来具体分析一下ConcurrentHashMap的实现，先看下初始化方法：

01
public ConcurrentHashMap(int initialCapacity,
02
                         float loadFactor, int concurrencyLevel) {
03
    if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
04
        throw new IllegalArgumentException();
05

06
    if (concurrencyLevel > MAX_SEGMENTS)
07
        concurrencyLevel = MAX_SEGMENTS;
08

09
    // Find power-of-two sizes best matching arguments
10
    int sshift = 0;
11
    int ssize = 1;
12
    while (ssize < concurrencyLevel) {
13
        ++sshift;
14
        ssize <<= 1;
15
    }
16
    segmentShift = 32 - sshift;
17
    segmentMask = ssize - 1;
18
    this.segments = Segment.newArray(ssize);
19

20
    if (initialCapacity > MAXIMUM_CAPACITY)
21
        initialCapacity = MAXIMUM_CAPACITY;
22
    int c = initialCapacity / ssize;
23
    if (c * ssize < initialCapacity)
24
        ++c;
25
    int cap = 1;
26
    while (cap < c)
27
        cap <<= 1;
28

29
    for (int i = 0; i < this.segments.length; ++i)
30
        this.segments[i] = new Segment<K,V>(cap, loadFactor);
31
}
CurrentHashMap的初始化一共有三个参数，一个initialCapacity，表示初始的容量，一个loadFactor，表示负载参数，最后一个是concurrentLevel，代表ConcurrentHashMap内部的Segment的数量，ConcurrentLevel一经指定，不可改变，后续如果ConcurrentHashMap的元素数量增加导致ConrruentHashMap需要扩容，ConcurrentHashMap不会增加Segment的数量，而只会增加Segment中链表数组的容量大小，这样的好处是扩容过程不需要对整个ConcurrentHashMap做rehash，而只需要对Segment里面的元素做一次rehash就可以了。

整个ConcurrentHashMap的初始化方法还是非常简单的，先是根据concurrentLevel来new出Segment，这里Segment的数量是不大于concurrentLevel的最大的2的指数，就是说Segment的数量永远是2的指数个，这样的好处是方便采用移位操作来进行hash，加快hash的过程。接下来就是根据intialCapacity确定Segment的容量的大小，每一个Segment的容量大小也是2的指数，同样使为了加快hash的过程。

这边需要特别注意一下两个变量，分别是segmentShift和segmentMask，这两个变量在后面将会起到很大的作用，假设构造函数确定了Segment的数量是2的n次方，那么segmentShift就等于32减去n，而segmentMask就等于2的n次方减一。

ConcurrentHashMap的get操作
前面提到过ConcurrentHashMap的get操作是不用加锁的，我们这里看一下其实现：

1
public V get(Object key) {
2
int hash = hash(key.hashCode());
3
return segmentFor(hash).get(key, hash);
4
}
看第三行，segmentFor这个函数用于确定操作应该在哪一个segment中进行，几乎对ConcurrentHashMap的所有操作都需要用到这个函数，我们看下这个函数的实现：

1
final Segment<K,V> segmentFor(int hash) {
2
return segments[(hash >>> segmentShift) & segmentMask];
3
}
这个函数用了位操作来确定Segment，根据传入的hash值向右无符号右移segmentShift位，然后和segmentMask进行与操作，结合我们之前说的segmentShift和segmentMask的值，就可以得出以下结论：假设Segment的数量是2的n次方，根据元素的hash值的高n位就可以确定元素到底在哪一个Segment中。

在确定了需要在哪一个segment中进行操作以后，接下来的事情就是调用对应的Segment的get方法：

01
V get(Object key, int hash) {
02
    if (count != 0) { // read-volatile
03
        HashEntry<K,V> e = getFirst(hash);
04
        while (e != null) {
05
            if (e.hash == hash && key.equals(e.key)) {
06
                V v = e.value;
07
                if (v != null)
08
                    return v;
09
                return readValueUnderLock(e); // recheck
10
            }
11
            e = e.next;
12
        }
13
    }
14
补充：软件开发 , Java ,