结构

与 HashMap 类似，使用数组 + 链表 + 红黑树存储键值对
ConcurrentHashMap源码学习基于JDK1.8 - 图1

成员变量

transient volatile Node[] table; // 存放 Node，第一次插入数据时候进行初始化，长度为 2 的倍数

private transient volatile Node[] nextTable; // 仅扩容时用到，将节点迁移到新数组

private static final int MIN_TRANSFER_STRIDE = 16 // 扩容线程每次最少要迁移16 个 hash 桶，在扩容中，参与的单个线程允许处理的最少 table 桶首节点个数，虽然适当添加线程，会使得整个扩容过程变快，但需要考虑多线程内存同时分配的问题

private transient volatile int sizeCtl;

  • 默认为 0

  • -1 表示正在初始化

  • -(1+number) 表示有 number 个线程同时在扩容，线程必须竞争到这个共享变量，才能进行初始化或者扩容。

  • 正数， table 中元素数量阈值，超过这个阈值就会扩容
    ConcurrentHashMap源码学习基于JDK1.8 - 图2
    static final int MOVED = -1; // ForwardingNode 的 hash 值，为 -1

    • ForwardingNode，在扩容时使用，如果 index 处 Node 节点 hash 值为 -1，表示正在扩容。

static final int TREEBIN = -2; // 树节点的 Hash 值

static final int RESERVED = -3; // 临时保留的 Hash 值

static final int HASH_BITS = 0x7fffffff; // usable bits of normal

哈希桶 Table 初始化

  • 根据共享变量 sizeCtl 的值来决定是否由当前线程执行初始化操作（单线程进行初始化），若 sizeCtl < 0 表示正在扩容，该线程需要等待。

  • sizeCtl 为正数时，表示 table 的阈值（= 0.75*n），元素个数超过这个值将会扩容。

  1. private final Node<K,V>[] initTable() {
  2.     Node<K,V>[] tab; int sc;
  3.     while ((tab = table) == null || tab.length == 0) {
  4.         // 如果共享变量 sizeCtl < 0，说明有其它线程正在初始化或者扩容，让出 CPU，让其它线程先执行完
  5.         if ((sc = sizeCtl) < 0)
  6.             Thread.yield(); // lost initialization race; just spin
  7.         // CAS 方式获取到锁，那么由该线程进行初始化
  8.         else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
  9.             try {
  10.                    // 将 table 赋值给 tab，判断 tab 是否已经初始化
  11.                 if ((tab = table) == null || tab.length == 0) {
  12.                     // sc = sizeCtl > 0 表示已经初始化，否则使用默认容量 16。
  13.                     int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
  14.                     @SuppressWarnings("unchecked")
  15.                     Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
  16.                     table = tab = nt;
  17.                     // sc = 0.75*n，sc 表示阈值
  18.                     sc = n - (n >>> 2);
  19.                 }
  21.             } finally {
  22.                 sizeCtl = sc;
  23.             }
  24.             break;
  25.         }
  26.     }
  27.     return tab;
  28. }

key 对应到哈希桶的过程

  • hash 过程与 HashMap 类似，只是多了与 HASH_BITS 进行位运算

  • HASH_BITS 除了首位是 0，剩下的都是 1，按位与，得正数(首位为0)；它的作用就是为了让上面的 hash 值为正数

  1. // index 表示桶的位置
  2. int index = spread(key.hashCode) & (length - 1)
  4. static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
  6. static final int spread(int h) {
  7.     return (h ^ (h >>> 16)) & HASH_BITS;
  8. }

