您的位置: 首页  >  文章  >  【Java并发编程】深入分析ConcurrentHashMap(九)

【Java并发编程】深入分析ConcurrentHashMap(九)

分类: 文章 • 2024-01-30 19:12:52

转自：http://blog.****.net/liulongling/article/details/50717706

 本章是提高教程可能对于刚入门同学来说会有些难度，读懂本章你需要了解以下知识点：

一、 【Java基础提高】深入分析final关键字(一)
二、 【Java并发编程】深入分析volatile(四)
三、 【Java基础提高】HashTable源码分析(六)

一、Concurrent源码分析

 ConcurrentHashMap是由Segment(桶)、HashEntry(节点)2大数据结构组成。如下图所示：
【Java并发编程】深入分析ConcurrentHashMap(九)

   1.1 Segment类和属性

[java] view plain copy
  1. //Segment内部维护了一个链表数组  
  2. static final class Segment<K,V> extends ReentrantLock implements Serializable {  
  3.   
  4.     //链表数组，数组中的每一个元素代表了一个链表的头部  
  5.     transient volatile HashEntry<K,V>[] table;  
  6.   
  7.     //Segment中元素的数量  
  8.     transient int count;  
  9.   
  10.     //对table的大小造成影响的操作的数量（比如put或者remove操作）  
  11.     transient int modCount;  
  12.   
  13.     //阈值，Segment里面元素的数量超过这个值会对Segment进行扩容，扩容后大小＝old*2*负载因子  
  14.     transient int threshold;  
  15.   
  16.     //负载因子，用于确定threshold  
  17.     final float loadFactor;  
  18. }  

  Segment继承了ReentrantLock，这意味着每个segment都可以当做一个锁，每一把锁只锁住整个容器中的部分数据，这样不影响线程访问其它数据，当然如果是对全局改变时会锁定所有的segment段。比如：size()和containsValue()，注意的是要按顺序锁定所有段，操作完毕后，再按顺序释放所有段的锁。如果不按顺序的话，有可能会出现死锁。

   1.2 HashEntry类和属性

[java] view plain copy
  1. //HashEntry是一个单向链表    
  2. static final class HashEntry<K,V> {  
  3.     //哈希值    
  4.     final int hash;  
  5.     //存储的key和值value    
  6.     final K key;  
  7.     volatile V value;  
  8.     //指向的下一个HashEntry，即链表的下一个节点    
  9.     volatile HashEntry<K,V> next;  
  10. }  

  类似与HashMap节点Entry，HashEntry也是一个单向链表，它包含了key、hash、value和下一个节点信息。HashEntry和Entry的不同点：
 不同点一：使用了多个final关键字（final class 、final hash） ，这意味着不能从链表的中间或尾部添加或删除节点，后面删除操作时会讲到。
 不同点二：使用volatile，是为了更新值后能立即对其它线程可见。这里没有使用锁，效率更高。

   1.3 类的初始化

