本文共 44103 字，大约阅读时间需要 147 分钟。

目录

---

JDK1.7之前的ConcurrentHashMap使用分段锁机制实现，JDK1.8则使用数组+链表+红黑树数据结构和CAS原子操作实现ConcurrentHashMap；本文将分别介绍这两种方式的实现方案及其区别。

HashMap是线程不安全的，而其它两种HashTable和Collections.synchronizedMap性能又很差，因此在这种并发环境下，为了能够兼顾线程安全以及执行效率，ConcurrentHashMap就应运而出

那么问题来了

• 为什么HashTable慢? 它的并发度是什么? 那么ConcurrentHashMap并发度是什么?
• ConcurrentHashMap在JDK1.7和JDK1.8中实现有什么差别? JDK1.8解決了JDK1.7中什么问题
• ConcurrentHashMap JDK1.7实现的原理是什么? 分段锁机制
• ConcurrentHashMap JDK1.8实现的原理是什么? 数组+链表+红黑树，CAS
• ConcurrentHashMap JDK1.7中Segment数(concurrencyLevel)默认值是多少? 为何一旦初始化就不可再扩容?
• ConcurrentHashMap JDK1.7说说其put的机制?
• ConcurrentHashMap JDK1.7是如何扩容的? rehash(注：segment 数组不能扩容，扩容是 segment 数组某个位置内部的数组 HashEntry[] 进行扩容)
• ConcurrentHashMap JDK1.8是如何扩容的? tryPresize
• ConcurrentHashMap JDK1.8链表转红黑树的时机是什么? 临界值为什么是8?
• ConcurrentHashMap JDK1.8是如何进行数据迁移的? transfer

为什么HashTable慢

Hashtable之所以效率低下主要是因为其实现使用了synchronized关键字对put等操作进行加锁，而synchronized关键字加锁是对整个对象进行加锁，也就是说在进行put等修改Hash表的操作时，锁住了整个Hash表，从而使得其表现的效率低下。

ConcurrentHashMap - JDK 1.7

在JDK1.5~1.7版本，Java使用了分段锁机制实现ConcurrentHashMap.

简而言之，ConcurrentHashMap在对象中保存了一个Segment数组，即将整个Hash表划分为多个分段；而每个Segment元素，即每个分段则类似于一个Hashtable；这样，在执行put操作时首先根据hash算法定位到元素属于哪个Segment，然后对该Segment加锁即可。

因此，ConcurrentHashMap在多线程并发编程中可是实现多线程put操作。接下来分析JDK1.7版本中ConcurrentHashMap的实现原理。

数据结构

整个 ConcurrentHashMap 由一个个 Segment 组成，Segment 代表”部分“或”一段“的意思，所以很多地方都会将其描述为分段锁，也有很多地方用“槽”来代表一个 segment。

简单理解就是，ConcurrentHashMap 是一个 Segment 数组，Segment 通过继承 ReentrantLock 来进行加锁，所以每次需要加锁的操作锁住的是一个 segment，这样只要保证每个 Segment 是线程安全的，也就实现了全局的线程安全。

易懂的可以这样理解：ConcurrentHashMap 其底层可以看做一个二级的 HashMap，第一层存储的是 Segment 数组对象，每个 Segment 存储着一个 HashEntry 数组，其下存储着很多 key-value 键值对

concurrencyLevel: 并行级别、并发数、Segment 数，怎么翻译不重要，理解它。默认是 16，也就是说 ConcurrentHashMap 有 16 个 Segments，所以理论上，这个时候，最多可以同时支持 16 个线程并发写，只要它们的操作分别分布在不同的 Segment 上。这个值可以在初始化的时候设置为其他值，但是一旦初始化以后，它是不可以扩容的。

再具体到每个 Segment 内部，其实每个 Segment 很像之前介绍的 HashMap，不过它要保证线程安全，所以处理起来要麻烦些。

初始化

• initialCapacity: 初始容量，这个值指的是整个 ConcurrentHashMap 的初始容量，实际操作的时候需要平均分给每个 Segment。

• loadFactor: 负载因子，之前我们说了，Segment 数组不可以扩容，所以这个负载因子是给每个 Segment 内部使用的。

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;    // 计算并行级别 ssize，因为要保持并行级别是 2 的 n 次方    while (ssize < concurrencyLevel) {        ++sshift;        ssize <<= 1;    }    // 我们这里先不要那么烧脑，用默认值，concurrencyLevel 为 16，sshift 为 4    // 那么计算出 segmentShift 为 28，segmentMask 为 15，后面会用到这两个值    this.segmentShift = 32 - sshift;    this.segmentMask = ssize - 1;    if (initialCapacity > MAXIMUM_CAPACITY)        initialCapacity = MAXIMUM_CAPACITY;    // initialCapacity 是设置整个 map 初始的大小，    // 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小    // 如 initialCapacity 为 64，那么每个 Segment 或称之为"槽"可以分到 4 个    int c = initialCapacity / ssize;    if (c * ssize < initialCapacity)        ++c;    // 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2，这个值也是有讲究的，因为这样的话，对于具体的槽上，    // 插入一个元素不至于扩容，插入第二个的时候才会扩容    int cap = MIN_SEGMENT_TABLE_CAPACITY;     while (cap < c)        cap <<= 1;    // 创建 Segment 数组，    // 并创建数组的第一个元素 segment[0]    Segment
   
     s0 =        new Segment
    
     (loadFactor, (int)(cap * loadFactor),                         (HashEntry
     
      [])new HashEntry[cap]);    Segment
      
       [] ss = (Segment
       
        [])new Segment[ssize];    // 往数组写入 segment[0]    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]    this.segments = ss;}
       
      
     
    
   

初始化完成，我们得到了一个 Segment 数组。

我们就当是用 new ConcurrentHashMap() 无参构造函数进行初始化的，那么初始化完成后:

• Segment 数组长度为 16，不可以扩容
• Segment[i] 的默认大小为 2，负载因子是 0.75，得出初始阈值为 1.5，也就是以后插入第一个元素不会触发扩容，插入第二个会进行第一次扩容
• 这里初始化了 segment[0]，其他位置还是 null，至于为什么要初始化 segment[0]，后面的代码会介绍
• 当前 segmentShift 的值为 32 - 4 = 28，segmentMask 为 16 - 1 = 15，姑且把它们简单翻译为移位数和掩码，这两个值马上就会用到

 put 过程分析

我们先看 put 的主流程，对于其中的一些关键细节操作，后面会进行详细介绍。

public V put(K key, V value) {    Segment
   
     s;    if (value == null)        throw new NullPointerException();    // 1. 计算 key 的 hash 值    int hash = hash(key);    // 2. 根据 hash 值找到 Segment 数组中的位置 j    //    hash 是 32 位，无符号右移 segmentShift(28) 位，剩下高 4 位，    //    然后和 segmentMask(15) 做一次与操作，也就是说 j 是 hash 值的高 4 位，也就是槽的数组下标    int j = (hash >>> segmentShift) & segmentMask;    // 刚刚说了，初始化的时候初始化了 segment[0]，但是其他位置还是 null，    // ensureSegment(j) 对 segment[j] 进行初始化    if ((s = (Segment
    
     )UNSAFE.getObject          // nonvolatile; recheck         (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment        s = ensureSegment(j);    // 3. 插入新值到 槽 s 中    return s.put(key, hash, value, false);}
    
   

