1. 简介

我们知道 HashMap 不是线程安全的，HashTable 中的所有方法都是加了 synchronized ，在高并发情况下的性能比较低。还有一种方式 通过 Collections 的 Map<K,V> synchronizedMap(Map<K,V> m) 将 hashMap 包装成一个线程安全的 map。比如 SynchronzedMap 的 put 方法源码为：

  1. public V put(K key, V value) {
  2.     synchronized (mutex) {return m.put(key, value);}
  3. }

ConcurrentHashMap 就是线程安全的 map，其中利用了锁分段的思想提高了并发度。

JDK 1.6 版本关键要素：

  • segment 继承了 ReentrantLock 充当锁的角色，为每一个 segment 提供了线程安全的保障；
  • segment 维护了哈希散列表的若干个桶，每个桶由 HashEntry 构成的链表。

JDK 1.8 版本关键要素

  • 舍弃了 segment，并且大量使用了synchronized，以及 CAS 无锁操作以保证 ConcurrentHashMap 操作的线程安全性。
  • 至于为什么不用 ReentrantLock 而是 Synchronzied 呢？实际上，synchronzied 做了很多的优化，包括偏向锁，轻量级锁，重量级锁，可以依次向上升级锁状态，但不能降级。因此，使用 synchronized 相较于 ReentrantLock 的性能会持平甚至在某些情况更优。
  • 底层数据结构改变为采用 **数组+链表+红黑树** 的数据形式。


2. 关键属性及类

2.1 ConcurrentHashMap 的关键属性

  1. table：**transient volatile Node<K,V>[] table;**
    • 装载 Node 的数组，作为 ConcurrentHashMap 的数据容器，采用懒加载的方式，直到第一次插入数据的时候才会进行初始化操作，数组的大小总是为 2 的幂次方。
  2. nextTable：transient volatile Node<K,V>[] nextTable;
    • 扩容时使用，平时为 null，只有在扩容的时候才为非 null
  3. sizeCtl：**transient volatile int sizeCtl;**
  • 该属性用来控制 table 数组的大小，根据是否初始化和是否正在扩容有几种情况：
    1. 当值为负数时：
      1. 如果为 -1 表示正在初始化
      2. 如果为 -N 则表示当前正有 N-1 个线程进行扩容操作；
    2. 当值为正数时：
      1. 如果当前数组为 null 的话表示 table 在初始化过程中，sizeCtl 表示为需要新建数组的长度；
      2. 若已经初始化了，表示当前数据容器（table 数组）可用容量也可以理解成临界值（插入节点数超过了该临界值就需要扩容），具体指为数组的长度 n 乘以 加载因子 loadFactor；
      3. 当值为 0 时，即数组长度为默认初始值。
  1. Unsafe.u:static final sun.misc.Unsafe U;
    1. U.compareAndSwapXXXX 的方法利用了 CAS 算法保证了线程安全性，这是一种乐观策略，假设每一次操作都不会产生冲突，当且仅当冲突发生的时候再去尝试。
    2. 而 CAS 操作依赖于现代处理器指令集，通过底层 CMPXCHG 指令实现。
    3. CAS(V,O,N)核心思想为：若当前变量实际值 V 与期望的旧值 O 相同，则表明该变量没被其他线程进行修改，因此可以安全的将新值 N 赋值给变量；若当前变量实际值 V 与期望的旧值 O 不相同，则表明该变量已经被其他线程做了处理，此时将新值N赋给变量操作就是不安全的，再进行重试。而在大量的同步组件和并发容器的实现中使用CAS是通过 sun.misc.Unsafe 类实现的，该类提供了一些可以直接操控内存和线程的底层操作。该成员变量的获取是在静态代码块中：
  1.  static {
  2.      try {
  3.          U = sun.misc.Unsafe.getUnsafe();
  4.          .......
  5.      } catch (Exception e) {
  6.          throw new Error(e);
  7.      }
  8.  }

