Java ConcurrentHashMap的源码分析详解

概述

ConcurrentHashMap（CHM）是日常开发中使用频率非常高的一种数据结构，想对于普通的HashMap，CHM提供了线程安全的读写，CHM里面使用了许多比较精妙的优化&操作。本文主要对CHM的整体结构、初始化，查找，插入等做分析。

CHM在1.8之前和之后有比较大的变动，1.8之前主要通过Segment 分段锁 来解决并发问题，1.8及之后就没有这些臃肿的数据结构了，其数据结构与普通的HashMap一样，都是Node数组+链表+红黑树

一颗红黑树应满足如下性质：

1.根节点是黑色的

• 外部节点均为黑色（图中的 leaf 节点，通常在表述的时候会省略）

• 红色节点的孩子节点必为黑色（通常插入的节点为红色）

• 从任一外部节点到根节点的沿途，黑节点的数目相等

除了上面基本的数据结构之外，Node节点也是一个需要关心的数据结构，Node节点本质上是单向链表的节点，其中包含key、value、Hash、next属性

static class Node<K,V> implements Map.Entry<K,V> {
   final int hash;  
   final K key;
   volatile V val;
   volatile Node<K,V> next;
}

ForwardingNode节点

ForwardingNode节点（简称fwd节点）继承自Node节点，主要用于扩容，该节点里面固定Hash值为MOVED(值为-1)，同时持有新表的引用

static final class ForwardingNode<K,V> extends Node<K,V> {
   final Node<K,V>[] nextTable;
   ForwardingNode(Node<K,V>[] tab) {
       super(MOVED, null, null, null);
       this.nextTable = tab;
   }

Node<K,V> find(int h, Object k) {
       ...
   }
}

TreeNode

TreeNode节点也继承自Node节点，用于表示红黑树上的节点，主要属性如下所示

static final class TreeNode<K,V> extends Node<K,V> {
   TreeNode<K,V> parent;  // 父节点
   TreeNode<K,V> left;    // 左儿子
   TreeNode<K,V> right;   // 右儿子
   TreeNode<K,V> prev;    // 记录前驱节点，用于恢复链表
   boolean red;
}

TreeBin

TreeBin节点内部持有TreeNode节点的引用，内部实现了读写锁用于控制多线程并发在红黑树上的操作，主要属性如下所示

static final class TreeBin<K,V> extends Node<K,V> {
   TreeNode<K,V> root;  // 红黑树根节点
   volatile TreeNode<K,V> first;  // 链表根节点，读写分离时会用到
   volatile Thread waiter;  // 当前线程
   volatile int lockState;  // 当前红黑树的锁状态
   // values for lockState
   static final int WRITER = 1; // set while holding write lock
   static final int WAITER = 2; // set when waiting for write lock
   static final int READER = 4; // 读锁标记
}

SizeCtl

除了数据结构需要说明外，SizeCtl也是理解CHM十分重要的一个字段，他是一个整数，不同的值表示不同的状态

• 当SizeCtl > 0时，表示下次扩展的阈值，其中阈值计算方式：数组长度 * 扩展阈值（注意这里是固定的0.75）

• 当SizeCtl = 0时，表示还没有开始初始化

• 当sizeCtl = -1是，表示此时正在进行初始化

• 当SizeCtl < -1时，表示此时正在进行扩展，其中高16位表示扩容标识戳，低16位表示参与扩容的线程数+1

初始化

CHM的初始化是惰性初始化的，即当我们使用ConCurrentHashMap<String,string> map = new ConcurrentHashMap(20);创建一个CHM对象时，并不会真正的创建对象，而是只有在put时才会真正开始创建对象。

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;
}

初始化流程

private final Node<K,V>[] initTable() {
   Node<K,V>[] tab; int sc;
   // 判空，注意这里是while，当线程苏醒后会记性检查直到初始化完毕
   while ((tab = table) == null || tab.length == 0) {
       // 如果其他线程正在初始化，则让出cpu
       if ((sc = sizeCtl) < 0)
           Thread.yield(); // lost initialization race; just spin
       else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { // 当前线程尝试获取创建数组的重任
           try {
               // 这里需要再进行判断是否为空，防止当前线程创建完毕后又有其他线程进来重复创建
               if ((tab = table) == null || tab.length == 0) {
                   int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                   @SuppressWarnings("unchecked")
                       Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                   table = tab = nt;
                   // 设置阈值为0.75n
                   sc = n - (n >>> 2);
               }
           } finally {
               sizeCtl = sc;
           }
           break;
       }
   }
   return tab;
}

查找

get方法进行查找，针对不同情况有不同处理