根据 hash 值很快就能找到相应的 Segment，之后就是 Segment 内部的 put 操作了。

Segment 内部是由 数组+链表 组成的。

final V put(K key, int hash, V value, boolean onlyIfAbsent) {    // 在往该 segment 写入前，需要先获取该 segment 的独占锁    //    先看主流程，后面还会具体介绍这部分内容    HashEntry
   
     node = tryLock() ? null :        scanAndLockForPut(key, hash, value);    V oldValue;    try {        // 这个是 segment 内部的数组        HashEntry
    
     [] tab = table;        // 再利用 hash 值，求应该放置的数组下标        int index = (tab.length - 1) & hash;        // first 是数组该位置处的链表的表头        HashEntry
     
       first = entryAt(tab, index);        // 下面这串 for 循环虽然很长，不过也很好理解，想想该位置没有任何元素和已经存在一个链表这两种情况        for (HashEntry
      
        e = first;;) {            if (e != null) {                K k;                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 到底是不是 null，这个要看获取锁的过程，不过和这里都没有关系。                // 如果不为 null，那就直接将它设置为链表表头；如果是null，初始化并设置为链表表头。                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;}
       
      
     
    
   

整体流程还是比较简单的，由于有独占锁的保护，所以 segment 内部的操作并不复杂。至于这里面的并发问题，我们稍后再进行介绍。

到这里 put 操作就结束了，接下来，我们说一说其中几步关键的操作。

 初始化槽: ensureSegment

ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0]，对于其他槽来说，在插入第一个值的时候进行初始化。

这里需要考虑并发，因为很可能会有多个线程同时进来初始化同一个槽 segment[k]，不过只要有一个成功了就可以

private Segment
   
     ensureSegment(int k) {    final Segment
    
     [] ss = this.segments;    long u = (k << SSHIFT) + SBASE; // raw offset    Segment
     
       seg;    if ((seg = (Segment
      
       )UNSAFE.getObjectVolatile(ss, u)) == null) {        // 这里看到为什么之前要初始化 segment[0] 了，        // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k]        // 为什么要用“当前”，因为 segment[0] 可能早就扩容过了        Segment
       
         proto = ss[0];        int cap = proto.table.length;        float lf = proto.loadFactor;        int threshold = (int)(cap * lf);        // 初始化 segment[k] 内部的数组        HashEntry
        
         [] tab = (HashEntry
         
          [])new HashEntry[cap]; if ((seg = (Segment
          
           )UNSAFE.getObjectVolatile(ss, u)) == null) { // 再次检查一遍该槽是否被其他线程初始化了。 Segment
           
             s = new Segment
            
             (lf, threshold, tab); // 使用 while 循环，内部用 CAS，当前线程成功设值或其他线程成功设值后，退出 while ((seg = (Segment
             
              )UNSAFE.getObjectVolatile(ss, u)) == null) { if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; } } } return seg;}
             
            
           
          
         
        
       
      
     
    
   

总的来说，ensureSegment(int k) 比较简单，对于并发操作使用 CAS 进行控制。

 获取写入锁: scanAndLockForPut

前面我们看到，在往某个 segment 中 put 的时候，首先会调用 node = tryLock() ? null : scanAndLockForPut(key, hash, value)，也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁，如果失败，那么进入到 scanAndLockForPut 这个方法来获取锁。

下面我们来具体分析这个方法中是怎么控制加锁的。

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    // 循环获取锁    while (!tryLock()) {        HashEntry
       
         f; // to recheck first below        if (retries < 0) {            if (e == null) {                if (node == null) // speculatively create node                    // 进到这里说明数组该位置的链表是空的，没有任何元素                    // 当然，进到这里的另一个原因是 tryLock() 失败，所以该槽存在并发，不一定是该位置                    node = new HashEntry
        
         (hash, key, value, null); retries = 0; } else if (key.equals(e.key)) retries = 0; else // 顺着链表往下走 e = e.next; } // 重试次数如果超过 MAX_SCAN_RETRIES(单核1多核64)，那么不抢了，进入到阻塞队列等待锁 // lock() 是阻塞方法，直到获取锁后返回 else if (++retries > MAX_SCAN_RETRIES) { lock(); break; } else if ((retries & 1) == 0 && // 这个时候是有大问题了，那就是有新的元素进到了链表，成为了新的表头 // 所以这边的策略是，相当于重新走一遍这个 scanAndLockForPut 方法 (f = entryForHash(this, hash)) != first) { e = first = f; // re-traverse if entry changed retries = -1; } } return node;}
        
       
      
     
    
   

这个方法有两个出口，一个是 tryLock() 成功了，循环终止，另一个就是重试次数超过了 MAX_SCAN_RETRIES，进到 lock() 方法，此方法会阻塞等待，直到成功拿到独占锁。

这个方法就是看似复杂，但是其实就是做了一件事，那就是获取该 segment 的独占锁，如果需要的话顺便实例化了一下 node。

扩容: rehash

重复一下，segment 数组不能扩容，扩容是 segment 数组某个位置内部的数组 HashEntry[] 进行扩容，扩容后，容量为原来的 2 倍。

首先，我们要回顾一下触发扩容的地方，put 的时候，如果判断该值的插入会导致该 segment 的元素个数超过阈值，那么先进行扩容，再插值，读者这个时候可以回去 put 方法看一眼。

该方法不需要考虑并发，因为到这里的时候，是持有该 segment 的独占锁的。

// 方法参数上的 node 是这次扩容后，需要添加到新的数组中的数据。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];    // 新的掩码，如从 16 扩容到 32，那么 sizeMask 为 31，对应二进制 ‘000...00011111’    int sizeMask = newCapacity - 1;    // 遍历原数组，老套路，将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置    for (int i = 0; i < oldCapacity ; i++) {        // e 是链表的第一个元素        HashEntry
       
         e = oldTable[i];        if (e != null) {            HashEntry
        
          next = e.next; // 计算应该放置在新数组中的位置， // 假设原数组长度为 16，e 在 oldTable[3] 处，那么 idx 只可能是 3 或者是 3 + 16 = 19 int idx = e.hash & sizeMask; if (next == null) // 该位置处只有一个元素，那比较好办 newTable[idx] = e; else { // Reuse consecutive sequence at same slot // e 是链表表头 HashEntry
         
           lastRun = e; // idx 是当前链表的头结点 e 的新位置 int lastIdx = idx; // 下面这个 for 循环会找到一个 lastRun 节点，这个节点之后的所有元素是将要放到一起的 for (HashEntry
          
            last = next; last != null; last = last.next) { int k = last.hash & sizeMask; if (k != lastIdx) { lastIdx = k; lastRun = last; } } // 将 lastRun 及其之后的所有节点组成的这个链表放到 lastIdx 这个位置 newTable[lastIdx] = lastRun; // 下面的操作是处理 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); } } } } // 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部 int nodeIndex = node.hash & sizeMask; // add the new node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable;}
             
            
           
          
         
        
       
      
     
    
   

这里的扩容比之前的 HashMap 要复杂一些，代码难懂一点。上面有两个挨着的 for 循环，第一个 for 有什么用呢?

仔细一看发现，如果没有第一个 for 循环，也是可以工作的，但是，这个 for 循环下来，如果 lastRun 的后面还有比较多的节点，那么这次就是值得的。因为我们只需要克隆 lastRun 前面的节点，后面的一串节点跟着 lastRun 走就是了，不需要做任何操作。

