首页 > JDK 源码剖析 —— ConcurrentHashMap
头像
何人听我楚狂声
编辑于 2021-02-25 14:55
+ 关注

JDK 源码剖析 —— ConcurrentHashMap

前言

ConcurrentHashMap 是 java.util.concurrent 包(J.U.C)下的一个线程安全的 HashMap。在面试中经常作为进阶内容而被提到,但在内卷的当下,进阶内容就是基本内容……

众所周知,HashMap 是线程不安全的,在多线程同时操作造成 transfer() 时,可能会出现链表成环的情况。在 ConcurrentHashMap 出现之前,可以使用 HashTable 来解决同步问题。HashTable 的实现和 HashMap 基本一样,唯一的区别是,它的几乎所有方法都被声明为 synchronized 方法,这实际上是给 HashMap 加了一个全表锁,强行让并发的操作串行化。

ConcurrentHashMap 在 JDK 1.7 之前使用的是分段锁,将多个链表根结点化作一段,共用一把锁,如果两个线程同时操作不同段,就不会触发同步机制。如下图(图源见水印):

JDK 1.8 之后的 ConcurrentHashMap,跟随 HashMap 的进化,也采用了红黑树机制,同时,细化了锁的粒度,在每个根节点(无论是链表还是红黑树)上都加了锁,这样,只要多个线程之间操作的节点不发生哈希碰撞,就不会触发同步机制。

其他一些小细节,如 ConcurrentHashMap 的 key 不可为 null 等,会在源码中具体讲解。

源码基于 JDK 1.8。

类签名与成员变量

ConcurrentHashMap 的签名如下:

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>, Serializable

和 HashMap 相同,它也继承了 AbstractMap 类,实现了 Serializable 接口。ConcurrentHashMap 又实现了 ConcurrentMap 接口。这个接口表示一个支持并发访问的 Map 集合,它在 Map 接口的基础上又增加了 4 种方法,扩展了原有的 Map 的功能:

V putIfAbsent(K key, V value);                                        //插入元素
boolean remove(Object key, Object value);                    //移除元素
boolean replace(K key, V oldValue, V newValue);        //替换元素
V replace(K key, V value);                                                //替换元素

这四个方法并没有办法体现这个接口的并发安全性,所以可以说这只是一个标志接口。

成员变量如下:

private static final int MAXIMUM_CAPACITY = 1 << 30;
private static final int DEFAULT_CAPACITY = 16;
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
private static final float LOAD_FACTOR = 0.75f;
static final int TREEIFY_THRESHOLD = 8;
static final int UNTREEIFY_THRESHOLD = 6;
static final int MIN_TREEIFY_CAPACITY = 64;
private static final int MIN_TRANSFER_STRIDE = 16;
private static int RESIZE_STAMP_BITS = 16;
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

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
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash

很多变量与 HashMap 中的变量意义基本一致,如 DEFAULT_CAPACITYLOAD_FACTORTREEIFY_THRESHOLD 等。其他的相关变量会在源码使用时说明。

内部存储结构

和 JDK 1.8 的 HashMap 类似,ConcurrentHashMap 的内部节点也分别分为链表节点(Node<K,V>)和红黑树节点(TreeNode<K,V>),其中 TreeNode<K, V> 继承了 Node<K, V>

哈希桶的实现是 transient volatile Node<K, V>[] table,相比于 HashMap 的实现增加了 volatile 关键字,以保证 table 在多线程情况下的可见性。还有一个哈希桶的引用:private transient volatile Node<K,V>[] nextTable,nextTable 会在扩容的时候使用。

还有一个比较关键的属性,private transient volatile int sizeCtl,该属性用于指示 table 数组的初始化状态和扩容大小。当该属性为负数时,table 数组正在被初始化或扩容:-1 时为初始化,-(1+n) 时表示正在有 n 个线程参与扩容。当 table 为 null 时,该属性保存 table 的初始化大小,若是未来会初始化为默认大小(16),则该属性为 0。初始化后,该属性保存了下次扩容需要扩容到的 table 大小,这时这个变量的作用就类似于 HashMap 中的 threshold 的作用。