2.2 ConcurrentHashMap 中关键内部类

  • Node 类实现了 Map.Entry 接口，主要存放 key-value 键值对，并且具有 next 域

    1. static class Node<K,V> implements Map.Entry<K,V> {
    2.        final int hash;
    3.        final K key;
    4.        volatile V val;
    5.        volatile Node<K,V> next;
    6.        ......
    7. }

    另外可以看出很多属性都是用 volatile 进行修饰的，也就是为了保证内存可见性。

  • TreeNode 树节点，继承于承载数据的 Node 类。而红黑树的操作是针对 TreeBin 类的，从该类的注释也可以看出，也就是 TreeBin 会将 TreeNode 进行再一次封装

    1. **
    2. * Nodes for use in TreeBins
    3. */
    4. static final class TreeNode<K,V> extends Node<K,V> {
    5.        TreeNode<K,V> parent;  // red-black tree links
    6.        TreeNode<K,V> left;
    7.        TreeNode<K,V> right;
    8.        TreeNode<K,V> prev;    // needed to unlink next upon deletion
    9.        boolean red;
    10.        ......
    11. }


  • TreeBin 这个类并不负责包装用户的 key、value 信息，而是包装的很多 TreeNode 节点。实际的 ConcurrentHashMap 数组中，存放的是 TreeBin 对象，而不是 TreeNode 对象。

    1. static final class TreeBin<K,V> extends Node<K,V> {
    2.        TreeNode<K,V> root;
    3.        volatile TreeNode<K,V> first;
    4.        volatile Thread waiter;
    5.        volatile int lockState;
    6.        // values for lockState
    7.        static final int WRITER = 1; // set while holding write lock
    8.        static final int WAITER = 2; // set when waiting for write lock
    9.        static final int READER = 4; // increment value for setting read lock
    10.        ......
    11. }


  • ForwardingNode 在扩容时才会出现的特殊节点，其 key,value,hash 全部为 null。并拥有 nextTable 指针引用新的 table 数组。

    1. static final class ForwardingNode<K,V> extends Node<K,V> {
    2.    final Node<K,V>[] nextTable;
    3.    ForwardingNode(Node<K,V>[] tab) {
    4.        super(MOVED, null, null, null);
    5.        this.nextTable = tab;
    6.    }
    7.   .....
    8. }

3.3 cas 关键操作

  • tabAt

    1. static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
    2.    return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
    3. }

    该方法用来获取 table 数组中索引为 i 的 Node 元素。

  • casTabAt

    1. static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
    2.                                    Node<K,V> c, Node<K,V> v) {
    3.    return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
    4. }

    利用CAS操作设置 table 数组中索引为i的元素

  • setTabAt

    1. static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
    2.    U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
    3. }

    该方法用来设置 table 数组中索引为 i 的元素

3. 重点方法

3.1 实例构造器方法

ConcurrentHashMap 对象，一共提供了如下几个构造器方法：

  1. // 1. 构造一个空的map，即table数组还未初始化，初始化放在第一次插入数据时，默认大小为16
  2. ConcurrentHashMap()
  3. // 2. 给定map的大小
  4. ConcurrentHashMap(int initialCapacity) 
  5. // 3. 给定一个map
  6. ConcurrentHashMap(Map<? extends K, ? extends V> m)
  7. // 4. 给定map的大小以及加载因子
  8. ConcurrentHashMap(int initialCapacity, float loadFactor)
  9. // 5. 给定map大小，加载因子以及并发度（预计同时操作数据的线程）
  10. ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)

第2种构造器，传入指定大小时的情况，该构造器源码为：

  1. public ConcurrentHashMap(int initialCapacity) {
  2.     // 1. 小于0直接抛异常
  3.     if (initialCapacity < 0)
  4.         throw new IllegalArgumentException();
  5.     // 2. 判断是否超过了允许的最大值，超过了话则取最大值，否则再对该值进一步处理
  6.     int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
  7.                MAXIMUM_CAPACITY :
  8.                tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
  9.     // 3. 赋值给sizeCtl
  10.     this.sizeCtl = cap;
  11. }

当调用构造器方法之后，sizeCtl 的大小应该就代表了 ConcurrentHashMap 的大小，即 table 数组长度。

进入上边的tableSizeFor方法：

  1. /**
  2. * Returns a power of two table size for the given desired capacity.
  3. * See Hackers Delight, sec 3.2
  4. */
  5. private static final int tableSizeFor(int c) {
  6.     int n = c - 1;
  7.     n |= n >>> 1;
  8.     n |= n >>> 2;
  9.     n |= n >>> 4;
  10.     n |= n >>> 8;
  11.     n |= n >>> 16;
  12.     return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
  13. }

该方法会将调用构造器方法时指定的大小转换成一个2的幂次方数，也就是说 ConcurrentHashMap 的大小一定是 2 的幂次方。调用构造器方法的时候并未构造出table数组（可以理解为ConcurrentHashMap的数据容器），只是算出 table 数组的长度，当第一次向 ConcurrentHashMap 插入数据的时候才真正的完成初始化创建 table 数组的工作（懒加载）。

3.2 initTable 方法