我觉得 Doug Lea 的这个想法也是挺有意思的，不过比较坏的情况就是每次 lastRun 都是链表的最后一个元素或者很靠后的元素，那么这次遍历就有点浪费了。不过 Doug Lea 也说了，根据统计，如果使用默认的阈值，大约只有 1/6 的节点需要克隆。

 get 过程分析

相对于 put 来说，get 就很简单了。

• 计算 hash 值，找到 segment 数组中的具体位置，或我们前面用的“槽”
• 槽中也是一个数组，根据 hash 找到数组中具体的位置
• 到这里是链表了，顺着链表进行查找即可

public V get(Object key) {    Segment
   
     s; // manually integrate access methods to reduce overhead    HashEntry
    
     [] tab;    // 1. hash 值    int h = hash(key);    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;    // 2. 根据 hash 找到对应的 segment    if ((s = (Segment
     
      )UNSAFE.getObjectVolatile(segments, u)) != null &&        (tab = s.table) != null) {        // 3. 找到segment 内部数组相应位置的链表，遍历        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;}
       
      
     
    
   

并发问题分析

现在我们已经说完了 put 过程和 get 过程，我们可以看到 get 过程中是没有加锁的，那自然我们就需要去考虑并发问题。

添加节点的操作 put 和删除节点的操作 remove 都是要加 segment 上的独占锁的，所以它们之间自然不会有问题，我们需要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操作。

• put 操作的线程安全性。
  • 初始化槽，这个我们之前就说过了，使用了 CAS 来初始化 Segment 中的数组。
  • 添加节点到链表的操作是插入到表头的，所以，如果这个时候 get 操作在链表遍历的过程已经到了中间，是不会影响的。当然，另一个并发问题就是 get 操作在 put 之后，需要保证刚刚插入表头的节点被读取，这个依赖于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。
  • 扩容。扩容是新创建了数组，然后进行迁移数据，最后面将 newTable 设置给属性 table。所以，如果 get 操作此时也在进行，那么也没关系，如果 get 先行，那么就是在旧的 table 上做查询操作；而 put 先行，那么 put 操作的可见性保证就是 table 使用了 volatile 关键字。
• remove 操作的线程安全性。
  • remove 操作我们没有分析源码，所以这里说的读者感兴趣的话还是需要到源码中去求实一下的。
  • get 操作需要遍历链表，但是 remove 操作会"破坏"链表。
  • 如果 remove 破坏的节点 get 操作已经过去了，那么这里不存在任何问题。
  • 如果 remove 先破坏了一个节点，分两种情况考虑。 1、如果此节点是头结点，那么需要将头结点的 next 设置为数组该位置的元素，table 虽然使用了 volatile 修饰，但是 volatile 并不能提供数组内部操作的可见性保证，所以源码中使用了 UNSAFE 来操作数组，请看方法 setEntryAt。2、如果要删除的节点不是头结点，它会将要删除节点的后继节点接到前驱节点中，这里的并发保证就是 next 属性是 volatile 的。

 ConcurrentHashMap - JDK 1.8

在JDK1.7之前，ConcurrentHashMap是通过分段锁机制来实现的，所以其最大并发度受Segment的个数限制。因此，在JDK1.8中，ConcurrentHashMap的实现原理摒弃了这种设计，而是选择了与HashMap类似的数组+链表+红黑树的方式实现，而加锁则采用CAS和synchronized实现。

较jdk1.7的改进

• 取消了 Segment 的设计，取而代之的是直接的Node对象，使用Node数组来存储数据，并对每个数组中的元素考虑进行加锁
• 底层引入红黑树，当Node元素下的链表长度大于8时，该链表由于过长，查询元素时效率较低(O(n))于是就将其转化为红黑树，从而提高查找效率
• 取消jdk1.7的Segment分段锁机制，改为CAS+Synchronized实现线程安全

并发控制方面，它使用更小粒度的锁——对每个哈希桶的头节点加锁。虽然这样使得效率更高，能让读写操作最大程序的并发执行，但也造成了读写操作的一致性很弱，比如size()返回的大小可能已经与真实大小不一样，比如clear()调用返回后Map中却拥有着元素。

 数据结构

结构上和 Java8 的 HashMap 基本上一样，不过它要保证线程安全性，所以在源码上确实要复杂一些。

源码

常量

    static final int MOVED     = -1; // hash for forwarding nodes

    static final int TREEBIN   = -2; // hash for roots of trees

    static final int RESERVED  = -3; // hash for transient reservations

ConcurrentHashMap中的正常节点的hash值都会是>=0的数，但还有三种特殊节点，它们的hash值可能是上面的三个值。

MOVED，代表此节点是扩容期间的转发节点，这个节点持有新table的引用。

TREEBIN，代表此节点是红黑树的节点。 RESERVED，代表此节点是一个占位节点，不包含任何实际数据。

成员

//存放Node的数组,正常情况下节点都在这个数组里transient volatile Node
   
    [] table;//一个过渡用的table表,在扩容时节点会暂时跑到这个数组上来private transient volatile Node
    
     [] nextTable;//计数器值 = baseCount + 每个CounterCell[i].value。所以baseCount只是计数器的一部分private transient volatile long baseCount;//1. 数组没新建时，暂存容量//2. 数组正在新建时，为-1//3. 正常情况时，存放阈值//4. 扩容时，高16bit存放旧容量唯一对应的一个标签值，低16bit存放进行扩容的线程数量private transient volatile int sizeCtl;//扩容时使用，平时为0，扩容刚开始时为容量，代表下一次领取的扩容任务的索引上界private transient volatile int transferIndex;//CounterCell相配套一个独占锁private transient volatile int cellsBusy;//counterCells也是计数器的一部分private transient volatile CounterCell[] counterCells;
    
   

table。存放节点的数组，只有第一个插入节点时才初始化，默认容量为16。扩容会让容量变成2倍，即保持为2的幂。

nextTable。扩容时需要使用，暂存新table，扩容完毕时，会把新table赋值给table。

sizeCtl。如果使用ConcurrentHashMap的无参构造器，那么它默认为0。其他的值，它都有不同的含义：

数组没新建时，暂存容量。第一个插入节点时，会按照这个容量新建数组。

数组正在新建时，为− 1 -1−1。

正常情况时，存放阈值。阈值 = 0.75 * 容量。

扩容时，高16bit存放旧容量唯一对应的一个标签值，低16bit存放进行扩容的线程数量。

transferIndex。扩容时每个线程通过CAS修改该成员，来领取扩容任务。

baseCount和counterCells。是计数器的实现依赖，类似于LongAdder，它利用ThreadLocalRandom的探针机制来避免频繁的CAS失败，从而减少了因CAS失败而产生的自旋。

 

初始化

// 这构造函数里，什么都不干public ConcurrentHashMap() {}public ConcurrentHashMap(int initialCapacity) {    if (initialCapacity < 0)        throw new IllegalArgumentException();    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?               MAXIMUM_CAPACITY :               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));    this.sizeCtl = cap;}

这个初始化方法有点意思，通过提供初始容量，计算了 sizeCtl，sizeCtl = 【 (1.5 * initialCapacity + 1)，然后向上取最近的 2 的 n 次方】。如 initialCapacity 为 10，那么得到 sizeCtl 为 16，如果 initialCapacity 为 11，得到 sizeCtl 为 32。

sizeCtl 这个属性使用的场景很多，不过只要跟着文章的思路来，就不会被它搞晕了。

但要知道，这里的initialCapacity + (initialCapacity >>> 1) + 1其实是一个，正确的做法应该是下面这个构造器的做法。

public ConcurrentHashMap(int initialCapacity,                             float loadFactor, int concurrencyLevel) {        if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)            throw new IllegalArgumentException();        if (initialCapacity < concurrencyLevel)   // Use at least as many bins            initialCapacity = concurrencyLevel;   // as estimated threads        long size = (long)(1.0 + (long)initialCapacity / loadFactor);        //获得cap的正确做法        int cap = (size >= (long)MAXIMUM_CAPACITY) ?            MAXIMUM_CAPACITY : tableSizeFor((int)size);        this.sizeCtl = cap;    }

put 过程分析

put方法既是难点也是重点。难点在于它涉及到很多操作：初始化数组、插入动作、计数增加、扩容。

public V put(K key, V value) {        return putVal(key, value, false);    }final V putVal(K key, V value, boolean onlyIfAbsent) {    // 如果有空值或空键，会直接抛出异常    if (key == null || value == null) throw new NullPointerException();    // 基于key计算hash值，并进行一定的扰动（目的是使结果分步平均）    // 这个值一定是一个整数，方便后面添加元素，判断该节点的类型    int hash = spread(key.hashCode());    //记录某个桶上元素的个数，如果超过8个，会转成红黑树    int binCount = 0;    for (Node
   
    [] tab = table;;) {        Node
    
      f; int n, i, fh;        //如果数组还未初始化，先对数组进行初始化        if (tab == null || (n = tab.length) == 0)            // 解读源码1，数组初始化            tab = initTable();        // if判断是指，hash函数计算得到的数组下标对应的桶中若为空，就利用cas直接把元素放入数组        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {            // 在这里也使用了cas自旋锁操作。因为有可能有两个线程进入当前位置，确保只能有一个线程访问临界资源            if (casTabAt(tab, i, null,new Node
     
      (hash, key, value, null)))                break;                   // no lock when adding to empty bin        }        // 如果hash计算得到的桶位置元素的hash值为MOVED，证明正在扩容，那么协助扩容        else if ((fh = f.hash) == MOVED)            tab = helpTransfer(tab, f);        else {            V oldVal = null;            // 保证这个位置的桶元素插入时线程安全的，即对桶加锁            // 不影响其他元素的桶位置插入；既保证安全，又不影响效率            // hashtable则是锁了整个数组            synchronized (f) {                // 保证还在该位置，比如变成树或者扩容之后，位置改变了                if (tabAt(tab, i) == f) {                    // 判断hash值大于0 ，就表示当前情况下该位置桶还是链式结构                    if (fh >= 0) {                        binCount = 1;                        // 遍历链表                        for (Node
      
        e = f;; ++binCount) {                            K ek;                            // 如果在链表中找到了put中key值，那么就替换                            if (e.hash == hash &&                                ((ek = e.key) == key ||                                 (ek != null && key.equals(ek)))) {                                oldVal = e.val;                                if (!onlyIfAbsent)                                    e.val = value;                                // 完成替换之后就跳出循环                                break;                            }                            // 如果没有找到该值，就在使用尾插法将Entry插入链表的尾部                            Node
       
         pred = e;                            if ((e = e.next) == null) {                                pred.next = new Node
        
         (hash, key,value, null); break; } } } // 当前位置为树结构，将元素添加到红黑树中 else if (f instanceof TreeBin) { Node
         
           p; binCount = 2; if ((p = ((TreeBin
          
           )f).putTreeVal(hash, key,value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } // 以上就是ConcurrentHashMap添加元素的安全操作 // 从上面代码可以得到，ConcurrentHashMap是通过对桶加锁而不是对整个数组加锁，对效率有提高 if (binCount != 0) { // 如果元素个数大于等于8且数组长度大于64，就变成了树 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); return null;}
          
         
        
       
      
     
    
   

1. 首先进入自旋过程，直到抢占到所该线程put成功
2. 如果数组没有初始化，先进行初始化操作吊桶initTable()方法
3. 如果没有发生哈希冲突，就调用casTabAt()方法，执行CAS操作
4. 此时如果有线程正在执行扩容操作，则扩容操作先一级进行
5. 如果发现哈希冲突，就去抢占锁，当链表时直接尾插，当为红黑树时按其树结构插入
6. 如果插入前是链表，插入结束后链表长度大于8，则将链表转化为红黑树
7. 如最后添加成功则调用addCount()方法统计size，检查是否需要扩容

加锁情况

putVal是写操作，当定位到某个非空的哈希桶，需要对这个哈希桶的头节点synchronized加锁。相反，当定位到的是空的哈希桶，则只需要CAS修改就好了。

对于链表来说，头节点不会变，所以对它加锁好使。

对于红黑树，TreeBin节点封装红黑树的root节点（将root节点作为它的成员），这样，即使红黑树因为平衡操作而改变root，也只是改变了TreeBin节点的一个成员而已。所以对TreeBin节点加锁好使。  

红黑树的binCount固定为2

之所以红黑树的binCount固定为2，是因为：

已经检测到该哈希桶是红黑树了，就不用再树化（binCount >= TREEIFY_THRESHOLD）。

传入addCount(1L, binCount)的第二个参数为2，保证之后能进行扩容检查。 返回情况 返回null，说明putVal执行的新建节点的操作。 返回非null值，说明putVal检测到了重复节点，至于替换与否，根据onlyIfAbsent决定。 如果onlyIfAbsent为false，将替换。否则，不替换。  

通过源码分析可以得到，ConcurrentHashMap插入元素大体来说和HashMap差不多，不同的是，ConcurrentHashMap添加了不少同步操作，如图红色标记，这样就实现了同步安全。

不同于HashTable的是，ConcurrentHashMap主要采用的是CAS自旋锁，提高了效率。

此外，ConcurrentHashMap锁的对象是数组中的每一个桶而不是整个数组，这就意味着，在多线程操作的时候，同一个数组不同的桶之间操作不影响，也就是说，同一个时间，可以有多个线程对数组有插入元素的操作，提高了效率。

 

初始化数组: initTable

这个比较简单，主要就是初始化一个合适大小的数组，然后会设置 sizeCtl。

初始化方法中的并发问题是通过对 sizeCtl 进行一个 CAS 操作来控制的。

// 数组初始化private final Node
   
    [] initTable() {    Node
    
     [] tab; int sc;    // table 表示初始数组    // 进行cas+自旋锁，保证线程安全，对数进行初始化    while ((tab = table) == null || tab.length == 0) {        // 如果sizeCtl小于0，说明此时正在初始化，让出cpu        if ((sc = sizeCtl) < 0)            Thread.yield(); // lost initialization race; just spin        // cas修改sizeCtl的值为-1，如果修改成功，进行数组初始化，如果修改失败，继续自选        // 就是sc和SIZECTL对比        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {            try {                // 完成yield后，sc不是小于0                if ((tab = table) == null || tab.length == 0) {                    // 如果sizeCtl值为0，取默认长度16；否则取sizeCtl中的值                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;                    @SuppressWarnings("unchecked")                    // 基于初始长度，构造数组对象                    Node
     
      [] nt = (Node
      
       [])new Node
       [n];                    table = tab = nt;                    // 计算扩容阈值，并赋值给sc，就是0.75*n                    sc = n - (n >>> 2);                }            } finally {                // 最后将扩容阈值赋值给sizeCtl                sizeCtl = sc;            }            break;        }    }    return tab;}
      
     
    
   

 

链表转红黑树: treeifyBin

前面我们在 put 源码分析也说过，treeifyBin 不一定就会进行红黑树转换，也可能是仅仅做数组扩容。我们还是进行源码分析吧。

private final void treeifyBin(Node
   
    [] tab, int index) {    Node
    
      b; int n, sc;    if (tab != null) {        // MIN_TREEIFY_CAPACITY 为 64        // 所以，如果数组长度小于 64 的时候，其实也就是 32 或者 16 或者更小的时候，会进行数组扩容        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)            // 后面我们再详细分析这个方法            tryPresize(n << 1);        // b 是头结点        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {            // 加锁            synchronized (b) {                if (tabAt(tab, index) == b) {                    // 下面就是遍历链表，建立一颗红黑树                    TreeNode
     
       hd = null, tl = null;                    for (Node
      
        e = b; e != null; e = e.next) {                        TreeNode
       
         p =                            new TreeNode
        
         (e.hash, e.key, e.val, null, null); if ((p.prev = tl) == null) hd = p; else tl.next = p; tl = p; } // 将红黑树设置到数组相应位置中 setTabAt(tab, index, new TreeBin
         
          (hd)); } } } }}
         
        
       
      
     
    
   

