Java中的ConcurrentHashMap详解编程语言

ConcurrentHashMap是JDK1.5并发包中提供的线程安全的HashMap的实现，ConcurrentHashMap 的实现是依赖于 Java 内存模型。

效率低下的HashTable容器

HashTable容器使用synchronized来保证线程安全，但在线程竞争激烈的情况下HashTable的效率非常低下。因为当一个线程访问HashTable的同步方法时，其他线程访问HashTable的同步方法时，可能会进入阻塞或轮询状态。如线程1使用put进行添加元素，线程2不但不能使用put方法添加元素，并且也不能使用get方法来获取元素，所以竞争越激烈效率越低。

锁分段技术

HashTable容器在竞争激烈的并发环境下表现出效率低下的原因是所有访问HashTable的线程都必须竞争同一把锁，那假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效的提高并发访问效率，这就是ConcurrentHashMap所使用的锁分段技术，首先将数据分成一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也能被其他线程访问。

实现原理：

锁分离 (Lock Stripping)

ConcurrentHashMap允许多个修改操作并发进行，其关键在于使用了锁分离技术。它使用了多个锁来控制对hash表的不同部分进行的修改。ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分，其中每一个片段是一个类似于HashMap的结构，它有一个HashEntry的数组，数组的每一项又是一个链表，通过HashEntry的next引用串联起来，它们有自己的锁。

有些方法需要跨段，比如size()和containsValue()，它们可能需要锁定整个表而而不仅仅是某个段，这需要按顺序锁定所有段，操作完毕后，又按顺序释放所有段的锁。这里“按顺序”是很重要的，否则极有可能出现死锁，在ConcurrentHashMap内部，段数组是final的，并且其成员变量实际上也是final的，但是，仅仅是将数组声明为final的并不保证数组成员也是final的，这需要实现上的保证。这可以确保不会出现死锁，因为获得锁的顺序是固定的。

不变(Immutable)和易变(Volatile)

ConcurrentHashMap完全允许多个读操作并发进行，读操作并不需要加锁。如果使用传统的技术，如HashMap中的实现，如果允许可以在hash链的中间添加或删除元素，读操作不加锁将得到不一致的数据。ConcurrentHashMap实现技术是保证HashEntry几乎是不可变的。HashEntry代表每个hash链中的一个节点。

HashEntry 的学习可以类比着 HashMap 中的 Entry。我们的存储键值对的过程中，散列的时候如果发生“碰撞”，将采用“分离链表法”来处理碰撞：把碰撞的 HashEntry 对象链接成一个链表。

如下图，我们在一个空桶中插入 A、B、C 两个 HashEntry 对象后的结构图（其实应该为键值对，在这进行了简化以方便更容易理解）：

ConcurrentHashMap：