put()

和 HashMap 类似,put() 也是一个包装方法,具体的实现都在 putVal() 中。

public V put(K key, V value) {
    return putVal(key, value, false);
}

putVal() 方法和 HashMap 中同名方法流程类似,只不过增加了一些小细节以保证线程安全。

    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)
                // table 需要初始化
                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
            }
            // 如果 hash 冲突了,且 hash 值为 MOVED,说明是 ForwardingNode 对象(这是一个占位符对象,保存了扩容后的容器),table 正在扩容,当前线程参与帮助扩容
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            // 如果 hash 冲突了,且 hash 值不为 -1
            else {
                V oldVal = null;
                // 获取 f 节点(根结点)的锁,防止增加链表的时候导致链表成环
                synchronized (f) {
                    // 如果对应的下标位置的节点没有改变
                    if (tabAt(tab, i) == f) {
                        // 并且 f 节点的 hash 值大于等于 0
                        if (fh >= 0) {
                            // 链表初始长度
                            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;
                                }
                            }
                        }
                        // 如果 f 节点的 hash 小于 0 并且 f 是树类型
                        else if (f instanceof 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;
                            }
                        }
                    }
                }
                // 链表长度大于等于 8 时,将该节点改成红黑树
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 判断是否需要扩容
        addCount(1L, binCount);
        return null;
    }

这里只说明一些和 HashMap 不同的地方:

  • 第 6 行有一个死循环,整个添加过程是在死循环中的,直到添加成功才会跳出
  • 第 10 行,如果容器没有初始化,会通过专用的 initTable() 方法进行初始化,后续会分析
  • 第 19 行和 20 行,在扩容迁移的时候,根节点哈希会被设置为 MOVED,其他线程看到时会辅助迁移节点
  • 第 25 行,synchronized 关键字锁住头节点 f,粒度较小

除此以外,还有一些其他地方,如获取节点或者数据加一时,都会采用如 CAS 等方式保证线程安全。

putVal() 简要流程如下:

helpTransfer()

put() 存入键值对的过程当中,如果遇到根节点的哈希值为 MOVED 时,说明当前的 map 正在扩容,这时当前线程就会主动调用 helpTransfer() 方法协助进行节点迁移。

实际上,当根结点的哈希为 MOVED 时,根结点的类型实际上是 ForwardingNode 类型,这个类型是 Node 类型的子类,它只在扩容时会被创建。

helpTransfer() 的实现很短,如下:

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    // 如果 table 不是空,且 node 节点是转移类型,数据检验
    // 且 node 节点的 nextTable(新 table) 不是空,同样也是数据校验
    // 尝试帮助扩容
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
          // 根据 length 得到一个标识符号
        int rs = resizeStamp(tab.length);
          // 如果 nextTab 没有被并发修改 且 tab 也没有被并发修改
          // 且 sizeCtl < 0 (说明还在扩容)
        while (nextTab == nextTable && table == tab &&
               (sc = sizeCtl) < 0) {
              // 如果 sizeCtl 无符号右移 16 不等于 rs(sc 前 16 位如果不等于标识符,则标识符变化了)
              // 或者 sizeCtl == rs + 1 (扩容结束了,不再有线程进行扩容)
              // 或者 sizeCtl == rs + 65535(如果达到最大帮助线程的数量,即 65535)
              // 或者转移下标正在调整(扩容结束)
              // 结束循环,返回 table
            if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                sc == rs + MAX_RESIZERS || transferIndex <= 0)
                break;
            // 如果以上都不是, 将 sizeCtl + 1(表示增加了一个线程帮助其扩容)
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                  // 进行转移
                transfer(tab, nextTab);
                // 结束循环
                break;
            }
        }
        return nextTab;
    }
    return table;
}