扩容: tryPresize

如果说 Java8 ConcurrentHashMap 的源码不简单，那么说的就是扩容操作和迁移操作。

这个方法要完完全全看懂还需要看之后的 transfer 方法，读者应该提前知道这点。

这里的扩容也是做翻倍扩容的，扩容后数组容量为原来的 2 倍。

// 首先要说明的是，方法参数 size 传进来的时候就已经翻了倍了private final void tryPresize(int size) {    // c: size 的 1.5 倍，再加 1，再往上取最近的 2 的 n 次方。    int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :        tableSizeFor(size + (size >>> 1) + 1);    int sc;    while ((sc = sizeCtl) >= 0) {        Node
   
    [] tab = table; int n;        // 这个 if 分支和之前说的初始化数组的代码基本上是一样的，在这里，我们可以不用管这块代码        if (tab == null || (n = tab.length) == 0) {            n = (sc > c) ? sc : c;            if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {                try {                    if (table == tab) {                        @SuppressWarnings("unchecked")                        Node
    
     [] nt = (Node
     
      [])new Node
      [n];                        table = nt;                        sc = n - (n >>> 2); // 0.75 * n                    }                } finally {                    sizeCtl = sc;                }            }        }        else if (c <= sc || n >= MAXIMUM_CAPACITY)            break;        else if (tab == table) {            // 我没看懂 rs 的真正含义是什么，不过也关系不大            int rs = resizeStamp(n);            if (sc < 0) {                Node
      
       [] nt;                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||                    transferIndex <= 0)                    break;                // 2. 用 CAS 将 sizeCtl 加 1，然后执行 transfer 方法                //    此时 nextTab 不为 null                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))                    transfer(tab, nt);            }            // 1. 将 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2)            //     我是没看懂这个值真正的意义是什么? 不过可以计算出来的是，结果是一个比较大的负数            //  调用 transfer 方法，此时 nextTab 参数为 null            else if (U.compareAndSwapInt(this, SIZECTL, sc,                                         (rs << RESIZE_STAMP_SHIFT) + 2))                transfer(tab, null);        }    }}
      
     
    
   