public V get(Object key) {
   Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 扰动运算
int h = spread(key.hashCode());
// 判断表是否为空，表长度是否为0，以及元素对应下标是否为空
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;
   }
   else if (eh < 0) // 判断Node节点的Hash值是否小于0，如果小于0的话，则会在他的子类上进行查找
       //这里情况比较复杂，不同的节点有不同的处理，如果当前节点为fwd节点，则去新表上找，如果为红黑树
       //节点，则在红黑树上进行查找，后文会展开红黑树上的查找流程
       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;
}

插入

final V putVal(K key, V value, boolean onlyIfAbsent) {
   if (key == null || value == null) throw new NullPointerException();
   int hash = spread(key.hashCode());
   int binCount = 0;
   for (Node<K,V>[] tab = table;;) { // 注意这里是个死循环
       Node<K,V> f; int n, i, fh;
       // 判断是否初始化
       if (tab == null || (n = tab.length) == 0)
           tab = initTable();
       else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 判断对应节点是否是空节点，如果是
           //直接通过cas创建节点
           if (casTabAt(tab, i, null,
                        new Node<K,V>(hash, key, value, null)))
               break;                   // no lock when adding to empty bin
       }
       else if ((fh = f.hash) == MOVED) // 如果当前节点是fwd节点（正在扩容），则帮助扩容
           tab = helpTransfer(tab, f);
       else {
           V oldVal = null;
           synchronized (f) { // 这里加锁
               if (tabAt(tab, i) == f) { // 这里需要继续判断是否当前位置的节点没有变化，因为其他线程可能
                   // 改变此节点
                   if (fh >= 0) { // fh >= 0表示当前节点是链表节点，直接next往下找就行
                       binCount = 1;
                       for (Node<K,V> e = f;; ++binCount) {
                           K ek;
                           if (e.hash == hash &&
                               ((ek = e.key) == key ||
                                (ek != null && key.equals(ek)))) {
                               oldVal = e.val;
                               if (!onlyIfAbsent)
                                   e.val = value;
                               break;
                           }
                           Node<K,V> pred = e;
                           if ((e = e.next) == null) {
                               pred.next = new Node<K,V>(hash, key,
                                                         value, null);
                               break;
                           }
                       }
                   }
                   else if (f instanceof TreeBin) { // 如果此时节点是TreeBin节点，则需要再红黑树上进行插入，具体
                       // 插入流程后文展开
                       Node<K,V> p;
                       binCount = 2;
                       if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                             value)) != null) {
                           oldVal = p.val;
                           if (!onlyIfAbsent)
                               p.val = value;
                       }
                   }
               }
           }
           // 判断是否需要树化
           if (binCount != 0) {
               if (binCount >= TREEIFY_THRESHOLD)
                   treeifyBin(tab, i);
               if (oldVal != null)
                   return oldVal;
               break;
           }
       }
   }
   // 计数器加1，这里使用了计数器而不是AtomicLong这种
   addCount(1L, binCount);
   return null;
}

扩容

CHM的扩容利用了多线程并发的去扩容

CHM在两种条件下会发生扩容：

• 单个链表长度大于8，并且数组长度小于64时，会发生扩容

• 元素个数超过阈值会发生扩容

扩容流程：

• 创建新的Node表，长度为当前数组长度的两倍

• 从后往前分配任务区间，最小长度是16，即每个线程每次扩容最少需要迁移16个桶，具体迁移数量由cpu核数决定

• 判断当前元素是否为空，为空直接cas操作当前节点为fwd节点，否则判断当前元素是否为fwd节点，如果是，则说明其他线程再次区间扩容，此时需要重新选定区间，否则就对当前桶开始进行迁移

• 其他元素在put时如果发现当前桶位是fwd节点，会先协助扩容再put