直接看源码：

  1. /**
  2. * Initializes table, using the size recorded in sizeCtl.
  3. */
  4. private final Node<K,V>[] initTable() {
  5.     Node<K,V>[] tab; int sc;
  6.     while ((tab = table) == null || tab.length == 0) {
  7.         if ((sc = sizeCtl) < 0)
  8.             // 1. 保证只有一个线程正在进行初始化操作。
  9.             // 因为 sizeCtl < 0 意味着已有线程进行初始化或者扩容
  10.             Thread.yield(); // 让线程变成就绪状态等待 CPU 调度后执行
  11.         else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
  12.             try {
  13.                 if ((tab = table) == null || tab.length == 0) {
  14.                     // 2. 得出数组的大小
  15.                     int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
  16.                     // 3. 这里才真正的初始化数组
  17.                     Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
  18.                     table = tab = nt;
  19.                     // 4. 计算数组中可用的大小：实际大小n*0.75（加载因子）
  20.                     sc = n - (n >>> 2);
  21.                 }
  22.             } finally {
  23.                 sizeCtl = sc;
  24.             }
  25.             break;
  26.         }
  27.     }
  28.     return tab;
  29. }
  • 如果多个线程同时走到这个方法中，为了保证能够正确初始化，在第1步中会先通过 if 进行判断，若当前已经有一个线程正在初始化（即 sizeCtl 值变为 -1），这个时候其他想要初始化的线程调用 Thread.yield() 让出 CPU 时间片然后自旋等待。
  • 正在进行初始化的线程会调用 U.compareAndSwapInt 方法将 sizeCtl 改为 -1 即正在初始化的状态。
  • 另外还需要注意的事情是，在第四步中会进一步计算数组中可用的大小即为数组实际大小 n 乘以加载因子0.75。可以看看这里乘以0.75是怎么算的，0.75为四分之三，这里n - (n >>> 2)是不是刚好是n-(1/4)n=(3/4)n。如果选择是无参的构造器的话，这里在 new Node 数组的时候会使用默认大小为 DEFAULT_CAPACITY（16），然后乘以加载因子 0.75 为 12，也就是说数组的可用大小为 12。

3.3 put 方法

调用 put 方法时实际具体实现是 putVal 方法，源码如下：

  1. public V put(K key, V value) {
  2.     return putVal(key, value, false);
  3. }
  5. final V putVal(K key, V value, boolean onlyIfAbsent) {
  6.     if (key == null || value == null) throw new NullPointerException();
  7.     //1. 计算key的hash值
  8.     int hash = spread(key.hashCode());
  9.     int binCount = 0;
  10.     for (Node<K,V>[] tab = table;;) {
  11.         Node<K,V> f; int n, i, fh;
  12.         //2. 如果当前table还没有初始化先调用initTable方法将tab进行初始化
  13.         if (tab == null || (n = tab.length) == 0)
  14.             tab = initTable();
  15.         //3. tab中索引为i的位置的元素为null，则直接使用CAS将值插入即可
  16.         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  17.             if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
  18.                 break;                   // no lock when adding to empty bin
  19.         }
  20.         //4. 当前正在扩容
  21.         else if ((fh = f.hash) == MOVED)
  22.             tab = helpTransfer(tab, f);
  23.         else {
  24.             V oldVal = null;
  25.             //5.不上以上情况，即初始化完成、没有在扩容、tab中索引为i的位置元素不为null，加锁处理
  26.             synchronized (f) {
  27.                 if (tabAt(tab, i) == f) {
  28.                     //5. 当前为链表，在链表中插入新的键值对
  29.                     if (fh >= 0) {
  30.                         binCount = 1;
  31.                         for (Node<K,V> e = f;; ++binCount) {
  32.                             K ek;
  33.                             if (e.hash == hash &&
  34.                                 ((ek = e.key) == key ||
  35.                                  (ek != null && key.equals(ek)))) {
  36.                                 oldVal = e.val;
  37.                                 if (!onlyIfAbsent)
  38.                                     e.val = value;
  39.                                 break;
  40.                             }
  41.                             Node<K,V> pred = e;
  42.                             if ((e = e.next) == null) {
  43.                                 pred.next = new Node<K,V>(hash, key,
  44.                                                           value, null);
  45.                                 break;
  46.                             }
  47.                         }
  48.                     }
  49.                     // 6.当前为红黑树，将新的键值对插入到红黑树中
  50.                     else if (f instanceof TreeBin) {
  51.                         Node<K,V> p;
  52.                         binCount = 2;
  53.                         if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  54.                                                               value)) != null) {
  55.                             oldVal = p.val;
  56.                             if (!onlyIfAbsent)
  57.                                 p.val = value;
  58.                         }
  59.                     }
  60.                 }
  61.             }
  62.             // 7.插入完键值对后再根据实际大小（大于等于8转为红黑树）看是否需要转换成红黑树
  63.             if (binCount != 0) {
  64.                 if (binCount >= TREEIFY_THRESHOLD) // TREEIFY_THRESHOLD=8
  65.                     treeifyBin(tab, i);
  66.                 if (oldVal != null)
  67.                     return oldVal;
  68.                 break;
  69.             }
  70.         }
  71.     }
  72.     //8.对当前容量大小进行检查，如果超过了临界值（实际大小*加载因子）就需要扩容 
  73.     addCount(1L, binCount);
  74.     return null;
  75. }