这个方法的核心在于 sizeCtl 值的操作，首先将其设置为一个负数，然后执行 transfer(tab, null)，再下一个循环将 sizeCtl 加 1，并执行 transfer(tab, nt)，之后可能是继续 sizeCtl 加 1，并执行 transfer(tab, nt)。

所以，可能的操作就是执行 1 次 transfer(tab, null) + 多次 transfer(tab, nt)，这里怎么结束循环的需要看完 transfer 源码才清楚。

数据迁移: transfer

下面这个方法有点长，将原来的 tab 数组的元素迁移到新的 nextTab 数组中。

虽然我们之前说的 tryPresize 方法中多次调用 transfer 不涉及多线程，但是这个 transfer 方法可以在其他地方被调用，典型地，我们之前在说 put 方法的时候就说过了，请往上看 put 方法，是不是有个地方调用了 helpTransfer 方法，helpTransfer 方法会调用 transfer 方法的。

此方法支持多线程执行，外围调用此方法的时候，会保证第一个发起数据迁移的线程，nextTab 参数为 null，之后再调用此方法的时候，nextTab 不会为 null。

阅读源码之前，先要理解并发操作的机制。原数组长度为 n，所以我们有 n 个迁移任务，让每个线程每次负责一个小任务是最简单的，每做完一个任务再检测是否有其他没做完的任务，帮助迁移就可以了，而 Doug Lea 使用了一个 stride，简单理解就是步长，每个线程每次负责迁移其中的一部分，如每次迁移 16 个小任务。所以，我们就需要一个全局的调度者来安排哪个线程执行哪几个任务，这个就是属性 transferIndex 的作用。

第一个发起数据迁移的线程会将 transferIndex 指向原数组最后的位置，然后从后往前的 stride 个任务属于第一个线程，然后将 transferIndex 指向新的位置，再往前的 stride 个任务属于第二个线程，依此类推。当然，这里说的第二个线程不是真的一定指代了第二个线程，也可以是同一个线程，这个读者应该能理解吧。其实就是将一个大的迁移任务分为了一个个任务包

