ConcurrentHashMap1。7版本源码深度解读

　　ConcurrentHashMap是concurrent并发包下重要的工具类，它的设计和实现非常的巧妙，它将数据分段，每段数据使用一个AQS锁，从而减小了锁的粒度。1.ConcurrentHashMap的结构
　　结构图
　　一个ConcurrentHashMap是由多个Segment（段）组成的。Segment类继承了ReentrantLock类，这意味着每个Segment拥有了一个AQS，多个线程的操作落到同一个Segment上时，发生了锁的竞争。ConcurrentHashMap默认有16个Segment，在初始化之后，Segment个数不可修改。
　　一个Segment包含了一个HashEntry的数组，每个HashEntry都是一个单向链表，HashEntry的数组可以扩容，扩容后数组的长度为原来的2倍。HashEntry类如下图所示：
　　我们看到value和next都是volatile修饰的，这保证了数据的可见性。2.put方法详解public V put(K key, V value) {     Segment s;     if (value == null)         throw new NullPointerException();     //计算key的hash值     int hash = hash(key);     //hash为32位无符号数，segmentShift默认为28，向右移28位，剩下高4位     //然后和segmentMask（默认值为15）做与运算，结果还是高4位     int j = (hash >>> segmentShift) & segmentMask;     if ((s = (Segment)UNSAFE.getObject          // nonvolatile; recheck          (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment        // 对 segment[j] 进行初始化       s = ensureSegment(j);     //放入到对应的段内     return s.put(key, hash, value, false); }
　　第一步是根据hash值快速获取到相应的Segment，第二步就是Segment内部的put操作了。final V put(K key, int hash, V value, boolean onlyIfAbsent) {     //获取该segment的独占锁     HashEntry node = tryLock() ? null :         scanAndLockForPut(key, hash, value);     V oldValue;     try {         HashEntry[] tab = table;         //用hash值和（数组长度-1）做与运算，得出数组的下标         int index = (tab.length - 1) & hash;         //first 是数组该位置处的链表的表头         HashEntry first = entryAt(tab, index);         for (HashEntry e = first;;) {             //判断是不是到了尾部，尾部==null             if (e != null) {                 K k;                 //key相同，值更新                 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 {                 //node不为空，则node作为头节点，使用的是头插法                 if (node != null)                     node.setNext(first);                 else                     node = new HashEntry(hash, key, value, first);                 int c = count + 1;                 // 如果超过了该 segment 的阈值，这个 segment 需要扩容                 if (c > threshold && tab.length < MAXIMUM_CAPACITY)                     rehash(node);                 else                     // 没有达到阈值，将 node 放到数组 tab 的 index 位置，                     // 其实就是将新的节点设置成原链表的表头                     setEntryAt(tab, index, node);                 ++modCount;                 count = c;                 oldValue = null;                 break;             }         }     } finally {         unlock();     }     return oldValue; }
　　如何进行扩容：private void rehash(HashEntry node) {          HashEntry[] oldTable = table;     int oldCapacity = oldTable.length;     //扩容为原来的2倍     int newCapacity = oldCapacity << 1;     //计算阀值     threshold = (int)(newCapacity * loadFactor);     HashEntry[] newTable =         (HashEntry[]) new HashEntry[newCapacity];     int sizeMask = newCapacity - 1;     //循环旧的HashEntry数组     for (int i = 0; i < oldCapacity ; i++) {         HashEntry e = oldTable[i];         if (e != null) {             HashEntry next = e.next;             int idx = e.hash & sizeMask;             //该链表上只有一个节点             if (next == null)   //  Single node on list                 newTable[idx] = e;             else { // Reuse consecutive sequence at same slot                 HashEntry lastRun = e;                 int lastIdx = idx;                 for (HashEntry last = next;                      last != null;                      last = last.next) {                     //计算在新数组中的下标                     int k = last.hash & sizeMask;                     //当前节点的下标和上一个节点的下标不一致时，修改最终节点值                     //注意如果后面的节点和前面的节点下标一致，                     //那么后面的节点保持原有的顺序，直接带到新tab[k]的链表中                     if (k != lastIdx) {                         lastIdx = k;                         lastRun = last;                     }                 }                 //采用头插法，最后一个节点作为头节点                 newTable[lastIdx] = lastRun;                 //重新计算节点在数组中的位置，采用头插法插入到链表                 for (HashEntry p = e; p != lastRun; p = p.next) {                     V v = p.value;                     int h = p.hash;                     int k = h & sizeMask;                     HashEntry n = newTable[k];                     newTable[k] = new HashEntry(h, p.key, v, n);                 }             }         }     }     //添加新节点     int nodeIndex = node.hash & sizeMask; // add the new node     node.setNext(newTable[nodeIndex]);     newTable[nodeIndex] = node;     //将newTable赋值给该segment的table     table = newTable; }
　　自旋获取aqs锁：private HashEntry scanAndLockForPut(K key, int hash, V value) {     HashEntry first = entryForHash(this, hash);     HashEntry e = first;     HashEntry node = null;     int retries = -1; // negative while locating node      //如果尝试加锁失败，那么就对segment[hash]对应的链表进行遍历找到需要put的这个entry所在的链表中的位置，     //这里之所以进行一次遍历找到坑位，主要是为了通过遍历过程将遍历过的entry全部放到CPU高速缓存中，     //这样在获取到锁了之后，再次进行定位的时候速度会十分快，这是在线程无法获取到锁前并等待的过程中的一种预热方式。     while (!tryLock()) {         HashEntry f; // to recheck first below         //获取锁失败，初始时retries=-1必然开始先进入第一个if         if (retries < 0) {             //e=null代表两种意思，             //1.第一种就是遍历链表到了最后，仍然没有发现指定key的entry；             //2.第二种情况是刚开始时entryForHash(通过hash找到的table中对应位置链表的结点)找到的HashEntry就是空的             if (e == null) {                 /*                 当然这里之所以还需要对node==null进行判断，是因为有可能在第一次给node赋值完毕后，然后预热准备工作已经搞定，然后进行循环尝试获取锁，在循环次数还未达到<2>64次以前，某一次在条件<3>判断时发现有其它线程对这个segment进行了修改，那么retries被重置为-1，从而再一次进入到<1>条件内，此时如果再次遍历到链表最后时，因为上一次遍历时已经给node赋值过了，所以这里判断node是否为空，从而避免第二次创建对象给node重复赋值。                 */                 if (node == null) // speculatively create node                     node = new HashEntry(hash, key, value, null);                 retries = 0;             }             else if (key.equals(e.key))                 retries = 0;             else                 e = e.next;         }         else if (++retries > MAX_SCAN_RETRIES) {             // 尝试获取锁次数超过设置的最大值，直接进入阻塞等待，这就是所谓的有限制的自旋获取锁，             //之所以这样是因为如果持有锁的线程要过很久才释放锁，这期间如果一直无限制的自旋其实是对cpu性能有消耗的，             //这样无限制的自旋是不利的，所以加入最大自旋次数，超过这个次数则进入阻塞状态等待对方释放锁并获取锁             lock();             break;         }         // 遍历过程中，有可能其它线程改变了遍历的链表，这时就需要重新进行遍历了。         //判断是否初始化了结点 并且 判断链表头结点是否改变(1.7使用头插法)         else if ((retries & 1) == 0 &&                  (f = entryForHash(this, hash)) != first) {             e = first = f; // re-traverse if entry changed             retries = -1;         }     }     return node; }
　　这个方法用了一个while循环去获取aqs锁，有两种情况需要说明下：
　　1.如果尝试的次数超过了最大自旋次数，会进入到aqs的等待队列，避免了cpu的空转。
　　2.如果在循环的过程中，其他的线程获取到了锁，并且改变了遍历的链表，那么自旋计数器重置为-1，从链表的头节点重新开始遍历。3.get方法public V get(Object key) {     Segment s; // manually integrate access methods to reduce overhead     HashEntry[] tab;     int h = hash(key);     long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;     if ((s = (Segment)UNSAFE.getObjectVolatile(segments, u)) != null &&         (tab = s.table) != null) {         for (HashEntry e = (HashEntry) 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; }
　　get方法并没有加锁，基本思路是：
　　1.先定位到所在的segment
　　2.定位对应的segment的tab数组内的位置
　　3.然后遍历链表元素,如果找到相同的key就返回对应的value 4.总结：
　　ConcurrentHashMap1.7采用了分而治之的思想，将数据分段，每个段持有一把aqs锁。
　　它的写操作都是需要获取锁之后再操作，而读操作不需要获取锁，这也说明它适合读多写少的业务场景。
　　线程在获取不到aqs锁的情况下，不会立即进入到等待队列，会进行一定次数的自旋。