ConcurrentHashMap原理分析
Posted 平凡希
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ConcurrentHashMap原理分析相关的知识，希望对你有一定的参考价值。
一、出现背景
1、线程不安全的HashMap
因为多线程环境下，使用Hashmap进行put操作会引起死循环，导致CPU利用率接近100%，所以在并发情况下不能使用HashMap。
2、效率低下的HashTable容器
HashTable容器使用synchronized来保证线程安全，但在线程竞争激烈的情况下HashTable的效率非常低下。因为当一个线程访问HashTable的同步方法时，其他线程访问HashTable的同步方法时，可能会进入阻塞或轮询状态。如线程1使用put进行添加元素，线程2不但不能使用put方法添加元素，并且也不能使用get方法来获取元素，所以竞争越激烈效率越低。也就是说对于Hashtable而言，synchronized是针对整张Hash表的，即每次锁住整张表让线程独占。相当于所有线程进行读写时都去竞争一把锁，导致效率非常低下。
3、ConcurrentHashMap的锁分段技术
HashTable容器在竞争激烈的并发环境下表现出效率低下的原因，是因为所有访问HashTable的线程都必须竞争同一把锁。那假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效的提高并发访问效率，这就是ConcurrentHashMap所使用的锁分段技术，首先将数据分成一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也能被其他线程访问。 另外，ConcurrentHashMap可以做到读取数据不加锁，并且其内部的结构可以让其在进行写操作的时候能够将锁的粒度保持地尽量地小，不用对整个ConcurrentHashMap加锁。
ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁ReentrantLock，在ConcurrentHashMap里扮演锁的角色，HashEntry则用于存储键值对数据。一个ConcurrentHashMap里包含一个Segment数组，Segment的结构和HashMap类似，是一种数组和链表结构， 一个Segment里包含一个HashEntry数组，每个HashEntry是一个链表结构的元素， 每个Segment守护着一个HashEntry数组里的元素，当对HashEntry数组的数据进行修改时，必须首先获得它对应的Segment锁。
二、ConcurrentHashMap的内部结构
ConcurrentHashMap为了提高本身的并发能力，在内部采用了一个叫做Segment的结构，一个Segment其实就是一个类Hash Table的结构，Segment内部维护了一个链表数组，我们用下面这一幅图来看下ConcurrentHashMap的内部结构：
从上面的结构我们可以了解到，ConcurrentHashMap定位一个元素的过程需要进行两次Hash操作，第一次Hash定位到Segment，第二次Hash定位到元素所在的链表的头部，因此，这一种结构的带来的副作用是Hash的过程要比普通的HashMap要长，但是带来的好处是写操作的时候可以只对元素所在的Segment进行加锁即可，不会影响到其他的Segment，这样，在最理想的情况下，ConcurrentHashMap可以最高同时支持Segment数量大小的写操作（刚好这些写操作都非常平均地分布在所有的Segment上），所以，通过这一种结构，ConcurrentHashMap的并发能力可以大大的提高。
1、Segment
我们再来具体了解一下Segment的数据结构：
static final class Segment<K,V> extends ReentrantLock implements Serializable { transient volatile int count; transient int modCount; transient int threshold; transient volatile HashEntry<K,V>[] table; final float loadFactor; }
详细解释一下Segment里面的成员变量的意义：
- count：Segment中元素的数量
- modCount：对table的大小造成影响的操作的数量（比如put或者remove操作）
- threshold：阈值，Segment里面元素的数量超过这个值依旧就会对Segment进行扩容
- table：链表数组，数组中的每一个元素代表了一个链表的头部
- loadFactor：负载因子，用于确定threshold
count用来统计该段数据的个数，它是volatile变量，它用来协调修改和读取操作，以保证读取操作能够读取到几乎最新的修改。协调方式是这样的，每次修改操作做了结构上的改变，如增加/删除节点(修改节点的值不算结构上的改变)，都要写count值，每次读取操作开始都要读取count的值。这利用了 Java 5中对volatile语义的增强，对同一个volatile变量的写和读存在happens-before关系。modCount统计段结构改变的次数，主要是为了检测对多个段进行遍历过程中某个段是否发生改变，在讲述跨段操作时会还会详述。threashold用来表示需要进行rehash的界限值。table数组存储段中节点，每个数组元素是个hash链，用HashEntry表示。table也是volatile，这使得能够读取到最新的 table值而不需要同步。loadFactor表示负载因子。
2、HashEntry
Segment中的元素是以HashEntry的形式存放在链表数组中的，看一下HashEntry的结构：
static final class HashEntry<K,V> { final K key; final int hash; volatile V value; final HashEntry<K,V> next; }
可以看到HashEntry的一个特点，除了value以外，其他的几个变量都是final的，这意味着不能从hash链的中间或尾部添加或删除节点，因为这需要修改next 引用值，所有的节点的修改只能从头部开始。对于put操作，可以一律添加到Hash链的头部。但是对于remove操作，可能需要从中间删除一个节点，这就需要将要删除节点的前面所有节点整个复制一遍，最后一个节点指向要删除结点的下一个结点。这在讲解删除操作时还会详述。为了确保读操作能够看到最新的值，将value设置成volatile，这避免了加锁。
三、ConcurrentHashMap的初始化
下面我们来结合源代码来具体分析一下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次方减一。
四、ConcurrentHashMap的get操作
前面提到过ConcurrentHashMap的get操作是不用加锁的，我们这里看一下其实现：
1 public V get(Object key) { 2 int hash = hash(key.hashCode()); 3 return segmentFor(hash).get(key, hash); 4 }
第二行，对hash值进行了二次hash，之所以要进行再哈希，其目的是为了减少哈希冲突，使元素能够均匀的分布在不同的Segment上，从而提高容器的存取效率。
看第三行，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方法：
1 V get(Object key, int hash) { 2 if (count != 0) { // read-volatile // ① 3 HashEntry<K,V> e = getFirst(hash); 4 while (e != null) { 5 if (e.hash == hash && key.equals(e.key)) { 6 V v = e.value; 7 if (v != null) // ② 注意这里
8 return v; 9 return readValueUnderLock(e); // recheck 10 } 11 e = e.next; 12 } 13 } 14 return null; 15 }
先看第二行代码，这里对count进行了一次判断，其中count表示Segment中元素的数量。我们可以来看一下count的定义：
transient volatile int count;
可以看到count是volatile的，实际上这里面利用了volatile的语义：
对volatile字段的写入操作happens-before于每一个后续的同一个字段的读操作。因为实际上put、remove等操作也会更新count的值，所以当竞争发生的时候，volatile的语义可以保证写操作在读操作之前，也就保证了写操作对后续的读操作都是可见的，这样后面get的后续操作就可以拿到完整的元素内容。
然后，在第三行，调用了getFirst()来取得链表的头部：
1 HashEntry<K,V> getFirst(int hash) { 2 HashEntry<K,V>[] tab = table; 3 return tab[hash & (tab.length - 1)]; 4 }
同样，这里也是用位操作来确定链表的头部，hash值和HashTable的长度减一做与操作，最后的结果就是hash值的低n位，其中n是HashTable的长度以2为底的结果。
在确定了链表的头部以后，就可以对整个链表进行遍历，看第4行，取出key对应的value的值，如果拿出的value的值是null，则可能这个key，value对正在put的过程中，如果出现这种情况，那么就加锁来保证取出的value是完整的，如果不是null，则直接返回value。
get方法没有使用锁来同步，只是判断获取的entry的value是否为null，为null时才使用加锁的方式再次去获取。
这个实现很微妙，没有锁同步的话，靠什么保证同步呢？我们一步步分析。
第一步，先判断一下 count != 0；count变量表示segment中存在entry的个数。如果为0就不用找了。假设这个时候恰好另一个线程put或者remove了这个segment中的一个entry，会不会导致两个线程看到的count值不一致呢？看一下count 变量的定义：
transient volatile int count;
它使用了volatile来修改。在Java5之后，JMM实现了对volatile的保证：对volatile域的写入操作happens-before于每一个后续对同一个域的读写操作。所以，每次判断count变量的时候，即使恰好其他线程改变了segment也会体现出来。
第二步，获取到要该key所在segment中的索引地址，如果该地址有相同的hash对象，顺着链表一直比较下去找到该entry。当找到entry的时候，先做了一次比较： if(v != null) 我们用红色注释的地方。这是为何呢？考虑一下，如果这个时候，另一个线程恰好新增/删除了entry，或者改变了entry的value，会如何？
前面说过HashEntry类的结构，除了 value，其它成员都是final修饰的，也就是说value可以被改变，其它都不可以改变，包括指向下一个HashEntry的next也不能被改变。
（1）在get代码的①和②之间，另一个线程新增了一个entry。如果另一个线程新增的这个entry又恰好是我们要get的，这事儿就比较微妙了。下图大致描述了put 一个新的entry的过程。
因为每个HashEntry中的next也是final的，没法对链表最后一个元素增加一个后续entry所以新增一个entry的实现方式只能通过头结点来插入了。
newEntry对象是通过 new HashEntry(K k , V v, HashEntry next) 来创建的。如果另一个线程刚好new 这个对象时，当前线程来get它。因为没有同步，就可能会出现当前线程得到的newEntry对象是一个没有完全构造好的对象引用。没有锁同步的话，new 一个对象对于多线程看到这个对象的状态是没有保障的，这里同样有可能一个线程new这个对象的时候还没有执行完构造函数就被另一个线程得到这个对象引用。所以才需要判断一下：if (v != null) 如果确实是一个不完整的对象，则使用锁的方式再次get一次。
有没有可能会put进一个value为null的entry？ 不会的，已经做了检查，这种情况会抛出异常，所以 ②处的判断完全是出于对多线程下访问一个new出来的对象的状态检测。
（2）在get代码的①和②之间，另一个线程修改了一个entry的value。value是用volitale修饰的，可以保证读取时获取到的是修改后的值。
（3）在get代码的①之后，另一个线程删除了一个entry。
假设我们的链表元素是：e1-> e2 -> e3 -> e4 我们要删除 e3这个entry，因为HashEntry中next的不可变，所以我们无法直接把e2的next指向e4，而是将要删除的节点之前的节点复制一份，形成新的链表。它的实现大致如下图所示：
如果我们get的也恰巧是e3，可能我们顺着链表刚找到e1，这时另一个线程就执行了删除e3的操作，而我们线程还会继续沿着旧的链表找到e3返回。这里没有办法实时保证了。
我们第①处就判断了count变量，它保障了在 ①处能看到其他线程修改后的。①之后到②之间，如果再次发生了其他线程再删除了entry节点，就没法保证看到最新的了。不过这也没什么关系，即使我们返回e3的时候，它被其他线程删除了，暴漏出去的e3也不会对我们新的链表造成影响。
这其实是一种乐观设计，设计者假设 ①之后到②之间 发生被其它线程增、删、改的操作可能性很小，所以不采用同步设计，而是采用了事后（其它线程这期间也来操作，并且可能发生非安全事件）弥补的方式。而因为其他线程的“改”和“删”对我们的数据都不会造成影响，所以只有对“新增”操作进行了安全检查，就是②处的非null检查，如果确认不安全事件发生，则采用加锁的方式再次get。这样做减少了使用互斥锁对并发性能的影响。可能有人怀疑remove操作中复制链表的方式是否代价太大，这里我没有深入比较，不过既然Java5中这么实现，我想new一个对象的代价应该已经没有早期认为的那么严重。
五、ConcurrentHashMap的put操作
看完了get操作，再看下put操作，put操作的前面也是确定Segment的过程，这里不再赘述，直接看关键的segment的put方法：
1 V put(K key, int hash, V value, boolean onlyIfAbsent) { 2 lock(); 3 try { 4 int c = count; 5 if (c++ > threshold) // ensure capacity 6 rehash(); 7 HashEntry<K,V>[] tab = table; 8 int index = hash & (tab.length - 1); 9 HashEntry<K,V> first = tab[index]; 10 HashEntry<K,V> e = first; 11 while (e != null && (e.hash != hash || !key.equals(e.key))) 12 e = e.next; 13 14 V oldValue; 15 if (e != null) { 16 oldValue = e.value; 17 if (!onlyIfAbsent) 18 e.value = value; 19 } 20 else { 21 oldValue = null; 22 ++modCount; 23 tab[index] = new HashEntry<K,V>(key, hash, first, value); 24 count = c; // write-volatile 25 } 26 return oldValue; 27 } finally { 28 unlock(); 29 } 30 }
首先对Segment的put操作是加锁完成的，然后在第五行，如果Segment中元素的数量超过了阈值（由构造函数中的loadFactor算出）这需要进行对Segment扩容，并且要进行rehash，关于rehash的过程大家可以自己去了解，这里不详细讲了。
第8和第9行的操作就是getFirst的过程，确定链表头部的位置。
第11行这里的这个while循环是在链表中寻找和要put的元素相同key的元素，如果找到，就直接更新更新key的value，如果没有找到，则进入21行这里，生成一个新的HashEntry并且把它加到整个Segment的头部，然后再更新count的值。
该方法也是在持有段锁(锁定整个segment)的情况下执行的，这当然是为了并发的安全，修改数据是不能并发进行的，必须得有个判断是否超限的语句以确保容量不足时能够rehash。接着是找是否存在同样一个key的结点，如果存在就直接替换这个结点的值。否则创建一个新的结点并添加到hash链的头部，这时一定要修改modCount和count的值，同样修改count的值一定要放在最后一步。put方法调用了rehash方法，reash方法实现得也很精巧，主要利用了table的大小为2^n，这里就不介绍了。而比较难懂的是这句int index = hash & (tab.length - 1)，原来segment里面才是真正的hashtable，即每个segment是一个传统意义上的hashtable，如上图，从两者的结构就可以看出区别，这里就是找出需要的entry在table的哪一个位置，之后得到的entry就是这个链的第一个节点，如果e!=null，说明找到了，这是就要替换节点的值（onlyIfAbsent == false），否则，我们需要new一个entry，它的后继是first，而让tab[index]指向它，什么意思呢？实际上就是将这个新entry插入到链头，剩下的就非常容易理解了。
六、ConcurrentHashMap的remove操作
Remove操作的前面一部分和前面的get和put操作一样，都是定位Segment的过程，然后再调用Segment的remove方法：
1 V remove(Object key, int hash, Object value) { 2 lock(); 3 try { 4 int c = count - 1; 5 HashEntry<K,V>[] tab = table; 6 int index = hash & (tab.length - 1); 7 HashEntry<K,V> first = tab[index]; 8 HashEntry<K,V> e = first; 9 while (e != null && (e.hash != hash || !key.equals(e.key))) 10 e = e.next; 11 12 V oldValue = null; 13 if (e != null) { 14 V v = e.value; 15 if (value == null || value.equals(v)) { 16 oldValue = v; 17 // All entries following removed node can stay 18 // in list, but all preceding ones need to be 19 // cloned. 20 ++modCount; 21 HashEntry<K,V> newFirst = e.next; 22 for (HashEntry<K,V> p = first; p != e; p = p.next) 23 newFirst = new HashEntry<K,V>(p.key, p.hash, 24 newFirst, p.value); 25 tab[index] = newFirst; 26 count = c; // write-volatile 27 } 28 } 29 return oldValue; 30 } finally { 31 unlock(); 32 } 33 }
首先remove操作也是确定需要删除的元素的位置，不过这里删除元素的方法不是简单地把待删除元素的前面的一个元素的next指向后面一个就完事了，我们之前已经说过HashEntry中的next是final的，一经赋值以后就不可修改，在定位到待删除元素的位置以后，程序就将待删除元素前面的那一些元素全部复制一遍，然后再一个一个重新接到链表上去，看一下下面这一幅图来了解这个过程：
假设链表中原来的元素如上图所示，现在要删除元素3，那么删除元素3以后的链表就如下图所示：
整个操作是在持有段锁的情况下执行的，空白行之前(第11行之前)的行主要是定位到要删除的节点e。接下来，如果不存在这个节点就直接返回null，否则就要将e前面的结点复制一遍，尾结点指向e的下一个结点。e后面的结点不需要复制，它们可以重用。
中间那个for循环是做什么用的呢？（第22行）从代码来看，就是将定位之后的所有entry克隆并拼回前面去，但有必要吗？每次删除一个元素就要将那之前的元素克隆一遍？这点其实是由entry的不变性来决定的，仔细观察entry定义，发现除了value，其他所有属性都是用final来修饰的，这意味着在第一次设置了next域之后便不能再改变它，取而代之的是将它之前的节点全都克隆一次。至于entry为什么要设置为不变性，这跟不变性的访问不需要同步从而节省时间有关。
整个remove实现并不复杂，但是需要注意如下几点。第一，当要删除的结点存在时，删除的最后一步操作要将count的值减一。这必须是最后一步操作，否则读取操作可能看不到之前对段所做的结构性修改。第二，remove执行的开始就将table赋给一个局部变量tab，这是因为table是 volatile变量，读写volatile变量的开销很大。编译器也不能对volatile变量的读写做任何优化，直接多次访问非volatile实例变量没有多大影响，编译器会做相应优化。
七、总结
ConcurrentHashMap背景：出现的原因是因为我们起先使用的是HashMap和HashTable，但是随着并发量的增加，HashMap并没有使用同步，在多线程情况下使用HashMap的时候就会出现并发问题，而HashTable虽然是安全的，但是使用的是synchronized 锁整表操作，这样在性能上将会产生很大的影响。那么如何能设计出一款即安全，在效率上又高的集合呢，这样就有了ConcurrentHashMap的产生。
ConcurrentHashMap采用的是锁分段技术，内部为Segment数组来进行细分，而每个Segment又通过HashEntry数组来进行组装，当进行写操作的时候，只需要对这个key对应的Segment进行加锁操作，加锁同时不会对其他的Segment造成影响。总的Map包含了16个Segment（默认数量），每个Segment内部包含16个HashEntry（默认数量），这样对于这个key所在的Segment加锁的同时，其他15个Segmeng还能正常使用，在性能上有了大大的提升。
同时ConcurrentHashMap只是针对put方法进行了加锁，而对于get方法并没有采用加锁的操作，因为具体的值，在Segment的HashEntry里面是volatile的，基于happens-before（先行发生）原则，对数据的写先行发生于对数据的读，所以再读取的时候获取到的必然是最新的结果。
因为对数组的操作，在主内存和工作内存中，load和use、assgin和store是必然连在一起的，一旦使用（use）发生，那load必先行发生于use之前，use前必然从主内存中加载最新的值到工作内存的变量副本里。而一旦赋值（assgin），必然先行发生于store将值传递给主内存，在write到主内存中去。所以put方式无需加锁也能获取到最新的结果。
size操作是先请求2次的count数量，如果有发生变化，则对put、remove、clean进行加锁，在统计完之后unlock。
以上是关于ConcurrentHashMap原理分析的主要内容，如果未能解决你的问题，请参考以下文章
HashMap和ConcurrentHashMap实现原理及源码分析