[java] view plain copy
  1. /** 
  2.  *  
  3.  * @param initialCapacity  初始容量 
  4.  * @param loadFactor    负载因子 
  5.  * @param concurrencyLevel 代表ConcurrentHashMap内部的Segment的数量， 
  6.  * ConcurrentLevel 并发级别 
  7.  */  
  8. public ConcurrentHashMap(int initialCapacity,  
  9.         float loadFactor, int concurrencyLevel) {  
  10.     if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)  
  11.         throw new IllegalArgumentException();  
  12.     if (concurrencyLevel > MAX_SEGMENTS)  
  13.         concurrencyLevel = MAX_SEGMENTS;  
  14.     // Find power-of-two sizes best matching arguments  
  15.     int sshift = 0;  
  16.     int ssize = 1;  
  17.     while (ssize < concurrencyLevel) {  
  18.         ++sshift;  
  19.         ssize <<= 1;//ssize左移一位也就是每次ssize＝2*ssize。  
  20.     }  
  21.     //主要使用于put()和segmentForHash()方法，结合hash计算出元素在哪一个Segment中。  
  22.         //假如concurrencyLevel是16，那么sshift＝4、segmentShift＝28、segmentMask=15;  
  23.     this.segmentShift = 32 - sshift;  
  24.     this.segmentMask = ssize - 1;  
  25.     if (initialCapacity > MAXIMUM_CAPACITY)  
  26.         initialCapacity = MAXIMUM_CAPACITY;  
  27.     int c = initialCapacity / ssize;  
  28.     if (c * ssize < initialCapacity)  
  29.         ++c;  
  30.     int cap = MIN_SEGMENT_TABLE_CAPACITY;  
  31.     while (cap < c)  
  32.         cap <<= 1;  
  33.     // create segments and segments[0]  
  34.     Segment<K,V> s0 =  
  35.             new Segment<K,V>(loadFactor, (int)(cap * loadFactor),  
  36.                     (HashEntry<K,V>[])new HashEntry[cap]);  
  37.     Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];  
  38.     UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]  
  39.     this.segments = ss;  
  40. }  

  ConcurrencyLevel默认情况下内部按并发级别为16来创建。对于每个segment的容量，默认情况也是16。其中concurrentLevel和segment的初始容量都是可以通过构造函数设定的。要注意的是ConcurrencyLevel一经指定，不可改变，后续如果ConcurrentHashMap的元素数量增加导致ConrruentHashMap需要扩容，ConcurrentHashMap不会增加Segment的数量，而只会增加Segment中链表数组的容量大小，这样的好处是扩容过程不需要对整个ConcurrentHashMap做rehash，而只需要对Segment里面的元素做一次rehash就可以了。

   1.4 ensureSegment()方法

 该方法返回给定索引位置的Segment，如果Segment不存在，则参考Segment表中的第一个Segment的参数创建一个Segment并通过CAS操作将它记录到Segment表中去。
[java] view plain copy
  1. private Segment<K,V> ensureSegment(int k) {  
  2.       final Segment<K,V>[] ss = this.segments;  
  3.       long u = (k << SSHIFT) + SBASE; // raw offset  
  4.       Segment<K,V> seg;  
  5.       if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {  
  6.           Segment<K,V> proto = ss[0]; // use segment 0 as prototype  
  7.           int cap = proto.table.length;  
  8.           float lf = proto.loadFactor;  
  9.           int threshold = (int)(cap * lf);  
  10.           HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];  
  11.           if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))  
  12.               == null) { // recheck  
  13.               Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);  
  14.               while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))  
  15.                      == null) {  
  16.                   if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))  
  17.                       break;  
  18.               }  
  19.           }  
  20.       }  
  21.       return seg;  
  22.   }  

   1.5 entryAt()方法

 entryAt()方法是从链表中查找节点。在方法参数里注意到有传入tab链表数组和index索引，那为什么还要调用entryAt()方法获取数组项的值而不是通过tab[index]方式直接获取？那我们从源头（put）开始分析，见1.6put()操作。
[java] view plain copy
  1. static final <K,V> HashEntry<K,V> entryAt(HashEntry<K,V>[] tab, int i) {  
  2.         return (tab == null) ? null :  
  3.             (HashEntry<K,V>) UNSAFE.getObjectVolatile  
  4.             (tab, ((long)i << TSHIFT) + TBASE);  
  5.    }  

   1.6 put()操作

   1.6.1 锁分离技术

  大家知道HashTable是使用了synchronized来保证线程安全，但是其效率非常差。它效率非常差的原因是多个线程访问HashTable时需要竞争同一把锁，如果我们有多把锁，每一把锁只锁住一部分数据，那么多线程在访问不同的数据时也就不会存在竞争，能提高访问效率。这种做法我们称为锁分离技术。在《Java并发编程实战》一书中作者提到过分拆锁和分离锁技术：
 分拆锁(lock spliting)就是若原先的程序中多处逻辑都采用同一个锁，但各个逻辑之间又相互独立，就可以拆(Spliting)为使用多个锁，每个锁守护不同的逻辑。
 分拆锁有时候可以被扩展，分成可大可小加锁块的集合，并且它们归属于相互独立的对象，这样的情况就是分离锁(lock striping)。