private final void transfer(Node
   
    [] tab, Node
    
     [] nextTab) {    int n = tab.length, stride;    // stride 在单核下直接等于 n，多核模式下为 (n>>>3)/NCPU，最小值是 16    // stride 可以理解为”步长“，有 n 个位置是需要进行迁移的，    //   将这 n 个任务分为多个任务包，每个任务包有 stride 个任务    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)        stride = MIN_TRANSFER_STRIDE; // subdivide range    // 如果 nextTab 为 null，先进行一次初始化    //    前面我们说了，外围会保证第一个发起迁移的线程调用此方法时，参数 nextTab 为 null    //       之后参与迁移的线程调用此方法时，nextTab 不会为 null    if (nextTab == null) {        try {            // 容量翻倍            Node
     
      [] nt = (Node
      
       [])new Node
       [n << 1];            nextTab = nt;        } catch (Throwable ex) {      // try to cope with OOME            sizeCtl = Integer.MAX_VALUE;            return;        }        // nextTable 是 ConcurrentHashMap 中的属性        nextTable = nextTab;        // transferIndex 也是 ConcurrentHashMap 的属性，用于控制迁移的位置        transferIndex = n;    }    int nextn = nextTab.length;    // ForwardingNode 翻译过来就是正在被迁移的 Node    // 这个构造方法会生成一个Node，key、value 和 next 都为 null，关键是 hash 为 MOVED    // 后面我们会看到，原数组中位置 i 处的节点完成迁移工作后，    //    就会将位置 i 处设置为这个 ForwardingNode，用来告诉其他线程该位置已经处理过了    //    所以它其实相当于是一个标志。    ForwardingNode
       
         fwd = new ForwardingNode
        
         (nextTab); // advance 指的是做完了一个位置的迁移工作，可以准备做下一个位置的了 boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab /* * 下面这个 for 循环，最难理解的在前面，而要看懂它们，应该先看懂后面的，然后再倒回来看 * */ // i 是位置索引，bound 是边界，注意是从后往前 for (int i = 0, bound = 0;;) { Node
         
           f; int fh; // 下面这个 while 真的是不好理解 // advance 为 true 表示可以进行下一个位置的迁移了 // 简单理解结局: i 指向了 transferIndex，bound 指向了 transferIndex-stride while (advance) { int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; // 将 transferIndex 值赋给 nextIndex // 这里 transferIndex 一旦小于等于 0，说明原数组的所有位置都有相应的线程去处理了 else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { // 看括号中的代码，nextBound 是这次迁移任务的边界，注意，是从后往前 bound = nextBound; i = nextIndex - 1; advance = false; } } if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) { // 所有的迁移操作已经完成 nextTable = null; // 将新的 nextTab 赋值给 table 属性，完成迁移 table = nextTab; // 重新计算 sizeCtl: n 是原数组长度，所以 sizeCtl 得出的值将是新数组长度的 0.75 倍 sizeCtl = (n << 1) - (n >>> 1); return; } // 之前我们说过，sizeCtl 在迁移前会设置为 (rs << RESIZE_STAMP_SHIFT) + 2 // 然后，每有一个线程参与迁移就会将 sizeCtl 加 1， // 这里使用 CAS 操作对 sizeCtl 进行减 1，代表做完了属于自己的任务 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { // 任务结束，方法退出 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; // 到这里，说明 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT， // 也就是说，所有的迁移任务都做完了，也就会进入到上面的 if(finishing){} 分支了 finishing = advance = true; i = n; // recheck before commit } } // 如果位置 i 处是空的，没有任何节点，那么放入刚刚初始化的 ForwardingNode ”空节点“ else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); // 该位置处是一个 ForwardingNode，代表该位置已经迁移过了 else if ((fh = f.hash) == MOVED) advance = true; // already processed else { // 对数组该位置处的结点加锁，开始处理数组该位置处的迁移工作 synchronized (f) { if (tabAt(tab, i) == f) { Node
          
            ln, hn; // 头结点的 hash 大于 0，说明是链表的 Node 节点 if (fh >= 0) { // 下面这一块和 Java7 中的 ConcurrentHashMap 迁移是差不多的， // 需要将链表一分为二， // 找到原链表中的 lastRun，然后 lastRun 及其之后的节点是一起进行迁移的 // lastRun 之前的节点需要进行克隆，然后分到两个链表中 int runBit = fh & n; Node
           
             lastRun = f; for (Node
            
              p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) { ln = lastRun; hn = null; } else { hn = lastRun; ln = null; } for (Node
             
               p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; if ((ph & n) == 0) ln = new Node
              
               (ph, pk, pv, ln); else hn = new Node
               
                (ph, pk, pv, hn); } // 其中的一个链表放在新数组的位置 i setTabAt(nextTab, i, ln); // 另一个链表放在新数组的位置 i+n setTabAt(nextTab, i + n, hn); // 将原数组该位置处设置为 fwd，代表该位置已经处理完毕， // 其他线程一旦看到该位置的 hash 值为 MOVED，就不会进行迁移了 setTabAt(tab, i, fwd); // advance 设置为 true，代表该位置已经迁移完毕 advance = true; } else if (f instanceof TreeBin) { // 红黑树的迁移 TreeBin
                
                  t = (TreeBin
                 
                  )f; TreeNode
                  
                    lo = null, loTail = null; TreeNode
                   
                     hi = null, hiTail = null; int lc = 0, hc = 0; for (Node
                    
                      e = t.first; e != null; e = e.next) { int h = e.hash; TreeNode
                     
                       p = new TreeNode
                      
                        (h, e.key, e.val, null, null); if ((h & n) == 0) { if ((p.prev = loTail) == null) lo = p; else loTail.next = p; loTail = p; ++lc; } else { if ((p.prev = hiTail) == null) hi = p; else hiTail.next = p; hiTail = p; ++hc; } } // 如果一分为二后，节点数少于 8，那么将红黑树转换回链表 ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) : (hc != 0) ? new TreeBin
                       
                        (lo) : t; hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) : (lc != 0) ? new TreeBin
                        
                         (hi) : t; // 将 ln 放置在新数组的位置 i setTabAt(nextTab, i, ln); // 将 hn 放置在新数组的位置 i+n setTabAt(nextTab, i + n, hn); // 将原数组该位置处设置为 fwd，代表该位置已经处理完毕， // 其他线程一旦看到该位置的 hash 值为 MOVED，就不会进行迁移了 setTabAt(tab, i, fwd); // advance 设置为 true，代表该位置已经迁移完毕 advance = true; } } } } }}
                        
                       
                      
                     
                    
                   
                  
                 
                
               
              
             
            
           
          
         
        
       
      
     
    
   

说到底，transfer 这个方法并没有实现所有的迁移任务，每次调用这个方法只实现了 transferIndex 往前 stride 个位置的迁移工作，其他的需要由外围来控制。

这个时候，再回去仔细看 tryPresize 方法可能就会更加清晰一些了。

JDK8中ConcurrentHashMap扩容安全总结

如果是多核CPU的前提下，那么每个线程划分任务，最小任务量是16个桶位的迁移。

在迁移过程中，通过自旋来控制整个过程的持续性，直到所有线程完成扩容任务。

对于桶位来说，如果桶位已经被迁移，会用ForwardingNode占位（这个节点的hash值为-1–MOVED）。使用advance标记线程是否完成扩容。那么，如果说当前迁移的桶位没有元素，那该怎么办呢？在源码中是直接在该位置添加一个fwd节点

在扩容的时候，需要计算下一次任务迁移的开始桶位，并将这个值赋值给transferIndex，这个过程是用cas完成的。

如果当前桶位需要被迁移，就好比在当前桶位插入数据一样，需要使用synchronized关键字来为该桶位加锁，保证多线程安全。

 

addCount

统计ConcurrentHashMap的size的任务，交给了baseCount成员和counterCells数组成员。

当counterCells为null时，baseCount的值就是size。

当counterCells不为null时，baseCount加上每个counterCells的数组元素的值，才是size。 之所以引入counterCells数组，是因为如果每个线程都在CAS修改baseCount这一个int成员的话，必定会造成大量线程因CAS失败而自旋，从而浪费CPU。现在引入一个counterCells数组，让每个数组元素也来承担计数的责任，则线程CAS修改失败的概率下降了不少。

所以，当counterCells不为null时，一定要去优先修改counterCells的某个数组成员的值，而且由于利用了ThreadLocalRandom的probe探针机制来获得数组的随机索引，所以很大概率上，不同线程获得的数组成员是不同的。既然不同线程获得的数组成员不同，那么不同线程尝试CAS修改某个数组成员，肯定不会失败了，从而减小了线程的因CAS失败而导致的自旋。

了解了以上知识，我们再来看addCount和fullAddCount的实现，就好懂多了。

 