get 方法

  1. key，value 都不能为 null，key 为 null 时抛出 NullPointerException

  2. 将 key 进行 hash 然后找到数组位置处的索引 index

  3. 比较 Hash 值，并且若 index == key || equals 返回 true，直接返回 index 处元素

  4. 若 index 处的 hash 值小于 0，表示正在扩容，进一步调用 Node 子类的 find 方法

  5. index 处的 hash 值不小于 0，通过链表 Next 指针遍历查找

  1. static final int TREEBIN   = -2; // hash for roots of trees
  3. public V get(Object key) {
  4.     Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
  5.     // 计算 hash 值
  6.     int h = spread(key.hashCode());
  7.     // 数组不为空 && 数组长度大于 0 && 该位置已经有元素
  8.     if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) {
  9.         // 如果 key 的 hash 值一样，则进一步判断
  10.         if ((eh = e.hash) == h) {
  11.             // 如果是同一个 key 或者 equal 方法返回 true
  12.             if ((ek = e.key) == key || (ek != null && key.equals(ek)))
  13.                 // 返回对应 value
  14.                 return e.val;  
  15.         }
  16.         // 如果 hash 值小于 0，表示数组正在扩容
  17.         // 此处 Node 节点为 ForwardingNode，调用 ForwardingNode 的 find 方法
  18.         else if (eh < 0)
  19.             return (p = e.find(h, key)) != null ? p.val : null;
  20.         // eh > 0 情况，链表遍历    
  21.         while ((e = e.next) != null) {
  22.             if (e.hash == h &&
  23.                 ((ek = e.key) == key || (ek != null && key.equals(ek))))
  24.                 return e.val;
  25.         }
  26.     }
  27.     // 没有元素
  28.     return null;
  29. }

ForwardingNode 介绍

ForwardingNode 是 Node 子类之一，当节点的 Hash 值为 -1 时，该节点就为 ForwardingNode ，标记此处正在扩容

  • 外层循环用于刷新 ForwardingNode 所在的 table 情况，因为扩容过程中 table 的节点情况在变化
  • 内层循环则去以 next 指针的形式遍历 table，直到找到 hashCode 相同 && （== || equals ）返回 true 的节点
  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.     }
  8.     Node<K,V> find(int h, Object k) {
  9.         // loop to avoid arbitrarily deep recursion on forwarding nodes
  10.         outer: for (Node<K,V>[] tab = nextTable;;) {
  11.             Node<K,V> e; int n;
  12.             // 先判断 null
  13.             if (k == null || tab == null || (n = tab.length) == 0 ||
  14.                 (e = tabAt(tab, (n - 1) & h)) == null)
  15.                 return null;
  16.             for (;;) {
  17.                 int eh; K ek;
  18.                 // 若 Hash 值相同进一步判断变量是否相同，之间返回
  19.                 if ((eh = e.hash) == h &&
  20.                     ((ek = e.key) == k || (ek != null && k.equals(ek))))
  21.                     return e;
  22.                 if (eh < 0) {
  23.                     if (e instanceof ForwardingNode) {
  24.                         tab = ((ForwardingNode<K,V>)e).nextTable;
  25.                         continue outer;
  26.                     }
  27.                     else
  28.                         return e.find(h, k);  // 链表形式遍历
  29.                 }
  30.                 if ((e = e.next) == null)
  31.                     return null;
  32.             }
  33.         }
  34.     }
  35. }