逻辑见代码注释，整体而言为了解决线程安全的问题，ConcurrentHashMap 使用了synchronzied + CAS + 红黑树的方式。在 jdk 1.8 之前，ConcurrentHashmap 结构图：
image.png

  1. ConcurrentHashMap 是一个哈希桶数组，如果不出现哈希冲突的时候，每个元素均匀的分布在哈希桶数组中。
  2. 当出现哈希冲突的时候，用标准的链地址的解决方式，将 hash 值相同的节点构成链表的形式，称为「拉链法」，另外，在1.8 版本中为了防止拉链过长，当链表的长度大于8的时候会将链表转换成红黑树。
  3. table 数组中的每个元素实际上是单链表的头结点或者红黑树的根节点。

当插入键值对时首先应该定位到要插入的桶，即插入 table 数组的索引 i 处。如何计算得出索引 i 呢？当然是根据 key 的 hashCode 值。

  • 1. **spread()** 重哈希，以减少 Hash 冲突

对于一个hash表来说，hash 值分散的不够均匀的话会大大增加哈希冲突的概率，从而影响到 hash 表的性能。因此通过 spread 方法进行了一次重hash从而大大减小哈希冲突的可能性。spread 方法为：

  1. static final int spread(int h) {
  2.     return (h ^ (h >>> 16)) & HASH_BITS;
  3. }

该方法主要是将 key 的 hashCode 的低16位于高16位进行异或运算，这样不仅能够使得 hash 值能够分散能够均匀减小 hash 冲突的概率，另外只用到了异或运算，在性能开销上也能兼顾，做到平衡的 trade-off。

  • 2. 初始化 table

第 2 步，会判断当前 table 数组是否初始化了，没有的话就调用 initTable 进行初始化，该方法在上面已经讲过了。

  • 3. 能否将新值直接插入 table 数组

如果插入值待插入的位置刚好所在的 table 数组为 null 的话就可以直接将值插入。

那么怎样根据 hash 确定在 table 中待插入的索引i呢？很显然可以通过 hash 值与数组的长度取模操作，从而确定新值插入到数组的哪个位置。而之前我们提过 ConcurrentHashMap 的大小总是2的幂次方，(n - 1) & hash运算等价于对长度 n 取模，也就是 hash%n，但是位运算比取模运算的效率要高很多，

确定好数组的索引i后，就可以用 tabAt() 方法获取该位置上的元素，如果当前 Node f 为 null 的话，就可以直接用 casTabAt 方法将新值插入即可。

  • 4. 当前是否正在扩容

如果当前节点不为 null，且该节点为特殊节点（forwardingNode）的话，就说明当前 concurrentHashMap 正在进行扩容操作，关于扩容操作，下面会作为一个具体的方法进行讲解。

那么怎样确定当前的这个 Node 是不是特殊的节点了？是通过判断该节点的 hash 值是不是等于 -1（MOVED）,代码为(fh = f.hash) == MOVED，对 MOVED 的解释在源码上也写的很清楚了：

  1. static final int MOVED = -1; // hash for forwarding nodes
  • 5. 当 table[i] 为链表的头结点，在链表中插入新值

在 table[i] 不为 null 且不为forwardingNode 时，并且当前 Node f的 hash 值大于0（fh >= 0）的话说明当前节点 f 为当前桶的所有的节点组成的链表的头结点。那么接下来，要想向 ConcurrentHashMap插入新值的话就是向这个链表插入新值。通过 synchronized (f) 的方式进行加锁以实现线程安全性。往链表中插入节点的部分代码为：

  1. if (fh >= 0) {
  2.     binCount = 1;
  3.     for (Node<K,V> e = f;; ++binCount) {
  4.         K ek;
  5.         // 找到hash值相同的key,覆盖旧值即可
  6.         if (e.hash == hash &&
  7.             ((ek = e.key) == key ||
  8.              (ek != null && key.equals(ek)))) {
  9.             oldVal = e.val;
  10.             if (!onlyIfAbsent)
  11.                 e.val = value;
  12.             break;
  13.         }
  14.         Node<K,V> pred = e;
  15.         if ((e = e.next) == null) {
  16.             //如果到链表末尾仍未找到，则直接将新值插入到链表末尾即可
  17.             pred.next = new Node<K,V>(hash, key, value, null);
  18.             break;
  19.         }
  20.     }
  21. }


  1. 在链表中如果找到了与待插入的键值对的 key 相同的节点，就直接覆盖即可；
  2. 如果直到找到了链表的末尾都没有找到的话，就直接将待插入的键值对追加到链表的末尾即可
  • 6. 当 table[i] 为红黑树的根节点，在红黑树中插入新值
  1. if (f instanceof TreeBin) {
  2.     Node<K,V> p;
  3.     binCount = 2;
  4.     if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) {
  5.         oldVal = p.val;
  6.         if (!onlyIfAbsent)
  7.             p.val = value;
  8.     }
  9. }