//参数x是需要增加的数量。 check用来判断是否需要检测，如果检测到需要扩容，就扩容。    private final void addCount(long x, int check) {        CounterCell[] as; long b, s;        /*计数部分*/        if ((as = counterCells) != null ||  //如果counterCells成员不为null，短路后面条件            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {//如果counterCells成员为null，但CAS失败            CounterCell a; long v; int m;            boolean uncontended = true;            //进入下面这个分支很容易：            //1. 如果as为null 2. 如果as的大小为0(这不可能，只是一种保护)            //3. 如果as的随机索引位置上的元素还没初始化  4. CAS修改一个cell的值失败了            //不进入这个分支很难：只有当 as不为null，且大小不为0，且随机索引位置元素不为null，            //且修改这个cell元素的value成功了，才不会进入分支。            if (as == null || (m = as.length - 1) < 0 ||                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||                !(uncontended =                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {                fullAddCount(x, uncontended);                return;            }            //执行到这里，说明 修改某个cell的value成功了，计数已经成功加上去了。                        //既然修改某个cell的value成功了，而且check参数还这么小，就直接返回，不用去做check动作了            if (check <= 1)                return;            //如果需要check，那么算出当前的size            s = sumCount();        }        //执行到这里，说明上面代码，要么CAS修改baseCount成功了，要么CAS修改某个cell的值成功了        //而且s已经是当前map的映射数量了。        /*扩容部分*/        if (check >= 0) {//如果需要check            Node
   
    [] tab, nt; int n, sc;            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&//大于等于了阈值                   (n = tab.length) < MAXIMUM_CAPACITY) {//当前容量小于最大容量，才可以扩容                int rs = resizeStamp(n);//按照当前容量n，得到与n唯一对应的一个标签值                //如果正在扩容                if (sc < 0) {                    //1. 如果sizeCtl得到的容量标签值，与之前得到的不一样                    //2. 如果扩容线程数为1，说明扩容结束（因为第一个线程，设置线程数量为2，最多增长到MAX_RESIZERS）                    //3. 如果扩容线程数为MAX_RESIZERS，说明可以帮忙扩容的线程的数量到达上限                    //4. 如果nextTable为null，说明扩容结束                    //5. 如果transferIndex <= 0，说明transfer任务被领光了，所有的哈希桶都有线程在帮忙扩容                    //以上情况，都说明当前线程不需要去扩容操作了                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||                        transferIndex <= 0)                        break;                    //如果看起来可以帮忙扩容，那么尝试增加一个线程数量                    //上面分支每个条件都为false才可能执行到这里，说明nt肯定不为null                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))                        //修改成功，相当于拿到许可，开始扩容                        transfer(tab, nt);//nt不为null                }                //如果sc大于0(这里其实不可能等于0),说明当前table没有在扩容，                //当前线程作为第一个线程，所以要设置线程数为2                else if (U.compareAndSwapInt(this, SIZECTL, sc,                                             (rs << RESIZE_STAMP_SHIFT) + 2))                    transfer(tab, null);                s = sumCount();//再次计算size，开始下一次循环。有可能下一次循环，还会继续扩容(如果s >= (long)(sc = sizeCtl))            }        }    }
   

我们把addCount的逻辑分为计数部分和扩容部分。

计数部分

计数部分的if分支有可能根本没有进入，因为当前counterCells成员为null，且修改baseCount成功了。此时计数部分已经完成了任务，并将s局部变量设置为了map的size。

现在考虑进入计数部分的if分支，此时有两种情况：

counterCells成员不为null，短路后面条件。这种情况，接下来会去尝试CAS修改某个cell的值。

如果CAS修改某个cell的值失败了，那么会去执行fullAddCount(x, uncontended)然后直接return掉。 fullAddCount(x, uncontended)传入的uncontended参数此时必为false。（uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)失败） counterCells成员为null，但CAS修改baseCount失败。这种情况一定去执行fullAddCount(x, uncontended)然后直接return掉。因为as == null成立，直接短路后面条件。 fullAddCount(x, uncontended)传入的uncontended参数此时必为true。 （boolean uncontended = true） 所以，当调用fullAddCount(x, uncontended)时，一定是因为之前，尝试CAS修改baseCount失败、或者CAS修改某个cell的值失败了。

fullAddCount(x, uncontended)传入的uncontended参数为false代表，CAS修改某个cell的值失败了。

计数部分结束时

当计数部分结束时，s局部变量的值就已经是map的size了。分为两种情况：

计数部分的if分支根本没有进入，此时counterCells成员为null，所以只需要增加baseCount就好。

计数部分的if分支进入了，而且CAS增加某个cell的值成功了，然后依靠s = sumCount()算出来。因为这种情况，计数分布在baseCount和counterCells数组上。 扩容部分 循环将一直执行，直到扩容结束。扩容中时，条件s >= (long)(sc = sizeCtl)一定成立，因为扩容时sizeCtl为负数。 注意，第一个开始扩容的线程，会将线程数设置为2，线程数最大可以增长到MAX_RESIZERS。之所以这样，是因为线程数为1，用来代表扩容结束。 调用transfer去做真正的扩容。 CAS失败影响扩容 从以上分析可知，不管是CAS修改baseCount失败、或者CAS修改某个cell的值失败了，只要是失败了，之后都会执行fullAddCount然后直接return，从而不去执行扩容部分。

只有CAS成功了，才有可能去调用扩容部分的代码。

这样的做法，可能会导致，该扩容的时候不会扩容。因为毕竟只有CAS直接成功的线程，才可能扩容。比如这种场景，线程A CAS成功，此时大小为阈值-1；然后线程B CAS失败，此时大小刚好为阈值，但由于失败，不会去扩容；只有等到第3个线程到来，才可能去扩容了。

fullAddCount

当addCount函数里CAS修改baseCount失败、或者CAS修改某个cell的值失败了，会调用到fullAddCount。该函数保证能够将x加到baseCount或某个cell上去。  