而ConcurrentHashMap就是使用了分离锁技术，对每个Segment配置一把锁，如下图所示：
【Java并发编程】深入分析ConcurrentHashMap(九)

   1.6.2 源码分析

  Segment的put操作原理如下图所示，图中展示的不是很详细，其中关于加锁的步骤没有加上去，原因是加了几次觉得加锁后看着很复杂。用图片展示是为了更加简单和明了，如果看着复杂也就没有意义了，我尽量用文字说清楚它的步骤。
【Java并发编程】深入分析ConcurrentHashMap(九)
步骤一：进入Segment的put操作时先进行加锁保护。如果加锁没有成功，调用scanAndLockForPut方法（详细步骤见下面scanAndLockForPut()源码分析）进入自旋状态，该方法持续查找key对应的节点链中是已存在该机节点，如果没有找到，则预创建一个新节点，并且尝试n次，直到尝试次数操作限制，才真正进入加锁等待状态，自旋结束并返回节点（如果返回了一个非空节点，则表示在链表中没有找到相应的节点）。对最大尝试次数，目前的实现单核次数为1，多核为64。
步骤二：使用(tab.length - 1) & hash计算第一个节点位置，再通过entryAt()方法去查找第一个节点。如果节点存在，遍历链表找到key值所在的节点，如果找到了这个节点则直接更新旧value，结束循环。其中value使用了volatile，它更新后的值立马对其它线程可见。如果节点不存在，将步骤一预创建的新节点（如果没有则重新创建）添加到链表中，添加前先检查添加后节点数量是否超过容器大小，如果超过了，则rehash操作。没有的话调用setNext或setEntryAt方法添加新节点;
要注意的是在更新链表时使用了Unsafe.putOrderedObject()方法，这个方法能够实现非堵塞的写入，这些写入不会被Java的JIT重新排序指令(instruction reordering)，使得它能更加快速的存储。
解决1.5问题：为什么还要调用entryAt()方法获取数组项的值而不是通过tab[index]方式直接获取？
  虽然在开始时volatile table将引用赋值给了变量tab，但是多线程下table里的值可能发生改变，使用tab[index]并不能获得最新的值。。为了保证接下来的put操作能够读取到上一次的更新结果，需要使用volatile的语法去读取节点链的链头.