首先在 if 中通过 f instanceof TreeBin 判断当前 table[i] 是否是树节点，这下也正好验证了我们在最上面介绍时说的 TreeBin 会对 TreeNode 做进一步封装，对红黑树进行操作的时候针对的是 TreeBin 而不是 TreeNode。

调用 putTreeVal 方法完成向红黑树插入新节点，同样的逻辑，如果在红黑树中存在于待插入键值对的 Key 相同（hash值相等并且equals方法判断为true）的节点的话，就覆盖旧值，否则就向红黑树追加新节点。

  • 7. 根据当前节点个数进行调整

当完成数据新节点插入之后，会进一步对当前链表大小进行调整，这部分代码为：

  1. if (binCount != 0) {
  2.     if (binCount >= TREEIFY_THRESHOLD)
  3.         treeifyBin(tab, i);
  4.     if (oldVal != null)
  5.         return oldVal;
  6.     break;
  7. }

如果当前链表节点个数大于等于 8（TREEIFY_THRESHOLD）的时候，就会调用treeifyBin方法将tabel[i]（第i个散列桶）拉链转换成红黑树。

3.3.1 put 方法总结

整体流程：

  1. 首先对于每一个放入的值，首先利用spread方法对 key 的 hashcode 进行一次 hash 计算，由此来确定这个值在 table 中的位置；
  2. 如果当前table数组还未初始化，先将 table 数组进行初始化操作；
  3. 如果这个位置是 null 的，那么使用 CAS 操作直接放入；
  4. 如果这个位置存在结点，说明发生了 hash 碰撞，首先判断这个节点的类型。如果该节点 fh==MOVED (代表forwardingNode,数组正在进行扩容)的话，说明正在进行扩容；
  5. 如果是链表节点（fh>0），则得到的结点就是 hash 值相同的节点组成的链表的头节点。需要依次向后遍历确定这个新加入的值所在位置。如果遇到 key 相同的节点，则只需要覆盖该结点的 value 值即可。否则依次向后遍历，直到链表尾插入这个结点；
  6. 如果这个节点的类型是 TreeBin 的话，直接调用红黑树的插入方法进行插入新的节点；
  7. 插入完节点之后再次检查链表长度，如果长度大于 8，就把这个链表转换成红黑树；
  8. 对当前容量大小进行检查，如果超过了临界值（实际大小*加载因子）就需要扩容。

3.4 get 方法

  1. public V get(Object key) {
  2.     Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
  3.     // 1. 重hash
  4.     int h = spread(key.hashCode());
  5.     if ((tab = table) != null && (n = tab.length) > 0 &&
  6.         (e = tabAt(tab, (n - 1) & h)) != null) {
  7.         // 2. table[i]桶节点的key与查找的key相同，则直接返回
  8.         if ((eh = e.hash) == h) {
  9.             if ((ek = e.key) == key || (ek != null && key.equals(ek)))
  10.                 return e.val;
  11.         }
  12.         // 3. 当前节点hash小于0说明为树节点，在红黑树中查找即可
  13.         else if (eh < 0)
  14.             return (p = e.find(h, key)) != null ? p.val : null;
  15.         while ((e = e.next) != null) {
  16.         //4. 从链表中查找，查找到则返回该节点的value，否则就返回null即可
  17.             if (e.hash == h &&
  18.                 ((ek = e.key) == key || (ek != null && key.equals(ek))))
  19.                 return e.val;
  20.         }
  21.     }
  22.     return null;
  23. }
  1. 首先先看当前的 hash 桶数组节点即 table[i] 是否为查找的节点，若是则直接返回；
  2. 若不是，则继续再看当前是不是树节点？通过看节点的 hash 值是否为小于 0，如果小于 0 则为树节点。
  3. 如果是树节点在红黑树中查找节点；
  4. 如果不是树节点，那就只剩下为链表的形式的一种可能性了，就向后遍历查找节点，若查找到则返回节点的value即可，若没有找到就返回 null。

3.5 transfer 方法