put 方法

  1. table 为 null 或者 tab.size = 0，进行初始化

  2. key 进行 hash 之后取模得到的索引位置，若在桶的位置元素为 null，那么直接插入元素。

  3. 桶位置元素不为 null，那么进行比较

  4. 判断 key 是否相同：

    • 如果 key 相同，e 保存该节点
    • 如果 key 不同，如果该是红黑树节点，那么执行红黑树的 put 方法；如果是链表节点，执行链表遍历操作，找到对应的节点并用 e 保存，如果链表长度大于 >=7，就将链表转为红黑树

  5. Node e 保存找到的节点，如果没有找到返回 null

  6. 如果 size 超过阈值，进行扩容

  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.     // key value 均不能为 null
  7.     if (key == null || value == null) throw new NullPointerException();
  8.     int hash = spread(key.hashCode());
  9.     int binCount = 0;
  10.     // 获取当前 table，进入死循环，直至插入成功
  11.     for (Node<K,V>[] tab = table;;) {
  12.         Node<K,V> f; int n, i, fh;
  13.         // 为空初始化数组
  14.         if (tab == null || (n = tab.length) == 0)
  15.             tab = initTable();
  16.         // 索引所在位置没有元素，通过 casTabAt 方法插入元素并跳出循环
  17.         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  18.             if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))){
  19.                 break;                   // no lock when adding to empty bin
  20.             }
  21.         }
  22.         // 索引位置处是 ForwardingNode，该线程帮助扩容
  23.         else if ((fh = f.hash) == MOVED)
  24.             tab = helpTransfer(tab, f);
  25.         // 该位置处节点为 普通节点，锁住该链表头结点并在尾部添加节点
  26.         else {
  27.             V oldVal = null;
  28.             synchronized (f) {
  29.                 // 加锁之后，再判断索引位置处 key 是否相同
  30.                 if (tabAt(tab, i) == f) {
  31.                     // hash >= 0，通过链表遍历方式找到并替换节点
  32.                     if (fh >= 0) {
  33.                         binCount = 1;
  34.                         for (Node<K,V> e = f;; ++binCount) {
  35.                             K ek;
  36.                             if (e.hash == hash &&
  37.                                 ((ek = e.key) == key ||
  38.                                  (ek != null && key.equals(ek)))) {
  39.                                 oldVal = e.val;
  40.                                 if (!onlyIfAbsent)
  41.                                     // 进行替换
  42.                                     e.val = value;
  43.                                 break;
  44.                             }
  45.                             Node<K,V> pred = e;
  46.                             if ((e = e.next) == null) {
  47.                                 // 链表尾部插入节点
  48.                                 pred.next = new Node<K,V>(hash, key,
  49.                                                           value, null);
  50.                                 break;
  51.                             }
  52.                         }
  53.                     }
  54.                     // index 处索引小于 0 （hash = -2 TREEBIN），并且是树节点，执行树的插入方法
  55.                     else if (f instanceof TreeBin) {
  56.                         Node<K,V> p;
  57.                         binCount = 2;
  58.                         if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  59.                                                        value)) != null) {
  60.                             oldVal = p.val;
  61.                             if (!onlyIfAbsent)
  62.                                 p.val = value;
  63.                         }
  64.                     }
  65.                 }
  66.             }
  67.             // binCount != 0 说明向链表或者红黑树中添加或修改一个节点成功
  68.             // binCount  == 0 说明 put 操作将一个新节点添加成为某个桶的首节点
  69.             if (binCount != 0) {
  70.                 // >= 8 执行转红黑树的操作
  71.                 if (binCount >= TREEIFY_THRESHOLD)
  72.                     treeifyBin(tab, i);
  73.                 // oldVal != null 说明此次操作是修改操作，直接返回旧值，无需下面扩容检查
  74.                 if (oldVal != null)
  75.                     return oldVal;
  76.                 break;
  77.             }
  78.         }
  79.     }
  80.     // 判断是否需要扩容
  81.     addCount(1L, binCount);
  82.     return null;
  83. }