public class ConcurrentHashMap K, V extends AbstractMap K, V 

 implements ConcurrentMap K, V , Serializable { 

 /** 

 * Mask value for indexing into segments. The upper bits of a 

 * keys hash code are used to choose the segment. 

 final int segmentMask; 

 /** 

 * Shift value for indexing within segments. 

 final int segmentShift; 

 /** 

 * The segments, each of which is a specialized hash table 

 final Segment K,V [] segments; 

} 

所有的成员都是final的，其中segmentMask和segmentShift主要是为了定位段。

ConcurrentHashMap的内部结构：

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

Segment

Segment 的类定义为static final class Segment K,V extends ReentrantLock implements Serializable。其继承于 ReentrantLock 类，从而使得 Segment 对象可以充当锁的角色。Segment 中包含HashEntry 的数组，其可以守护其包含的若干个桶（HashEntry的数组）。Segment 在某些意义上有点类似于 HashMap了，都是包含了一个数组，而数组中的元素可以是一个链表。

table:table 是由 HashEntry 对象组成的数组如果散列时发生碰撞，碰撞的 HashEntry 对象就以链表的形式链接成一个链表table数组的数组成员代表散列映射表的一个桶每个 table 守护整个 ConcurrentHashMap 包含桶总数的一部分如果并发级别为 16，table 则守护 ConcurrentHashMap 包含的桶总数的 1/16。

count 变量是计算器，表示每个 Segment 对象管理的 table 数组（若干个 HashEntry 的链表）包含的HashEntry 对象的个数。之所以在每个Segment对象中包含一个 count 计数器，而不在 ConcurrentHashMap 中使用全局的计数器，是为了避免出现“热点域”而影响并发性。

 span * Segments are specialized versions of hash tables. This 

 * subclasses from ReentrantLock opportunistically, just to 

 * simplify some locking and avoid separate construction. 

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

 /** 

 * The per-segment table. Elements are accessed via 

 * entryAt/setEntryAt providing volatile semantics. 

 transient volatile HashEntry K,V [] table; 

 /** 

 * The number of elements. Accessed only either within locks 

 * or among other volatile reads that maintain visibility. 

 transient int count; 

 transient int modCount; 

 /** 

 * 装载因子 

 final float loadFactor; 

 }

详细解释一下Segment里面的成员变量的意义：

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

我们通过下图来展示一下插入 ABC 三个节点后，Segment 的示意图：

HashEntry

Segment中的元素是以HashEntry的形式存放在链表数组中的，看一下HashEntry的结构：

 span static final class HashEntry K,V { 

 final K key; 

 final int hash; 

 volatile V value; 

 final HashEntry K,V next; 

}

可以看到HashEntry的一个特点，除了value以外，其他的几个变量都是final的，这样做是为了防止链表结构被破坏，出现ConcurrentModification的情况。
ConcurrentHashMap

ConcurrentHashMap 的结构中包含的 Segment 的数组，在默认的并发级别会创建包含 16 个 Segment 对象的数组。通过我们上面的知识，我们知道每个 Segment 又包含若干个散列表的桶，每个桶是由 HashEntry 链接起来的一个链表。如果 key 能够均匀散列，每个 Segment 大约守护整个散列表桶总数的 1/16。

为了加快定位段以及段中hash槽的速度，每个段hash槽的的个数都是2^n，这使得通过位运算就可以定位段和段中hash槽的位置。当并发级别为默认值16时，也就是段的个数，hash值的高4位决定分配在哪个段中。但是我们也不要忘记《算法导论》给我们的教训：hash槽的的个数不应该是2^n，这可能导致hash槽分配不均，这需要对hash值重新再hash一次。

这是重新hash的算法。

 span private static int hash(int h) { 

 // Spread bits to regularize both segment and index locations, 

 // using variant of single-word Wang/Jenkins hash. 

 h += (h 15) ^ 0xffffcd7d; 

 h ^= (h 10); 

 h += (h 3); 

 h ^= (h 6); 

 h += (h 2) + (h 14); 

 return h ^ (h 16); 

} 

这是定位段的方法：

 span final Segment K,V segmentFor(int hash) { 

 return segments[(hash segmentShift) segmentMask]; 

} 

ConcurrentHashMap的初始化操作：

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; 

 // Find power-of-two sizes best matching arguments 

 int sshift = 0; 

 int ssize = 1; 

 while (ssize concurrencyLevel) { 

 ++sshift; 

 ssize = 1; 

 segmentShift = 32 - sshift; 

 segmentMask = ssize - 1; 

 this.segments = Segment.newArray(ssize); 

 if (initialCapacity MAXIMUM_CAPACITY) 

 initialCapacity = MAXIMUM_CAPACITY; 

 int c = initialCapacity / ssize; 

 if (c * ssize initialCapacity) 

 ++c; 

 int cap = 1; 

 while (cap c) 

 cap = 1; 

 for (int i = 0; i this.segments.length; ++i) 

 this.segments[i] = new Segment K,V (cap, loadFactor); 

}

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次方减一。

获取操作get

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

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

看第三行，segmentFor这个函数用于确定操作应该在哪一个segment中进行，几乎对ConcurrentHashMap的所有操作都需要用到这个函数，我们看下这个函数的实现：

final Segment K,V segmentFor(int hash) { 

 return segments[(hash segmentShift) segmentMask]; 

}

这个函数用了位操作来确定Segment，根据传入的hash值向右无符号右移segmentShift位，然后和segmentMask进行与操作，结合我们之前说的segmentShift和segmentMask的值，就可以得出以下结论：假设Segment的数量是2的n次方，根据元素的hash值的高n位就可以确定元素到底在哪一个Segment中。

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

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; 

}

先看第二行代码，这里对count进行了一次判断，其中count表示Segment中元素的数量，我们可以来看一下count的定义：

transient volatile int count;

 可以看到count是volatile的，实际上这里里面利用了volatile的语义：

  写道

对volatile字段的写入操作happens-before于每一个后续的同一个字段的读操作。

因为实际上put、remove等操作也会更新count的值，所以当竞争发生的时候，volatile的语义可以保证写操作在读操作之前，也就保证了写操作对后续的读操作都是可见的，这样后面get的后续操作就可以拿到完整的元素内容。

然后，在第三行，调用了getFirst()来取得链表的头部：

HashEntry K,V getFirst(int hash) { 

 HashEntry K,V [] tab = table; 

 return tab[hash (tab.length - 1)]; 

}

同样，这里也是用位操作来确定链表的头部，hash值和HashTable的长度减一做与操作，最后的结果就是hash值的低n位，其中n是HashTable的长度以2为底的结果。

在确定了链表的头部以后，就可以对整个链表进行遍历，看第4行，取出key对应的value的值，如果拿出的value的值是null，则可能这个key，value对正在put的过程中，如果出现这种情况，那么就加锁来保证取出的value是完整的，如果不是null，则直接返回value。

接下来看put操作，同样地put操作也是委托给段的put方法。

put操作的前面也是确定Segment的过程，这里不再赘述，直接看关键的segment的put方法：

 span V put(K key, int hash, V value, boolean onlyIfAbsent) { 

 lock(); 

 try { 

 int c = count; 

 if (c++ threshold) // ensure capacity 

 rehash(); 

 HashEntry K,V [] tab = table; 

 int index = hash (tab.length - 1); 

 HashEntry K,V first = 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; 

 tab[index] = new HashEntry K,V (key, hash, first, value); 

 count = c; // write-volatile 

 return oldValue; 

 } finally { 

 unlock(); 

}

首先对Segment的put操作是加锁完成的，然后在第五行，如果Segment中元素的数量超过了阈值（由构造函数中的loadFactor算出）这需要进行对Segment扩容，并且要进行rehash，关于rehash的过程大家可以自己去了解，这里不详细讲了。

第8和第9行的操作就是getFirst的过程，确定链表头部的位置。

第11行这里的这个while循环是在链表中寻找和要put的元素相同key的元素，如果找到，就直接更新更新key的value，如果没有找到，则进入21行这里，生成一个新的HashEntry并且把它加到整个Segment的头部，然后再更新count的值。

remove操作：

Remove操作的前面一部分和前面的get和put操作一样，都是定位Segment的过程，然后再调用Segment的remove方法：

V remove(Object key, int hash, Object value) { 

 lock(); 

 try { 

 int c = count - 1; 

 HashEntry K,V [] tab = table; 

 int index = hash (tab.length - 1); 

 HashEntry K,V first = 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)) { 

 oldValue = v; 

 // All entries following removed node can stay 

 // in list, but all preceding ones need to be 

 // cloned. 

 ++modCount; 

 HashEntry K,V newFirst = e.next; 

 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(); 

}

首先remove操作也是确定需要删除的元素的位置，不过这里删除元素的方法不是简单地把待删除元素的前面的一个元素的next指向后面一个就完事了，我们之前已经说过HashEntry中的next是final的，一经赋值以后就不可修改，在定位到待删除元素的位置以后，程序就将待删除元素前面的那一些元素全部复制一遍，然后再一个一个重新接到链表上去，看一下下面这一幅图来了解这个过程：





假设链表中原来的元素如上图所示，现在要删除元素3，那么删除元素3以后的链表就如下图所示：

修改操作还有putAll和replace。putAll就是多次调用put方法，没什么好说的。replace甚至不用做结构上的更改，实现要比put和delete要简单得多，理解了put和delete，理解replace就不在话下了，这里也不介绍了。

另一个操作是containsKey，这个实现就要简单得多了，因为它不需要读取值：

boolean containsKey(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)) return true; e = e.next; } } return false; } 

 

跨段操作 

 

有些操作需要涉及到多个段，比如说size(), containsValaue()。先来看下size()方法：

 span public int size() { 

 final Segment K,V [] segments = this.segments; 

 long sum = 0; 

 long check = 0; 

 int[] mc = new int[segments.length]; 

 // Try a few times to get accurate count. On failure due to 

 // continuous async changes in table, resort to locking. 

 for (int k = 0; k RETRIES_BEFORE_LOCK; ++k) { 

 check = 0; 

 sum = 0; 

 int mcsum = 0; 

 for (int i = 0; i segments.length; ++i) { 

 sum += segments[i].count; 

 mcsum += mc[i] = segments[i].modCount; 

 if (mcsum != 0) { 

 for (int i = 0; i segments.length; ++i) { 

 check += segments[i].count; 

 if (mc[i] != segments[i].modCount) { 

 check = -1; // force retry 

 break; 

 if (check == sum) 

 break; 

 if (check != sum) { // Resort to locking all segments 

 sum = 0; 

 for (int i = 0; i segments.length; ++i) 

 segments[i].lock(); 

 for (int i = 0; i segments.length; ++i) 

 sum += segments[i].count; 

 for (int i = 0; i segments.length; ++i) 

 segments[i].unlock(); 

 if (sum Integer.MAX_VALUE) 

 return Integer.MAX_VALUE; 

 else 

 return (int)sum; 

} 

size方法主要思路是先在没有锁的情况下对所有段大小求和，如果不能成功（这是因为遍历过程中可能有其它线程正在对已经遍历过的段进行结构性更新），最多执行RETRIES_BEFORE_LOCK次，如果还不成功就在持有所有段锁的情况下再对所有段大小求和。在没有锁的情况下主要是利用Segment中的modCount进行检测，在遍历过程中保存每个Segment的modCount，遍历完成之后再检测每个Segment的modCount有没有改变，如果有改变表示有其它线程正在对Segment进行结构性并发更新，需要重新计算。

其实这种方式是存在问题的，在第一个内层for循环中，在这两条语句sum += segments[i].count; mcsum += mc[i] = segments[i].modCount;之间，其它线程可能正在对Segment进行结构性的修改，导致segments[i].count和segments[i].modCount读取的数据并不一致。这可能使size()方法返回任何时候都不曾存在的大小，很奇怪javadoc居然没有明确标出这一点，可能是因为这个时间窗口太小了吧。size()的实现还有一点需要注意，必须要先segments[i].count，才能segments[i].modCount，这是因为segment[i].count是对volatile变量的访问，接下来segments[i].modCount才能得到几乎最新的值（前面我已经说了为什么只是“几乎”了）。这点在containsValue方法中得到了淋漓尽致的展现：

 span public boolean containsValue(Object value) { if (value == null) throw new NullPointerException(); // See explanation of modCount use above final Segment K,V [] segments = this.segments; int[] mc = new int[segments.length]; // Try a few times without locking for (int k = 0; k RETRIES_BEFORE_LOCK; ++k) { int sum = 0; int mcsum = 0; for (int i = 0; i segments.length; ++i) { int c = segments[i].count; mcsum += mc[i] = segments[i].modCount; if (segments[i].containsValue(value)) return true; } boolean cleanSweep = true; if (mcsum != 0) { for (int i = 0; i segments.length; ++i) { int c = segments[i].count; if (mc[i] != segments[i].modCount) { cleanSweep = false; break; } } } if (cleanSweep) return false; } // Resort to locking all segments for (int i = 0; i segments.length; ++i) segments[i].lock(); boolean found = false; try { for (int i = 0; i segments.length; ++i) { if (segments[i].containsValue(value)) { found = true; break; } } } finally { for (int i = 0; i segments.length; ++i) segments[i].unlock(); } return found; } 

同样注意内层的第一个for循环，里面有语句int c = segments[i].count; 但是c却从来没有被使用过，即使如此，编译器也不能做优化将这条语句去掉，因为存在对volatile变量count的读取，这条语句存在的唯一目的就是保证segments[i].modCount读取到几乎最新的值。关于containsValue方法的其它部分就不分析了，它和size方法差不多。

跨段方法中还有一个isEmpty()方法，其实现比size()方法还要简单，也不介绍了。最后简单地介绍下迭代方法，如keySet(), values(), entrySet()方法，这些方法都返回相应的迭代器，所有迭代器都继承于Hash_Iterator类(提交时居然提醒我不能包含sh It，只得加了下划线)，里实现了主要的方法。其结构是：

 span abstract class Hash_Iterator{ 

 int nextSegmentIndex; 

 int nextTableIndex; 

 HashEntry K,V [] currentTable; 

 HashEntry K, V nextEntry; 

 HashEntry K, V lastReturned; 

} 

 nextSegmentIndex是段的索引，nextTableIndex是nextSegmentIndex对应段中中hash链的索引，currentTable是nextSegmentIndex对应段的table。调用next方法时主要是调用了advance方法：

 span final void advance() { 

 if (nextEntry != null (nextEntry = nextEntry.next) != null) 

 return; 

 while (nextTableIndex = 0) { 

 if ( (nextEntry = currentTable[nextTableIndex--]) != null) 

 return; 

 while (nextSegmentIndex = 0) { 

 Segment K,V seg = segments[nextSegmentIndex--]; 

 if (seg.count != 0) { 

 currentTable = seg.table; 

 for (int j = currentTable.length - 1; j --j) { 

 if ( (nextEntry = currentTable[j]) != null) { 

 nextTableIndex = j - 1; 

 return; 

} 

不想再多介绍了，唯一需要注意的是跳到下一个段时，一定要先读取下一个段的count变量。 

这种迭代方式的主要效果是不会抛出ConcurrentModificationException。一旦获取到下一个段的table，也就意味着这个段的头结点在迭代过程中就确定了，在迭代过程中就不能反映对这个段节点并发的删除和添加，对于节点的更新是能够反映的，因为节点的值是一个volatile变量。

在实际的应用中，散列表一般的应用场景是：除了少数插入操作和删除操作外，绝大多数都是读取操作，而且读操作在大多数时候都是成功的。正是基于这个前提，ConcurrentHashMap 针对读操作做了大量的优化。通过 HashEntry 对象的不变性和用 volatile 型变量协调线程间的内存可见性，使得 大多数时候，读操作不需要加锁就可以正确获得值。这个特性使得 ConcurrentHashMap 的并发性能在分离锁的基础上又有了近一步的提高。

ConcurrentHashMap 是一个并发散列映射表的实现，它允许完全并发的读取，并且支持给定数量的并发更新。相比于 HashTable 和用同步包装器包装的 HashMap（Collections.synchronizedMap(new HashMap())），ConcurrentHashMap 拥有更高的并发性。在 HashTable 和由同步包装器包装的 HashMap 中，使用一个全局的锁来同步不同线程间的并发访问。同一时间点，只能有一个线程持有锁，也就是说在同一时间点，只能有一个线程能访问容器。这虽然保证多线程间的安全并发访问，但同时也导致对容器的访问变成串行化的了。

ConcurrentHashMap 的高并发性主要来自于三个方面：

用分离锁实现多个线程间的更深层次的共享访问。 用 HashEntery 对象的不变性来降低执行读操作的线程在遍历链表期间对加锁的需求。 通过对同一个 Volatile 变量的写 / 读访问，协调不同线程间读 / 写操作的内存可见性。

使用分离锁，减小了请求 同一个锁的频率。

通过 HashEntery 对象的不变性及对同一个 Volatile 变量的读 / 写来协调内存可见性，使得 读操作大多数时候不需要加锁就能成功获取到需要的值。由于散列映射表在实际应用中大多数操作都是成功的 读操作，所以 2 和 3 既可以减少请求同一个锁的频率，也可以有效减少持有锁的时间。通过减小请求同一个锁的频率和尽量减少持有锁的时间 ，使得 ConcurrentHashMap 的并发性相对于 HashTable 和用同步包装器包装的 HashMap有了质的提高。

13760.html

cgojava