当 ConcurrentHashMap 容量不足的时候，需要对 table 进行扩容。这个方法的基本思想跟 HashMap 是很像的，但是由于它是支持并发扩容的，所以要复杂的多。原因是它支持多线程进行扩容操作，而并没有加锁。我想这样做的目的不仅仅是为了满足 concurrent 的要求，而是希望利用并发处理去减少扩容带来的时间影响。transfer 方法源码为：

  1. private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
  2.     int n = tab.length, stride;
  3.     if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
  4.         stride = MIN_TRANSFER_STRIDE; // subdivide range
  5.     //1. 新建Node数组，容量为之前的两倍
  6.     if (nextTab == null) {            // initiating
  7.         try {
  8.             @SuppressWarnings("unchecked")
  9.             Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
  10.             nextTab = nt;
  11.         } catch (Throwable ex) {      // try to cope with OOME
  12.             sizeCtl = Integer.MAX_VALUE;
  13.             return;
  14.         }
  15.         nextTable = nextTab;
  16.         transferIndex = n;
  17.     }
  18.     int nextn = nextTab.length;
  19.     //2. 新建forwardingNode引用，在之后会用到
  20.     ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
  21.     boolean advance = true;
  22.     boolean finishing = false; // to ensure sweep before committing nextTab
  23.     for (int i = 0, bound = 0;;) {
  24.         Node<K,V> f; int fh;
  25.         // 3. 确定遍历中的索引i
  26.         while (advance) {
  27.             int nextIndex, nextBound;
  28.             if (--i >= bound || finishing)
  29.                 advance = false;
  30.             else if ((nextIndex = transferIndex) <= 0) {
  31.                 i = -1;
  32.                 advance = false;
  33.             }
  34.             else if (U.compareAndSwapInt
  35.                      (this, TRANSFERINDEX, nextIndex,
  36.                       nextBound = (nextIndex > stride ?
  37.                                    nextIndex - stride : 0))) {
  38.                 bound = nextBound;
  39.                 i = nextIndex - 1;
  40.                 advance = false;
  41.             }
  42.         }
  43.         //4.将原数组中的元素复制到新数组中去
  44.         //4.5 for循环退出，扩容结束修改sizeCtl属性
  45.         if (i < 0 || i >= n || i + n >= nextn) {
  46.             int sc;
  47.             if (finishing) {
  48.                 nextTable = null;
  49.                 table = nextTab;
  50.                 sizeCtl = (n << 1) - (n >>> 1);
  51.                 return;
  52.             }
  53.             if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
  54.                 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
  55.                     return;
  56.                 finishing = advance = true;
  57.                 i = n; // recheck before commit
  58.             }
  59.         }
  60.         //4.1 当前数组中第i个元素为null，用CAS设置成特殊节点forwardingNode(可以理解成占位符)
  61.         else if ((f = tabAt(tab, i)) == null)
  62.             advance = casTabAt(tab, i, null, fwd);
  63.         //4.2 如果遍历到ForwardingNode节点  说明这个点已经被处理过了 直接跳过这里是控制并发扩容的核心
  64.         else if ((fh = f.hash) == MOVED)
  65.             advance = true; // already processed
  66.         else {
  67.             synchronized (f) {
  68.                 if (tabAt(tab, i) == f) {
  69.                     Node<K,V> ln, hn;
  70.                     if (fh >= 0) {
  71.                         //4.3 处理当前节点为链表的头结点的情况，构造两个链表，一个是原链表  另一个是原链表的反序排列
  72.                         int runBit = fh & n;
  73.                         Node<K,V> lastRun = f;
  74.                         for (Node<K,V> p = f.next; p != null; p = p.next) {
  75.                             int b = p.hash & n;
  76.                             if (b != runBit) {
  77.                                 runBit = b;
  78.                                 lastRun = p;
  79.                             }
  80.                         }
  81.                         if (runBit == 0) {
  82.                             ln = lastRun;
  83.                             hn = null;
  84.                         }
  85.                         else {
  86.                             hn = lastRun;
  87.                             ln = null;
  88.                         }
  89.                         for (Node<K,V> p = f; p != lastRun; p = p.next) {
  90.                             int ph = p.hash; K pk = p.key; V pv = p.val;
  91.                             if ((ph & n) == 0)
  92.                                 ln = new Node<K,V>(ph, pk, pv, ln);
  93.                             else
  94.                                 hn = new Node<K,V>(ph, pk, pv, hn);
  95.                         }
  96.                        //在nextTable的i位置上插入一个链表
  97.                        setTabAt(nextTab, i, ln);
  98.                        //在nextTable的i+n的位置上插入另一个链表
  99.                        setTabAt(nextTab, i + n, hn);
  100.                        //在table的i位置上插入forwardNode节点  表示已经处理过该节点
  101.                        setTabAt(tab, i, fwd);
  102.                        //设置advance为true 返回到上面的while循环中 就可以执行i--操作
  103.                        advance = true;
  104.                     }
  105.                     //4.4 处理当前节点是TreeBin时的情况，操作和上面的类似
  106.                     else if (f instanceof TreeBin) {
  107.                         TreeBin<K,V> t = (TreeBin<K,V>)f;
  108.                         TreeNode<K,V> lo = null, loTail = null;
  109.                         TreeNode<K,V> hi = null, hiTail = null;
  110.                         int lc = 0, hc = 0;
  111.                         for (Node<K,V> e = t.first; e != null; e = e.next) {
  112.                             int h = e.hash;
  113.                             TreeNode<K,V> p = new TreeNode<K,V>
  114.                                 (h, e.key, e.val, null, null);
  115.                             if ((h & n) == 0) {
  116.                                 if ((p.prev = loTail) == null)
  117.                                     lo = p;
  118.                                 else
  119.                                     loTail.next = p;
  120.                                 loTail = p;
  121.                                 ++lc;
  122.                             }
  123.                             else {
  124.                                 if ((p.prev = hiTail) == null)
  125.                                     hi = p;
  126.                                 else
  127.                                     hiTail.next = p;
  128.                                 hiTail = p;
  129.                                 ++hc;
  130.                             }
  131.                         }
  132.                         ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
  133.                             (hc != 0) ? new TreeBin<K,V>(lo) : t;
  134.                         hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
  135.                             (lc != 0) ? new TreeBin<K,V>(hi) : t;
  136.                         setTabAt(nextTab, i, ln);
  137.                         setTabAt(nextTab, i + n, hn);
  138.                         setTabAt(tab, i, fwd);
  139.                         advance = true;
  140.                     }
  141.                 }
  142.             }
  143.         }
  144.     }
  145. }