代码用了大量的判断,来保证这个线程加入的时候扩容还没有结束,且帮助扩容的线程数量不超限。

在之前讲到 sizeCtl 变量时提到过,当该变量为 -N 时,表示当前有 N-1 个线程正在进行扩容。其中,高 16 位是通过 table.length 生成的标识符,低 16 位就是正在参与扩容的线程数。

在第 19 行有一个判断,sizeCtl == rs + 1 这个条件满足时,表示当前扩容已结束。这其实是第一个发起扩容的线程开始扩容时,会将 sizeCtl 设置成 rs + 2,每个线程协助扩容时,就会将 sizeCtl 加 1,结束扩容时就会将 sizeCtl 减 1,而最初的发起线程在结束时也会将 sizeCtl 减 1,这时,所有线程都完成了扩容,sizeCtl == rs + 1。

transfer()

transfer() 方法用于在 map 扩容时迁移节点,

首先先对流程进行概述:

1. 计算 CPU 核心数和桶个数得出每个线程要处理多少桶,小于 16 时该值为 16
2. 初始化 nextTable,长度为原来的两倍
3. 死循环转移过程,根据 finishing 变量来判断转移是否结束
    1. 进入一个 while 循环,并分配 table 数组中的一个桶给线程,初始值是 16,按从大到小的顺序进行分配,当拿到分配值后,进行 i -- 递减。这个 i 就是数组下标。循环中有一个 bound 变量,指的是当前线程此次循环可以处理的区间的最小下标,超过这个下标,就需要重新领取区间或者结束扩容。还有一个 advance 变量,指的是是否继续递减以转移下一个桶,如果为 false,说明当前桶还没有处理完
    2. 进入 if 判断,判断扩容是否结束,如果扩容结束,清空 nextTable,并更新 table,更新 sizeCtl。如果没完成,但已经无法领取区间,该线程退出该方法,并将 sizeCtl 减一,表示参与扩容的线程少一个。
    3. 如果没有完成任务,且 i 对应的桶为 null,尝试 CAS 插入占位符 ForwardingNode 对象,这样其他线程进行 put 时就可以感知到并参与到扩容中来
    4. 如果对应的桶不是空,有占位符,说明当前已经有线程正在对这个桶进行迁移,当前线程跳过这个桶
    5. 如果以上都不是,说明桶不空,且不是占位符,说明这个桶尚未被处理,当前线程就会开始处理这个桶
4. 在迁移桶的时候是对头节点上锁的,防止有其他线程插入数据,迁移过程与 HashMap 一致(resize 方法)

这部分可以对照着源码阅读,源码的注释会在最后放出。

例如,假设 table 总长度为 64,table 就会被拆分成四个区间,每个线程处理 16 个桶,如果只有 3 个线程正在参与转移,就会有一个区间空闲等待(图源:@莫那·鲁道):

三个线程各自处理,互不影响。这就是 transfer() 方法最精华的部分。