• 最后一个扩容线程退出扩容时再次检查一遍旧桶，更新sizeCtl的值，同时引用新桶

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
   int n = tab.length, stride;
   // 确定任务长度
   if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
       stride = MIN_TRANSFER_STRIDE; // subdivide range
   if (nextTab == null) {            // 第一个扩容的线程需要创建新数组
       try {
           @SuppressWarnings("unchecked")
               Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
           nextTab = nt;
       } catch (Throwable ex) {      // try to cope with OOME
           sizeCtl = Integer.MAX_VALUE;
           return;
       }
       nextTable = nextTab;
       transferIndex = n;
   }
   int nextn = nextTab.length;
   ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
   boolean advance = true;
   boolean finishing = false; // 用于最后一次检查
   for (int i = 0, bound = 0;;) {
       Node<K,V> f; int fh;
       while (advance) {
           int nextIndex, nextBound;
           if (--i >= bound || finishing) //当前任务是否完成
               advance = false;
           else if ((nextIndex = transferIndex) <= 0) { // 没有任务了
               i = -1;
               advance = false;
           }
           else if (U.compareAndSwapInt
                    (this, TRANSFERINDEX, nextIndex,
                     nextBound = (nextIndex > stride ?
                                  nextIndex - stride : 0))) { //cas更新transferIndex
               bound = nextBound;
               i = nextIndex - 1;
               advance = false;
           }
       }
       if (i < 0 || i >= n || i + n >= nextn) { // 扩容完毕
           int sc;
           if (finishing) { // 二次检查后引用新表
               nextTable = null;
               table = nextTab;
               sizeCtl = (n << 1) - (n >>> 1);
               return;
           }
           if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
               if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                   return;
               finishing = advance = true;
               i = n; // recheck before commit
           }
       }
       else if ((f = tabAt(tab, i)) == null) // 当前节点为空，直接赋为fwd
           advance = casTabAt(tab, i, null, fwd);
       else if ((fh = f.hash) == MOVED) // 其他线程已经迁移好了，此时需要重新分配区间
           advance = true; // already processed
       else {
           synchronized (f) { //当前节点开始迁移，这里需要加锁，可能会有读写操作
               if (tabAt(tab, i) == f) {
                   Node<K,V> ln, hn;
                   if (fh >= 0) { //链表节点 这里只需要判断对应位置是0还是1就可决定迁移到高桶位还是低桶位
                       int runBit = fh & n;
                       Node<K,V> lastRun = f;
                       for (Node<K,V> 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<K,V> 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<K,V>(ph, pk, pv, ln);
                           else
                               hn = new Node<K,V>(ph, pk, pv, hn);
                       }
                       setTabAt(nextTab, i, ln);
                       setTabAt(nextTab, i + n, hn);
                       setTabAt(tab, i, fwd);
                       advance = true;
                   }
                   else if (f instanceof TreeBin) { // 红黑树节点，通过判断对应位是否为0决定放到高
                       // 桶位还是低桶位
                       TreeBin<K,V> t = (TreeBin<K,V>)f;
                       TreeNode<K,V> lo = null, loTail = null;
                       TreeNode<K,V> hi = null, hiTail = null;
                       int lc = 0, hc = 0;
                       for (Node<K,V> e = t.first; e != null; e = e.next) {
                           int h = e.hash;
                           TreeNode<K,V> p = new TreeNode<K,V>
                               (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;
                           }
                       }
                       ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                           (hc != 0) ? new TreeBin<K,V>(lo) : t;
                       hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                           (lc != 0) ? new TreeBin<K,V>(hi) : t;
                       setTabAt(nextTab, i, ln);
                       setTabAt(nextTab, i + n, hn);
                       setTabAt(tab, i, fwd);
                       advance = true;
                   }
               }
           }
       }
   }
}

红黑树的读&写

红黑树上的读写操作是基于TreeBin进行的，上文也对其进行了说明。TreeBin其中的lockState表示当前的读写状态

读操作

读操作和写操作可可以是并行的，当有现成正在写或者正在等待写时，读线程可以读，通过代码我们可以发现，此时并没有从红黑树上去读，而是通过链表去读了，这里和IO多路复用里面的epoll函数的底层原理一样。

final Node<K,V> find(int h, Object k) {
   if (k != null) {
       for (Node<K,V> e = first; e != null; ) {
           int s; K ek;
           // WAITER : .....010
           // WRITER : .....001
           // READER : .....100
           if (((s = lockState) & (WAITER|WRITER)) != 0) { //这里表明此时有正在写或者等待写的线程，直接从链表读
               if (e.hash == h &&
                   ((ek = e.key) == k || (ek != null && k.equals(ek))))
                   return e;
               e = e.next;
           }
           else if (U.compareAndSwapInt(this, LOCKSTATE, s, // 表明此时处于无锁或者读锁状态，直接红黑树上查找
                                        s + READER)) {
               TreeNode<K,V> r, p;
               try {
                   p = ((r = root) == null ? null :
                        r.findTreeNode(h, k, null));
               } finally {
                   Thread w;
                   // 读操作完毕后检查是否有写线程在等待，如果有，需要唤醒等待线程
                   // READER|WAITER 表示此时是最后一个读线程
                   if (U.getAndAddInt(this, LOCKSTATE, -READER) ==
                       (READER|WAITER) && (w = waiter) != null)
                       LockSupport.unpark(w);
               }
               return p;
           }
       }
   }
   return null;
}

写操作

红黑树上的写会先查找是否有对应的值，如果有，则更新值即可，如果没有找到，则插入新的节点，再插入节点的过程中，会调用lockRoot加写锁，如果没有抢到锁，则会调用contentLock方法继续尝试或者将自己挂起