helpTransfer 帮助扩容

  1. // 返回二进制最高位到第一个非 0 数字之间 0 的个数
  2. static final int resizeStamp(int n) {
  3.     return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
  4. }
  6. // 帮助扩容
  7. final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
  8.     Node<K,V>[] nextTab; int sc;
  9.     // tab != null && ForwardingNode && ForwardingNode.nextTable != null
  10.     if (tab != null && (f instanceof ForwardingNode) &&
  11.         (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
  12.         // 生成标志位 rs
  13.         int rs = resizeStamp(tab.length);
  14.         while (nextTab == nextTable && table == tab &&
  15.                (sc = sizeCtl) < 0) {
  16.             // 如果 sizeCtl 无符号右移 16 位与上面标志位不同
  17.             // 或者 sizeCtl == rs + 1  （扩容结束了，不再有线程进行扩容）
  18.             // 或者 sizeCtl == rs + 65535  （如果达到最大帮助线程的数量，即 65535）
  19.             // 或者转移下标正在调整 （扩容结束）
  20.             // 结束循环
  21.             if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  22.                 sc == rs + MAX_RESIZERS || transferIndex <= 0)
  23.                 break;
  24.             // 以上情况不满足， sizeCtl++ ，增加一个线程进行扩容
  25.             if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
  26.                 // 复制或者移动 bins 里面的 Node 到新的 table
  27.                 transfer(tab, nextTab);
  28.                 break;
  29.             }
  30.         }
  31.         return nextTab;
  32.     }
  33.     return table;
  34. }
  37. // 移动或者复制 Node 到新的 table
  38. private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
  39.     int n = tab.length, stride;
  40.     // 计算单个线程允许处理的最少 table 桶首节点个数，不能小于 16
  41.     if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
  42.         stride = MIN_TRANSFER_STRIDE; // subdivide range
  43.     // 如果刚开始扩容，就初始化 nextTab，长度为原来的两倍
  44.     if (nextTab == null) {            // initiating
  45.         try {
  46.             @SuppressWarnings("unchecked")
  47.             Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
  48.             nextTab = nt;
  49.         } catch (Throwable ex) {      // try to cope with OOME
  50.             sizeCtl = Integer.MAX_VALUE;
  51.             return;
  52.         }
  53.         nextTable = nextTab;
  54.         // transferIndex 指向最后一个桶，方便从后向前遍历
  55.         transferIndex = n;
  56.     }
  57.     int nextn = nextTab.length;
  58.     // fwd 作为标记，标记那些已经完成迁移的桶
  59.     ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
  60.     boolean advance = true;
  61.     boolean finishing = false; // to ensure sweep before committing nextTab
  62.     // i 指向当前桶，bound 指向当前线程需要处理的桶结点的区间下限
  63.     for (int i = 0, bound = 0;;) {
  64.         Node<K,V> f; int fh;
  65.         while (advance) {
  66.             int nextIndex, nextBound;
  67.             if (--i >= bound || finishing)
  68.                 advance = false;
  69.             // transferIndex 本来指向最后一个桶，小于等于 0 说明处理完成    
  70.             else if ((nextIndex = transferIndex) <= 0) {
  71.                 i = -1;
  72.                 advance = false;
  73.             }
  74.             // 更新 transferIndex，处理的桶区间为 (nextBound,nextIndex)
  75.             else if (U.compareAndSwapInt
  76.                      (this, TRANSFERINDEX, nextIndex,
  77.                       nextBound = (nextIndex > stride ?
  78.                                    nextIndex - stride : 0))) {
  79.                 bound = nextBound;
  80.                 i = nextIndex - 1;
  81.                 advance = false;
  82.             }
  83.         }
  84.         // 当前线程任务处理完成
  85.         if (i < 0 || i >= n || i + n >= nextn) {
  86.             int sc;
  87.             if (finishing) {
  88.                 nextTable = null;
  89.                 table = nextTab;
  90.                 sizeCtl = (n << 1) - (n >>> 1);
  91.                 return;
  92.             }
  93.             if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
  94.                 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
  95.                     return;
  96.                 finishing = advance = true;
  97.                 i = n; // recheck before commit
  98.             }
  99.         }
  100.         // 索引位置为 null，设置 ForwardingNode 表示处理过
  101.         else if ((f = tabAt(tab, i)) == null)
  102.             advance = casTabAt(tab, i, null, fwd);
  103.         // 该桶已经处理过
  104.         else if ((fh = f.hash) == MOVED)
  105.             advance = true; // already processed
  106.         else {
  107.             // 锁住该桶
  108.             synchronized (f) {
  109.                 // 再判断下桶有没有发生变化
  110.                 if (tabAt(tab, i) == f) {
  111.                     Node<K,V> ln, hn;
  112.                     if (fh >= 0) {  // 普通的 Node 节点
  113.                         int runBit = fh & n; // 移动步长，为 0 或者 N
  114.                         Node<K,V> lastRun = f;
  115.                         // 整个 for 循环为了找到整个桶中最后连续的 fh & n 不变的结点
  116.                         for (Node<K,V> p = f.next; p != null; p = p.next) {
  117.                             int b = p.hash & n;
  118.                             if (b != runBit) {
  119.                                 runBit = b;
  120.                                 lastRun = p;
  121.                             }
  122.                         }
  123.                         if (runBit == 0) {
  124.                             ln = lastRun;
  125.                             hn = null;
  126.                         }
  127.                         else {
  128.                             hn = lastRun;
  129.                             ln = null;
  130.                         }
  131.                         // 如果 fh&n 不变的链表的 runbit 都是 0，则 nextTab[i] 内元素 ln 前逆序，ln 及其之后顺序
  132.                         // 否则，nextTab[i+n]内元素全部相对原table逆序
  133.                         // 这是通过一个节点一个节点的往 nextTab 添加
  134.                         for (Node<K,V> p = f; p != lastRun; p = p.next) {
  135.                             int ph = p.hash; K pk = p.key; V pv = p.val;
  136.                             if ((ph & n) == 0)
  137.                                 ln = new Node<K,V>(ph, pk, pv, ln);
  138.                             else
  139.                                 hn = new Node<K,V>(ph, pk, pv, hn);
  140.                         }
  141.                         // 把两条链表整体迁移到 nextTab 中
  142.                         setTabAt(nextTab, i, ln);
  143.                         setTabAt(nextTab, i + n, hn);
  144.                         // 将原桶标识位已经处理
  145.                         setTabAt(tab, i, fwd);
  146.                         advance = true;
  147.                     }
  148.                     // 红黑树的复制算法
  149.                     else if (f instanceof TreeBin) {
  150.                         TreeBin<K,V> t = (TreeBin<K,V>)f;
  151.                         TreeNode<K,V> lo = null, loTail = null;
  152.                         TreeNode<K,V> hi = null, hiTail = null;
  153.                         int lc = 0, hc = 0;
  154.                         for (Node<K,V> e = t.first; e != null; e = e.next) {
  155.                             int h = e.hash;
  156.                             TreeNode<K,V> p = new TreeNode<K,V>
  157.                                 (h, e.key, e.val, null, null);
  158.                             if ((h & n) == 0) {
  159.                                 if ((p.prev = loTail) == null)
  160.                                     lo = p;
  161.                                 else
  162.                                     loTail.next = p;
  163.                                 loTail = p;
  164.                                 ++lc;
  165.                             }
  166.                             else {
  167.                                 if ((p.prev = hiTail) == null)
  168.                                     hi = p;
  169.                                 else
  170.                                     hiTail.next = p;
  171.                                 hiTail = p;
  172.                                 ++hc;
  173.                             }
  174.                         }
  175.                         // 节点数不够时退化为链表
  176.                         ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
  177.                             (hc != 0) ? new TreeBin<K,V>(lo) : t;
  178.                         hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
  179.                             (lc != 0) ? new TreeBin<K,V>(hi) : t;
  180.                         setTabAt(nextTab, i, ln);
  181.                         setTabAt(nextTab, i + n, hn);
  182.                         setTabAt(tab, i, fwd);
  183.                         advance = true;
  184.                     }
  185.                 }
  186.             }
  187.         }
  188.     }
  189. }
  • sizeCtl == rs + 1 扩容结束了，不再有线程进行扩容，这个判断可以在 addCount 方法中找到答案：默认第一个线程设置 sc == rs 左移 16 位 + 2，当第一个线程结束扩容了，就会将 sc 减一。这个时候，sc 就等于 rs + 1。