private final void fullAddCount(long x, boolean wasUncontended) {        int h;        //线程第一次调用fullAddCount，肯定会进入以下分支，因为只有localInit后，probe才不可能为0        //之后调用fullAddCount，就不可能进入这个分支。        if ((h = ThreadLocalRandom.getProbe()) == 0) {//探针为0，说明localInit从来没有调用过            ThreadLocalRandom.localInit();      // 先初始化再说            h = ThreadLocalRandom.getProbe();   // 这句get到的就肯定不为0了            wasUncontended = true;        }        boolean collide = false;        for (;;) {            CounterCell[] as; CounterCell a; int n; long v;            //如果counterCells数组不为null，那么当然优先在某个数组元素增加x            if ((as = counterCells) != null && (n = as.length) > 0) {                //探针取余后的下标的元素为null                if ((a = as[(n - 1) & h]) == null) {                    if (cellsBusy == 0) { //这个cellsBusy相当于一个独占锁的state，为0说明可以获得锁                         CounterCell r = new CounterCell(x); // 乐观地创建，因为假定获得锁后，该索引位置还是为null                        if (cellsBusy == 0 &&                            U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {//尝试获得锁                            boolean created = false;                            try {  // 获得锁后，还是需要检查该索引位置是否还是为null                                CounterCell[] rs; int m, j;                                if ((rs = counterCells) != null &&                                    (m = rs.length) > 0 &&                                    rs[j = (m - 1) & h] == null) {//该索引位置为null，接下来赋值                                    rs[j] = r;                                    created = true;//赋值成功                                }                            } finally {                                cellsBusy = 0;//最终无论如何，都要释放锁                            }                            if (created)//如果赋值成功，那么该函数任务完成                                break;                            continue;  //执行到这里，说明赋值没有成功，因为获得锁后，发现该索引位置却不为null了                        }                    }                    collide = false;                }                //执行到这里，说明探针取余下标的元素不为null                                //从addCount的逻辑来看，只可能addCount里CAS修改某个cell失败了，                //才会导致addCount调用此函数的wasUncontended为false                else if (!wasUncontended)       // 既然之前的探针冲突了，那么就执行advanceProbe获得下一个探针                    wasUncontended = true;                      //CAS修改当前探针指向的数组元素，如果成功了break                else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))                    break;                //如果counterCells更新了，或者数组大小大于CPU个数，短路后面所有if分支，优先移动探针                else if (counterCells != as || n >= NCPU)                    collide = false;                   //短路后面if分支，并消耗掉collide                       else if (!collide)                    collide = true;                //如果没有被前面短路，将进入扩容分支                else if (cellsBusy == 0 &&                         U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {//前提是获得锁成功                    try {                        if (counterCells == as) {//需要再次检查counterCells成员没有变化                            CounterCell[] rs = new CounterCell[n << 1];//扩容二倍                            for (int i = 0; i < n; ++i)                                rs[i] = as[i];//重复利用旧的成员，新位置为null                            counterCells = rs;                        }                    } finally {                        cellsBusy = 0;                    }                    collide = false;                    continue;   // 扩容完事，不移动探针                }                h = ThreadLocalRandom.advanceProbe(h);            }            //执行到这里，说明之前检测到counterCells数组为null            //cellsBusy代表是否可以进行初始化操作            else if (cellsBusy == 0 && counterCells == as &&//这里counterCells和as都为null                     U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {//将初始化许可设置为1                boolean init = false;                try {  //执行初始化操作                    if (counterCells == as) {                        CounterCell[] rs = new CounterCell[2];//建立大小为2的数组                        rs[h & 1] = new CounterCell(x);//探针取余得到下标，只设置一个数组元素(lazy init)                        counterCells = rs;                        init = true;                    }                } finally {                    cellsBusy = 0;                }                if (init)                    break;            }            //执行到这里，说明之前检测到counterCells成员为null，且没有抢到初始化操作的。            //那就退而求其次，去设置baseCount好了。            else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))                break;                          // Fall back on using base        }    }

wasUncontended的作用

wasUncontended为true，代表线程修改某cell无竞争。 wasUncontended为false，代表线程修改某cell有竞争。 不需要关心参数wasUncontended传进来为true的情况，因为这种情况不会造成什么影响。

我们已知，当fullAddCount的参数wasUncontended为false的情况，一定是因为addCount函数里CAS修改某个cell的值失败了（即有竞争）。但这个线程需要分为两种情况：

这种线程第一次进入fullAddCount，则一定会进入if ((h = ThreadLocalRandom.getProbe()) == 0)，然后将wasUncontended设置为true。

这种情况说明之前addCount函数里通过ThreadLocalRandom.getProbe()获得的探针为0，0是作为探针没有初始化的默认值的，也就是说，之前addCount函数里CAS修改某个cell的值失败，很可能是因为没有获得到正常非零的探针值才导致的。所以，获得非零探针值后，就将wasUncontended设置为true，代表之前的线程竞争其实不算事。 这种线程第二次（或以后）进入fullAddCount，则不会进入if ((h = ThreadLocalRandom.getProbe()) == 0)，wasUncontended不会被提前修改掉。 这种情况说明之前addCount函数里通过ThreadLocalRandom.getProbe()获得的探针为非零值，这是一个正常的探针值。正常的探针值，因为线程竞争导致CAS失败了，说明探针需要移动了。  

get 过程分析

get 方法从来都是最简单的，这里也不例外:

• 计算 hash 值
• 根据 hash 值找到数组对应位置: (n - 1) & h
• 根据该位置处结点性质进行相应查找
  • 如果该位置为 null，那么直接返回 null 就可以了
  • 如果该位置处的节点刚好就是我们需要的，返回该节点的值即可
  • 如果该位置节点的 hash 值小于 0，说明正在扩容，或者是红黑树，后面我们再介绍 find 方法
  • 如果以上 3 条都不满足，那就是链表，进行遍历比对即可

public V get(Object key) {    Node
   
    [] tab; Node
    
      e, p; int n, eh; K ek;    int h = spread(key.hashCode());    if ((tab = table) != null && (n = tab.length) > 0 &&        (e = tabAt(tab, (n - 1) & h)) != null) {        // 判断头结点是否就是我们需要的节点        if ((eh = e.hash) == h) {            if ((ek = e.key) == key || (ek != null && key.equals(ek)))                return e.val;        }        // 如果头结点的 hash 小于 0，说明 正在扩容，或者该位置是红黑树        else if (eh < 0)            // 参考 ForwardingNode.find(int h, Object k) 和 TreeBin.find(int h, Object k)            return (p = e.find(h, key)) != null ? p.val : null;        // 遍历链表        while ((e = e.next) != null) {            if (e.hash == h &&                ((ek = e.key) == key || (ek != null && key.equals(ek))))                return e.val;        }    }    return null;}
    
   

扩容不影响并发的读操作

从上面两种数据结构的分离操作来看，旧哈希桶的结构从来没有发生过变化。而对于读操作来说，有一个重要的时间节点，那就是处理哈希桶完毕时执行的setTabAt(tab, i, fwd)。

如果读操作在setTabAt(tab, i, fwd)之前，就获得i索引数组成员的引用，那么读操作就在原哈希桶里进行读操作。

如果读操作在setTabAt(tab, i, fwd)之后，此时已经执行过了setTabAt(nextTab, i, ln)和setTabAt(nextTab, i + n, hn)，说明低桶和高桶已经放到了新table上去了。所以，读操作可以通过ForwardingNode得到新table，从而在新table里继续读操作。

扩容期间，成员的变化

nextTable从null更新为二倍大小数组，transferIndex从0更新为n。 transferIndex不断减小，随着任务的领取。 nextTable成员不断更新，随着每个旧哈希桶分离出来的高桶低桶的赋值。 sizeCtl不断减小，随着线程归还许可。 随着最后一次任务的领取，transferIndex变成0。 随着最后一个完成任务的线程归还许可，sizeCtl的线程数变成1。 nextTable = null，nextTable成员置null。此时table数组的每个非null成员都是ForwardingNode。 table = nextTab，table成员置为新数组。 sizeCtl = (n << 1) - (n >>> 1)，此时sizeCtl才置为阈值。之前的步骤中，sizeCtl都还为负数。 可见在某些步骤上，其他线程可能会发现ConcurrentHashMap处于一种中间状态上。  

简单说一句，此方法的大部分内容都很简单，只有正好碰到扩容的情况，ForwardingNode.find(int h, Object k) 稍微复杂一些，不过在了解了数据迁移的过程后，这个也就不难了，所以限于篇幅这里也不展开说了。

对比总结

• ConcurrentHashMap 在 jdk1.7 与 jdk1.8的不同：
  底层实现不同：1.8中取消了 Segment 的二级HashMap结构，而是使用散列表+链表/红黑树来实现
  线程安全机制不同：1.7中使用 Segment 分段锁机制，1.8中使用 CAS+Synchronized 实现线程安全
  锁的粒度不同：1.7中是对 每个Segment对象加速，1.8中对每个元素，即Node结点加锁
  哈希冲突时存放相同hash值元素的底层数据结构不同，1.8在1.7链表储存的基础上，进行优化当链表长度超过阈值8时，链表会转化为红黑树，从而提高查询效率
   

转载地址：http://kexzb.baihongyu.com/