[java] view plain copy
  1. public V put(K key, V value) {  
  2.     Segment<K,V> s;  
  3.     if (value == null)  
  4.         throw new NullPointerException();  
  5.     int hash = hash(key);  
  6.     //计算Segment的位置，在初始化的时候对segmentShift和segmentMask做了解释  
  7.     int j = (hash >>> segmentShift) & segmentMask;  
  8.     //从Segment数组中获取segment元素的位置  
  9.     if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck  
  10.             (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment  
  11.         s = ensureSegment(j);  
  12.     //  
  13.     return s.put(key, hash, value, false);  
  14. }  
  15.   
  16. //往Segment的HashEntry中添加元素，使用了分锁机制  
  17. final V put(K key, int hash, V value, boolean onlyIfAbsent) {  
  18.     //tryLock 仅在调用时锁为空闲状态才获取该锁。如果锁可用，则获取锁，并立即返回值 true。否则是false  
  19.     //scanAndLockForPut 下面单独说scanAndLockForPut  
  20.     HashEntry<K,V> node = tryLock() ? null :scanAndLockForPut(key, hash, value);  
  21.     V oldValue;  
  22.     try {  
  23.         HashEntry<K,V>[] tab = table;  
  24.         int index = (tab.length - 1) & hash;  
  25.         HashEntry<K,V> first = entryAt(tab, index);  
  26.         for (HashEntry<K,V> e = first;;) {  
  27.             if (e != null) {  
  28.                 K k;  
  29.                 if ((k = e.key) == key ||  
  30.                         (e.hash == hash && key.equals(k))) {  
  31.                     oldValue = e.value;  
  32.                     if (!onlyIfAbsent) {  
  33.                         e.value = value;  
  34.                         ++modCount;  
  35.                     }  
  36.                     break;  
  37.                 }  
  38.                 e = e.next;  
  39.             }  
  40.             else {  
  41.                 if (node != null)  
  42.                     node.setNext(first);  
  43.                 else  
  44.                     node = new HashEntry<K,V>(hash, key, value, first);  
  45.                 int c = count + 1;  
  46.                 if (c > threshold && tab.length < MAXIMUM_CAPACITY)  
  47.                     rehash(node);  
  48.                 else  
  49.                     setEntryAt(tab, index, node);  
  50.                 ++modCount;  
  51.                 count = c;  
  52.                 oldValue = null;  
  53.                 break;  
  54.             }  
  55.         }  
  56.     } finally {  
  57.         unlock();  
  58.     }  
  59.     return oldValue;  
  60. }  

   1.5 segmentForHash()方法

[java] view plain copy
  1. /** 
  2.  * 查找Segment对象，这里Unsafe的主要作用是提供原子操作。 
  3.  */  
  4. @SuppressWarnings("unchecked")  
  5. private Segment<K,V> segmentForHash(int h) {  
  6.     long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;  
  7.     return (Segment<K,V>) UNSAFE.getObjectVolatile(segments, u);  
  8. }  

   1.6 scanAndLockForPut()方法

  在下面代码中，它先获取key对应的头节点，进入链表循环。如果链表中不存在要插入的节点，则预创建一个新节点，否则retries值递增，直到操作最大尝试次数而进入等待状态。这个方法要注意的是：当在自旋过程中发现链表链头发生了变化，则更新节点链的链头，并重置retries值为－1，重新为尝试获取锁而自旋遍历。
[java] view plain copy
  1. private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {  
  2.     //第一步：先找到HashEntry链表中的头节点  
  3.     HashEntry<K,V> first = entryForHash(this, hash);  
  4.     HashEntry<K,V> e = first;  
  5.     HashEntry<K,V> node = null;  
  6.     int retries = -1; // negative while locating node  
  7.     while (!tryLock()) {  
  8.         HashEntry<K,V> f; // to recheck first below  
  9.         //第一次一定执行该条件内代码  
  10.         if (retries < 0) {  
  11.             //第二步：分三种情景  
  12.             //情景一：没有找到，创建一个新的节点。  
  13.             if (e == null) {  
  14.                 if (node == null) // speculatively create node  
  15.                     node = new HashEntry<K,V>(hash, key, value, null);  
  16.                 retries = 0;  
  17.             }  
  18.             //情景二：找到相同key的节点  
  19.             else if (key.equals(e.key))  
  20.                 retries = 0;  
  21.             else  
  22.             //情景三：没找到key值对应的节点，指向下一个节点继续  
  23.                 e = e.next;  
  24.         }  
  25.         //尝试次数达到限制进入加锁等待状态。 对最大尝试次数，目前的实现单核次数为1，多核为64：  
  26.         else if (++retries > MAX_SCAN_RETRIES) {  
  27.             lock();  
  28.             break;  
  29.         }  
  30.         //retries是偶数并且不是头节点。在自旋中链头可能会发生变化  
  31.         else if ((retries & 1) == 0 &&  
  32.                  (f = entryForHash(this, hash)) != first) {  
  33.             e = first = f; // re-traverse if entry changed  
  34.             retries = -1;  
  35.         }  
  36.     }  
  37.     return node;  
  38. }  

   1.7 get()操作

[java] view plain copy
  1. public V get(Object key) {  
  2.        Segment<K,V> s; // manually integrate access methods to reduce overhead  
  3.        HashEntry<K,V>[] tab;  
  4.        int h = hash(key);  
  5.        long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;  
  6.        if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&  
  7.            (tab = s.table) != null) {  
  8.            for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile  
  9.                     (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);  
  10.                 e != null; e = e.next) {  
  11.                K k;  
  12.                if ((k = e.key) == key || (e.hash == h && key.equals(k)))  
  13.                    return e.value;  
  14.            }  
  15.        }  
  16.        return null;  
  17.    }  
  从代码可以看出get方法并没有调用锁，它使用了volatile的可见性来实现线程安全的。

参考资料

相关推荐