如果 sizeCtl == 标识符 + 1 ，说明扩容结束了，没有必要再扩容了。

扩容时机

  • addCount 方法判断节点数量是否达到阈值
  • treeifyBin 转树操作时，当数组长度还未超过 64 时，先进行扩容

size()

主要是 sumCount 的实现，元素个数 = baseCount + counterCells[i].value

当插入或删除数据时，会通过 addCount() 方法以 CAS 方式更新 baseCount 的值，当 CAS 更新失败，会调用 fullAddCount 方法进而更新 counterCells 里面的值。

其中内部类 CounterCell 被 @Contended 注解修饰，是为了防止 伪共享。

缓存系统中是以缓存行（cache line）为单位存储的。缓存行是2的整数幂个连续字节，一般为32-256个字节。最常见的缓存行大小是64个字节。当多线程修改互相独立的变量时，如果这些变量共享同一个缓存行，就会无意中影响彼此的性能，这就是伪共享。

JDK 1.8 建议使用 mappingCount 方法(返回 long) 代替 size() 方法返回元素个数，因为元素的个数可能大于 int 表示的范围。

  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. }
  8. final long sumCount() {
  9.     CounterCell[] as = counterCells; CounterCell a;
  10.     long sum = baseCount;
  11.     if (as != null) {
  12.         for (int i = 0; i < as.length; ++i) {
  13.             if ((a = as[i]) != null)
  14.                 sum += a.value;
  15.         }
  16.     }
  17.     return sum;
  18. }
  20. @sun.misc.Contended static final class CounterCell {
  21.     volatile long value;
  22.     CounterCell(long x) { value = x; }
  23. }
  25. public long mappingCount() {
  26.     long n = sumCount();
  27.     return (n < 0L) ? 0L : n; // ignore transient negative values
  28. }

博客参考

并发编程——ConcurrentHashMap#helpTransfer() 分析

为并发而生的 ConcurrentHashMap（Java 8）

https://www.iteye.com/blog/wujiu-2378812