final TreeNode<K,V> putTreeVal(int h, K k, V v) {
   Class<?> kc = null;
   boolean searched = false;
   for (TreeNode<K,V> p = root;;) {
       // 查找是否有值
       int dir, ph; K pk;
       if (p == null) {
           first = root = new TreeNode<K,V>(h, k, v, null, null);
           break;
       }
       else if ((ph = p.hash) > h)
           dir = -1;
       else if (ph < h)
           dir = 1;
       else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
           return p;
       else if ((kc == null &&
                 (kc = comparableClassFor(k)) == null) ||
                (dir = compareComparables(kc, k, pk)) == 0) {
           if (!searched) {
               TreeNode<K,V> q, ch;
               searched = true;
               if (((ch = p.left) != null &&
                    (q = ch.findTreeNode(h, k, kc)) != null) ||
                   ((ch = p.right) != null &&
                    (q = ch.findTreeNode(h, k, kc)) != null))
                   return q;
           }
           dir = tieBreakOrder(k, pk);
       }

TreeNode<K,V> xp = p;
       if ((p = (dir <= 0) ? p.left : p.right) == null) {
           TreeNode<K,V> x, f = first;
           first = x = new TreeNode<K,V>(h, k, v, f, xp);
           if (f != null)
               f.prev = x;
           if (dir <= 0)
               xp.left = x;
           else
               xp.right = x;
           if (!xp.red)
               x.red = true;
           else {
               // 锁住节点，平衡操作可能会导致树结构发生变化
               lockRoot();
               try {
                   root = balanceInsertion(root, x);
               } finally {
                   unlockRoot();
               }
           }
           break;
       }
   }
   assert checkInvariants(root);
   return null;
}

private final void lockRoot() {
   // 这里尝试去获取写锁，获取不到就调用contenedLock方法
   if (!U.compareAndSwapInt(this, LOCKSTATE, 0, WRITER))
       contendedLock(); // offload to separate method
}

private final void contendedLock() {
   boolean waiting = false;
   for (int s;;) {
       // ~WAITER 1111111101
       // 如果此时处于无锁，则重新获取锁
       if (((s = lockState) & ~WAITER) == 0) {
           if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) {
               if (waiting)
                   waiter = null;
               return;
           }
       } // 此时不是处于waiter状态，即其他线程没有等待，则自己进行等待。如果已经有线程在等待了，会一直自旋，也可看出这里是非公平锁
       else if ((s & WAITER) == 0) {
           if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {
               waiting = true;
               waiter = Thread.currentThread();
           }
       }
       else if (waiting) // 将当前线程挂起
           LockSupport.park(this);
   }
}

小结

在红黑树上进行读写时，我们可以发现，当有线程在树上写时，读线程是可以读的，不过不是从红黑树上去读，而不用阻塞，这里可能导致短暂的数据不一致的问题，类似于COW；当有线程在树上读时，此时写线程会将自己挂起，当最后一个读线程查找完毕后会检查是否有些线程在等待，如果有，则唤醒等待写的线程

容器计数

对于一个并发容器来说，当多线程同时写入时，此时容器如何计数成为了一个问题，最简单的是通过AtomicLong来保证原子性与可见性，但是在多线程情况下绝大多数线程会cas失败，然后重试。这无疑是浪费cpu性能的且会有性能瓶颈的。在CHM中引入了，使用分段计数思想，即通过一个数组来计数，当多线程并发计数时，记在数组的不同位置上，最后进行统计。

public int size() {
   long n = sumCount();
   return ((n < 0L) ? 0 :
           (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
           (int)n);
}

final long sumCount() {
   CounterCell[] as = counterCells; CounterCell a;
   long sum = baseCount;
   if (as != null) {
       // 累计cells数组
       for (int i = 0; i < as.length; ++i) {
           if ((a = as[i]) != null)
               sum += a.value;
       }
   }
   return sum;
}

// 计数
private final void addCount(long x, int check) {
   CounterCell[] as; long b, s;
   // 如果cells数组不为空或者cas操作baseCount失败，说明此时出现了竞争，需要再cells数组上计数
   if ((as = counterCells) != null ||
       !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
       CounterCell a; long v; int m;
       boolean uncontended = true;
       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;
       }
       if (check <= 1)
           return;
       s = sumCount();
   }
   // 判断是否需要扩容
   if (check >= 0) {
       Node<K,V>[] tab, nt; int n, sc;
       while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
              (n = tab.length) < MAXIMUM_CAPACITY) {
           int rs = resizeStamp(n);
           if (sc < 0) {
               if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                   sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                   transferIndex <= 0)
                   break;
               if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                   transfer(tab, nt);
           }
           else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                        (rs << RESIZE_STAMP_SHIFT) + 2))
               transfer(tab, null);
           s = sumCount();
       }
   }
}

来源：https://juejin.cn/post/7205455736468668476