代码逻辑请看注释，整个扩容操作分为两个部分：

第一部分是构建一个 nextTable，它的容量是原来的 2 倍，这个操作是单线程完成的。新建 table 数组的代码为:Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]，在原容量大小的基础上左移一位。

第二部分就是将原来 table 中的元素复制到 nextTable 中，主要是遍历复制的过程。 根据运算得到当前遍历的数组的位置 i，然后利用 tabAt 方法获得i位置的元素再进行判断：

  1. 如果这个位置为空，就在原 table 中的 i 位置放入forwardNode节点，这个也是触发并发扩容的关键点；
  2. 如果这个位置是 Node 节点（fh>=0），如果它是一个链表的头节点，就构造一个反序链表，把他们分别放在 nextTable 的 i 和 i+n 的位置上
  3. 如果这个位置是TreeBin 节点（fh<0），也做一个反序处理，并且判断是否需要 untreefi，把处理的结果分别放在 nextTable 的 i 和 i+n 的位置上
  4. 遍历过所有的节点以后就完成了复制工作，这时让 nextTable 作为新的 table，并且更新 sizeCtl 为新容量的 0.75倍 ，完成扩容。设置为新容量的 0.75 倍代码为 sizeCtl = (n << 1) - (n >>> 1)，仔细体会下是不是很巧妙，n<<1 相当于 n 右移一位表示 n 的两倍即 2n,n>>>1 左右一位相当于 n 除以 2 即 0.5n,然后两者相减为 2n-0.5n=1.5n，是不是刚好等于新容量的 0.75倍即 2n*0.75=1.5n。最后用一个示意图来进行总结：

image.png

3.6 与 size 相关的一些方法

