concurrentHashMap 到底采取了什么措施使得它比synchronized(HashMap)好

矫情吗；* 2022-08-06 13:04 13阅读 0赞

免锁容器背后的通用策略：

1使用Lock机制 而不是synchronized

2 对容器的修改可以与读取操作同时发生，只要读取者只能看到完成修改的结果即可。（volatile）

3 修改是在容器数据结构的某一部分的一个单独副本上执行的，这样就能避免锁住整个容器

Java Memory Model中，Memory分为两类，  
  
main memory和working memory，main memory为所有线程共享，working memory中存放的是线程所需要的变量的拷贝（线程要对main memory中的内容进行操作的话，首先需要拷贝到自己的working memory，一般为了速度，working memory一般是在cpu的cache中的）。

volatile的变量在被操作的时候不会产生working memory的拷贝，而是直接操作main memory，当然volatile虽然解决了变量的可见性问题，但没有解决变量操作的原子性的问题，这个还需要synchronized或者CAS相关操作配合进行。

### 关于ConcurrentHashMap ###

[http://www.iteye.com/topic/1103980][http_www.iteye.com_topic_1103980]

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

### ConcurrentHashMap的内部结构 ###

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

第一次Hash：

/\*\*  
     \* The default concurrency level for this table, used when not  
     \* otherwise specified in a constructor.  
     \*/  
    static final int DEFAULT\_CONCURRENCY\_LEVEL = 16;

/\*\*  
     \* Mask value for indexing into segments. The upper bits of a  
     \* key's hash code are used to choose the segment.  
     \*/  
    final int segmentMask;  
  
  
    /\*\*  
     \* Shift value for indexing within segments.  
     \*/  
    final int segmentShift;

concurrentLevel，代表ConcurrentHashMap内部的Segment的数量，ConcurrentLevel一经指定，不可改变，

Segment的数量永远是2的指数个，这样的好处是方便采用移位操作来进行hash.

segmentShift和segmentMask，假设构造函数确定了Segment的数量是2的n次方，那么segmentShift就等于32减去n，而segmentMask就等于2的n次方减一。

对hashCode的处理：

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);  
    \}

通过这种再哈希能让数字的每一位都能参加到哈希运算当中，从而减少哈希冲突。

(hash >>> segmentShift) & segmentMask

仅使用了处理过的高n位。

**Segment**

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

Java代码 [ ][http_www.iteye.com_topic_1103980]

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的结构：

Java代码 [ ][http_www.iteye.com_topic_1103980]

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.  \}

**ConcurrentHashMap的get操作**

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

Java代码

public V get(Object key) \{

int hash = hash(key.hashCode());

return segmentFor(hash).get(key, hash);

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

Java代码 [ ][http_www.iteye.com_topic_1103980]

1.  final Segment<K,V> segmentFor(int hash) \{ 
2.  return segments\[(hash >>> segmentShift) & segmentMask\]; 
3.  \}

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方法：

Java代码 [ ][http_www.iteye.com_topic_1103980]

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的定义：

Java代码 [ ][http_www.iteye.com_topic_1103980]

1.  transient volatile int count;

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

写道

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

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

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

Java代码 [ ][http_www.iteye.com_topic_1103980]

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。

这似乎有些费解，理论上结点的值不可能为空，这是因为 put的时候就进行了判断，如果为空就要抛NullPointerException。

空值的唯一源头就是HashEntry中的默认值，因为 HashEntry中的value不是final的，非同步读取有可能读取到空值。

仔细看下put操作的语句：tab\[index\] = new HashEntry<K,V>(key, hash, first, value)，在这条语句中，HashEntry构造函数中对value的赋值以及对tab\[index\]的赋值可能被重新排序，这就可能导致结点的值为空。

这种情况应当很罕见，一旦发生这种情况，ConcurrentHashMap采取的方式是在持有锁的情况下再读一遍，这能够保证读到最新的值，并且一定不会为空值。  
1. V readValueUnderLock(HashEntry<K,V> e) \{  
2. lock();  
3. try \{  
4. return e.value;  
5. \} finally \{  
6. unlock();  
7. \}  
8. \}

**ConcurrentHashMap的put操作**

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

Java代码 [ ][http_www.iteye.com_topic_1103980]

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的值。

[http_www.iteye.com_topic_1103980]: http://www.iteye.com/topic/1103980
[1]: http://pic.yupoo.com/goldendoc/Ba4GCFe1/nuEZ0.png
[1_1]: /images/20220806/3ba32151bfaf41e8b739e3d32c395bf4.png