JDK源码（二）ConcurrentHashMap-JDK1.7

1.背景

并发编程中，ConcurrentHashMap是一个使用度非常高的数据结构。

优点:

线程安全
相比于HashTable和Collections.synchronizedMap()效率高，使用了分段锁技术。

2.ConcurrentHashMap数据结构

Segment

Segment继承了ReentrantLock，所以它本身是个容器同时也是一个锁。所以segment中可以使用tryLock()，lock()，unLock()方法完成锁相关操作。

HashEntry

 static final class HashEntry<K,V> {final int hash;final K key;volatile V value;volatile HashEntry<K,V> next;}

单向链表，其中持有了键值对和key的hash值。

3.关键方法和实现原理

3.1初始化方法

 /** * @param initialCapacity 初始容量 * @param loadFactor 加载因子,扩容时机 * @param concurrencyLevel 并发级别 */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;int sshift = 0;// segments[]数组大小,假设并发度为17,ssize为32.int ssize = 1;while (ssize < concurrencyLevel) {++sshift;ssize <<= 1;}// segment偏移量this.segmentShift = 32 - sshift;// segmentMask:二进制表示111111,用于定位segments[]下标.this.segmentMask = ssize - 1;if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;// 计算每个sgment中HashEntry[]数组的大小int c = initialCapacity / ssize;if (c * ssize < initialCapacity)++c;int cap = MIN_SEGMENT_TABLE_CAPACITY;while (cap < c)cap <<= 1;// create segments and 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给ss数组的第0个元素赋值UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]this.segments = ss;}

3.2put()方法

 public V put(K key, V value) {Segment<K,V> s;if (value == null)throw new NullPointerException();// 计算key的hash值int hash = hash(key);// 将hash值右移偏移量位,并与上31(11111),所以j为0-31之间的数int j = (hash >>> segmentShift) & segmentMask;// 获取下标为j的segment的对象,如果未创建则用UNSAFE提供的CAS操作创建segment对象。并保证多个线程同时创建的正确性。if ((s = (Segment<K,V>)UNSAFE.getObject// nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) //in ensureSegments = ensureSegment(j);return s.put(key, hash, value, false);}/** * 并发情况下的ensureSegment()方法也是线程安全的 **/private Segment<K,V> ensureSegment(int k) {final Segment<K,V>[] ss = this.segments;long u = (k << SSHIFT) + SBASE; // raw offsetSegment<K,V> seg;// 使用轻量级同步volatile原语,保证读到的对象是最新的.if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {Segment<K,V> proto = ss[0]; // use segment 0 as prototypeint cap = proto.table.length;float lf = proto.loadFactor;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) { // recheckSegment<K,V> s = new Segment<K,V>(lf, threshold, tab);while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {// CAS操作if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))break;}}}return seg;}/*** segment中的put()方法*/final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 尝试获取锁,获取不到时,调用scan..预先创建节点并返回(有点自旋锁的意味)。HashEntry<K,V> node = tryLock() ? null :scanAndLockForPut(key, hash, value);V oldValue;try {HashEntry<K,V>[] tab = table;// 计算此key在HashEntry[]数组的下标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);elsenode = new HashEntry<K,V>(hash, key, value, first);int c = count + 1;if (c > threshold && tab.length < MAXIMUM_CAPACITY)// 超过阈值,扩容rehash(node);elsesetEntryAt(tab, index, node);++modCount;count = c;oldValue = null;break;}}} finally {unlock();}return oldValue;}/** * 扩容方法,因为扩容过后,每个节点的下标只会(不变)或者变为(现有下标+原数组长度),所以它遍历链表时将最后一小段要变化的链一起移动。 */private void rehash(HashEntry<K,V> node) {HashEntry<K,V>[] oldTable = table;// 原数组长度int oldCapacity = oldTable.length;// 扩容后长度*2int newCapacity = oldCapacity << 1;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) //Single node on listnewTable[idx] = e;else { // Reuse consecutive sequence at same slotHashEntry<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;// Clone remaining nodesfor (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; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] = node;table = newTable;}

3.3 get(Object key)方法

 // get方法无需加锁，由于其中涉及到的共享变量都使用volatile修饰，volatile可以保证内存可见性，所以不会读取到过期数据。public V get(Object key) {Segment<K,V> s; // manually integrate access methods to reduce overheadHashEntry<K,V>[] tab;int h = hash(key);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;}

3.4 remove(Object key)方法

remove()的目的就是删除key-value键值对。> 在删除之前，它会获取到Segment的互斥锁，在删除之后，再释放锁。它的删除过程也比较简单，它会先根据hash值，找到“Segment的HashEntry数组”中对应的“HashEntry”节点。根据Segment的数据结构，我们知道Segment中包含一个HashEntry数组对象，而每一个HashEntry本质上是一个单向链表。在找到“HashEntry”节点之后，就遍历该“HashEntry”节点对应的链表，找到key-value键值对对应的节点，然后删除。

3.5 size()方法

如果我们要统计整个ConcurrentHashMap里元素的大小，就必须统计所有Segment里元素的大小后求和。Segment里的全局变量count是一个volatile变量，那么在多线程场景下，我们是不是直接把所有Segment的count相加就可以得到整个ConcurrentHashMap大小了呢？不是的，虽然相加时可以获取每个Segment的count的最新值，但是拿到之后可能累加前使用的count发生了变化，那么统计结果就不准了。所以最安全的做法，是在统计size的时候把所有Segment的put，remove和clean方法全部锁住，但是这种做法显然非常低效。

因为在累加count操作过程中，之前累加过的count发生变化的几率非常小，所以ConcurrentHashMap的做法是先尝试2次通过不锁住Segment的方式来统计各个Segment大小，如果统计的过程中，容器的count发生了变化，则再采用加锁的方式来统计所有Segment的大小。

那么ConcurrentHashMap是如何判断在统计的时候容器是否发生了变化呢？使用modCount变量，在put , remove和clean方法里操作元素前都会将变量modCount进行加1，那么在统计size前后比较modCount是否发生变化，从而得知容器的大小是否发生变化。

4.总结

ConcurrentHashMap是线程安全的哈希表，它是通过分段锁来实现的。put和remove时都要先获取互斥锁。而对于读取操作，它是通过volatile去实现的，HashEntry数组是volatile类型的，HashEntry中的value,next属性也是volatile类型的。而volatile能保证可见性和有序性。以上这些方式，就是ConcurrentHashMap线程安全的实现原理。