太长不看的源码注释:

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    // 将 length / 8 然后除以 CPU 核心数。如果得到的结果小于 16,那么就使用 16。
    // 这里的目的是让每个 CPU 处理的桶一样多,避免出现转移任务不均匀的现象,如果桶较少的话,默认一个 CPU(一个线程)处理 16 个桶
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range 细分范围 stridea:TODO
    // 新的 table 尚未初始化
    if (nextTab == null) {            // initiating
        try {
            // 扩容  2 倍
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            // 更新
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            // 扩容失败, sizeCtl 使用 int 最大值。
            sizeCtl = Integer.MAX_VALUE;
            return;// 结束
        }
        // 更新成员变量
        nextTable = nextTab;
        // 更新转移下标,就是 老的 tab 的 length
        transferIndex = n;
    }
    // 新 tab 的 length
    int nextn = nextTab.length;
    // 创建一个 fwd 节点,用于占位。当别的线程发现这个槽位中是 fwd 类型的节点,则跳过这个节点。
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // 首次推进为 true,如果等于 true,说明需要再次推进一个下标(i--),反之,如果是 false,那么就不能推进下标,需要将当前的下标处理完毕才能继续推进
    boolean advance = true;
    // 完成状态,如果是 true,就结束此方法。
    boolean finishing = false; // to ensure sweep before committing nextTab
    // 死循环,i 表示下标,bound 表示当前线程可以处理的当前桶区间最小下标
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // 如果当前线程可以向后推进;这个循环就是控制 i 递减。同时,每个线程都会进入这里取得自己需要转移的桶的区间
        while (advance) {
            int nextIndex, nextBound;
            // 对 i 减一,判断是否大于等于 bound (正常情况下,如果大于 bound 不成立,说明该线程上次领取的任务已经完成了。那么,需要在下面继续领取任务)
            // 如果对 i 减一大于等于 bound(还需要继续做任务),或者完成了,修改推进状态为 false,不能推进了。任务成功后修改推进状态为 true。
            // 通常,第一次进入循环,i-- 这个判断会无法通过,从而走下面的 nextIndex 赋值操作(获取最新的转移下标)。其余情况都是:如果可以推进,将 i 减一,然后修改成不可推进。如果 i 对应的桶处理成功了,改成可以推进。
            if (--i >= bound || finishing)
                advance = false;// 这里设置 false,是为了防止在没有成功处理一个桶的情况下却进行了推进
            // 这里的目的是:1. 当一个线程进入时,会选取最新的转移下标。2. 当一个线程处理完自己的区间时,如果还有剩余区间的没有别的线程处理。再次获取区间。
            else if ((nextIndex = transferIndex) <= 0) {
                // 如果小于等于0,说明没有区间了 ,i 改成 -1,推进状态变成 false,不再推进,表示,扩容结束了,当前线程可以退出了
                // 这个 -1 会在下面的 if 块里判断,从而进入完成状态判断
                i = -1;
                advance = false;// 这里设置 false,是为了防止在没有成功处理一个桶的情况下却进行了推进
            }// CAS 修改 transferIndex,即 length - 区间值,留下剩余的区间值供后面的线程使用
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                bound = nextBound;// 这个值就是当前线程可以处理的最小当前区间最小下标
                i = nextIndex - 1; // 初次对i 赋值,这个就是当前线程可以处理的当前区间的最大下标
                advance = false; // 这里设置 false,是为了防止在没有成功处理一个桶的情况下却进行了推进,这样对导致漏掉某个桶。下面的 if (tabAt(tab, i) == f) 判断会出现这样的情况。
            }
        }// 如果 i 小于0 (不在 tab 下标内,按照上面的判断,领取最后一段区间的线程扩容结束)
        //  如果 i >= tab.length(不知道为什么这么判断)
        //  如果 i + tab.length >= nextTable.length  (不知道为什么这么判断)
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            if (finishing) { // 如果完成了扩容
                nextTable = null;// 删除成员变量
                table = nextTab;// 更新 table
                sizeCtl = (n << 1) - (n >>> 1); // 更新阈值
                return;// 结束方法。
            }// 如果没完成
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// 尝试将 sc -1. 表示这个线程结束帮助扩容了,将 sc 的低 16 位减一。
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)// 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了,说明没有线程在帮助他们扩容了。也就是说,扩容结束了。
                    return;// 不相等,说明没结束,当前线程结束方法。
                finishing = advance = true;// 如果相等,扩容结束了,更新 finising 变量
                i = n; // 再次循环检查一下整张表
            }
        }
        else if ((f = tabAt(tab, i)) == null) // 获取老 tab i 下标位置的变量,如果是 null,就使用 fwd 占位。
            advance = casTabAt(tab, i, null, fwd);// 如果成功写入 fwd 占位,再次推进一个下标
        else if ((fh = f.hash) == MOVED)// 如果不是 null 且 hash 值是 MOVED。
            advance = true; // already processed // 说明别的线程已经处理过了,再次推进一个下标
        else {// 到这里,说明这个位置有实际值了,且不是占位符。对这个节点上锁。为什么上锁,防止 putVal 的时候向链表插入数据
            synchronized (f) {
                // 判断 i 下标处的桶节点是否和 f 相同
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;// low, height 高位桶,低位桶
                    // 如果 f 的 hash 值大于 0 。TreeBin 的 hash 是 -2
                    if (fh >= 0) {
                        // 对老长度进行与运算(第一个操作数的的第n位于第二个操作数的第n位如果都是1,那么结果的第n为也为1,否则为0)
                        // 由于 Map 的长度都是 2 的次方(000001000 这类的数字),那么取于 length 只有 2 种结果,一种是 0,一种是1
                        //  如果是结果是0 ,Doug Lea 将其放在低位,反之放在高位,目的是将链表重新 hash,放到对应的位置上,让新的取于算法能够击中他。
                        int runBit = fh & n;
                        Node<K,V> lastRun = f; // 尾节点,且和头节点的 hash 值取于不相等
                        // 遍历这个桶
                        for (Node<K,V> p = f.next; p != null; p = p.next) {
                            // 取于桶中每个节点的 hash 值
                            int b = p.hash & n;
                            // 如果节点的 hash 值和首节点的 hash 值取于结果不同
                            if (b != runBit) {
                                runBit = b; // 更新 runBit,用于下面判断 lastRun 该赋值给 ln 还是 hn。
                                lastRun = p; // 这个 lastRun 保证后面的节点与自己的取于值相同,避免后面没有必要的循环
                            }
                        }
                        if (runBit == 0) {// 如果最后更新的 runBit 是 0 ,设置低位节点
                            ln = lastRun;
                            hn = null;
                        }
                        else {
                            hn = lastRun; // 如果最后更新的 runBit 是 1, 设置高位节点
                            ln = null;
                        }// 再次循环,生成两个链表,lastRun 作为停止条件,这样就是避免无谓的循环(lastRun 后面都是相同的取于结果)
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
                            int ph = p.hash; K pk = p.key; V pv = p.val;
                            // 如果与运算结果是 0,那么就还在低位
                            if ((ph & n) == 0) // 如果是0 ,那么创建低位节点
                                ln = new Node<K,V>(ph, pk, pv, ln);
                            else // 1 则创建高位
                                hn = new Node<K,V>(ph, pk, pv, hn);
                        }
                        // 其实这里类似 hashMap 
                        // 设置低位链表放在新链表的 i
                        setTabAt(nextTab, i, ln);
                        // 设置高位链表,在原有长度上加 n
                        setTabAt(nextTab, i + n, hn);
                        // 将旧的链表设置成占位符
                        setTabAt(tab, i, fwd);
                        // 继续向后推进
                        advance = true;
                    }// 如果是红黑树
                    else if (f instanceof TreeBin) {
                        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);
                            // 和链表相同的判断,与运算 == 0 的放在低位
                            if ((h & n) == 0) {
                                if ((p.prev = loTail) == null)
                                    lo = p;
                                else
                                    loTail.next = p;
                                loTail = p;
                                ++lc;
                            } // 不是 0 的放在高位
                            else {
                                if ((p.prev = hiTail) == null)
                                    hi = p;
                                else
                                    hiTail.next = p;
                                hiTail = p;
                                ++hc;
                            }
                        }
                        // 如果树的节点数小于等于 6,那么转成链表,反之,创建一个新的树
                        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;
                    }
                }
            }
        }
    }
}

结束语

ConcurrentHashMap 就是线程安全的 HashMap,所以在面试时主要就会聚焦其多线程的特性,问的最多的其实就是多线程辅助迁移的过程(transfer() 方法和 helpTransfer() 方法)。在阅读源码时,可以首先对整体有一个大概的把握,这样追究细节的时候也更加容易了。

更多模拟面试

全部评论

(5) 回帖
加载中...
话题 回帖

推荐话题

相关热帖

近期精华帖

热门推荐