对于 ConcurrentHashMap 来说，这个 table 里到底装了多少东西其实是个不确定的数量，因为不可能在调用 **size()** 方法的时候像 GC 的「stop the world」一样让其他线程都停下来让你去统计，因此只能说这个数量是个估计值。为了统计元素个数，ConcurrentHashMap 定义了一些变量和一个内部类

  1. /**
  2.  * A padded cell for distributing counts.  Adapted from LongAdder
  3.  * and Striped64.  See their internal docs for explanation.
  4.  */
  5. @sun.misc.Contended static final class CounterCell {
  6.     volatile long value;
  7.     CounterCell(long x) { value = x; }
  8. }
  10. /******************************************/ 
  12. /**
  13.  * 实际上保存的是 hashmap 中的元素个数  利用CAS锁进行更新
  14.  但它并不用返回当前 hashmap 的元素个数 
  16.  */
  17. private transient volatile long baseCount;
  18. /**
  19.  * Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
  20.  */
  21. private transient volatile int cellsBusy;
  23. /**
  24.  * Table of counter cells. When non-null, size is a power of 2.
  25.  */
  26. private transient volatile CounterCell[] counterCells;
  • mappingCount 与 size 方法
  1. public int size() {
  2.     long n = sumCount();
  3.     return ((n < 0L) ? 0 :
  4.             (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
  5.             (int)n);
  6. }
  7.  /**
  8.  * Returns the number of mappings. This method should be used
  9.  * instead of {@link #size} because a ConcurrentHashMap may
  10.  * contain more mappings than can be represented as an int. The
  11.  * value returned is an estimate; the actual count may differ if
  12.  * there are concurrent insertions or removals.
  13.  *
  14.  * @return the number of mappings
  15.  * @since 1.8
  16.  */
  17. public long mappingCount() {
  18.     long n = sumCount();
  19.     return (n < 0L) ? 0L : n; // ignore transient negative values
  20. }
  22.  final long sumCount() {
  23.     CounterCell[] as = counterCells; CounterCell a;
  24.     long sum = baseCount;
  25.     if (as != null) {
  26.         for (int i = 0; i < as.length; ++i) {
  27.             if ((a = as[i]) != null)
  28.                 sum += a.value;//所有counter的值求和
  29.         }
  30.     }
  31.     return sum;
  32. }

使用一个 volatile 类型的变量baseCount 记录元素的个数，当插入新数据或则删除数据时，会通过 addCount() 方法更新 baseCount

  • addCount 方法

在put方法结尾处调用了addCount方法，把当前 ConcurrentHashMap 的元素个数 +1 这个方法一共做了两件事：更新 baseCount 的值；检测是否进行扩容。

  1. private final void addCount(long x, int check) {
  2.     CounterCell[] as; long b, s;
  3.     //利用CAS方法更新baseCount的值 
  4.     if ((as = counterCells) != null ||
  5.         !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
  6.         CounterCell a; long v; int m;
  7.         boolean uncontended = true;
  8.         if (as == null || (m = as.length - 1) < 0 ||
  9.             (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
  10.             !(uncontended =
  11.               U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
  12.             fullAddCount(x, uncontended);
  13.             return;
  14.         }
  15.         if (check <= 1)
  16.             return;
  17.         s = sumCount();
  18.     }
  19.     //如果check值大于等于0 则需要检验是否需要进行扩容操作
  20.     if (check >= 0) {
  21.         Node<K,V>[] tab, nt; int n, sc;
  22.         while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
  23.                (n = tab.length) < MAXIMUM_CAPACITY) {
  24.             int rs = resizeStamp(n);
  25.             //
  26.             if (sc < 0) {
  27.                 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  28.                     sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
  29.                     transferIndex <= 0)
  30.                     break;
  31.                  //如果已经有其他线程在执行扩容操作
  32.                 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
  33.                     transfer(tab, nt);
  34.             }
  35.             //当前线程是唯一的或是第一个发起扩容的线程  此时nextTable=null
  36.             else if (U.compareAndSwapInt(this, SIZECTL, sc,
  37.                                          (rs << RESIZE_STAMP_SHIFT) + 2))
  38.                 transfer(tab, null);
  39.             s = sumCount();
  40.         }
  41.     }
  42. }

4. 总结

JDK 6、7 中的 ConcurrentHashmap 主要使用 Segment 来实现减小锁粒度，分割成若干个 Segment。其中Segment 在实现上继承了ReentrantLock，这样就自带了锁的功能。jdk1.7 中采用 Segment + HashEntry 的方式进行实现，结构如下：
image.png
在 put 的时候需要锁住 Segment，get 时候不加锁，使用 volatile 来保证可见性，当要统计全局时（比如 size），首先会尝试多次计算 modcount 来确定，这几次尝试中，是否有其他线程进行了修改操作，如果没有，则直接返回 size。如果有，则需要依次锁住所有的 Segment 来计算。

**1.8** 之前 put 定位节点时要先定位到具体的 segment，然后再在 segment 中定位到具体的桶。而在 1.8 的时候摒弃了 segment 臃肿的设计，取而代之的是采用 **Node** + **CAS** + **Synchronized**，直接针对的是 Node[] tale 数组中的每一个桶，进一步减小了锁粒度。并且防止拉链过长导致性能下降，当链表长度大于 8 的时候采用红黑树的设计。
image.png
主要设计上的变化有以下几点:

  1. 不采用 segment 而采用 node，锁住 node 来实现减小锁粒度。
  2. 设计了 MOVED 状态 当 resize 的中过程中 线程 2 还在 put 数据，线程 2 会帮助 resize。
  3. 使用 3 个 CAS 操作来确保 node 的一些操作的原子性，这种方式代替了锁。
  4. sizeCtl 的不同值来代表不同含义，起到了控制的作用。
  5. 采用 synchronized 而不是 ReentrantLock

更多关于 1.7 版本与 1.8 版本的 ConcurrentHashMap 的实现对比，可以参考这篇文章。
另外这篇文章也不错：《吊打面试官》系列-ConcurrentHashMap & Hashtable