不亏是阿里三面,ConcurrentHashMap多线程扩容机制被面试官装到了

简介: 不亏是阿里三面,ConcurrentHashMap多线程扩容机制被面试官装到了

一、引言

ConcurrentHashMap 技术在互联网技术使用如此广泛,几乎所有的后端技术面试官都要在 ConcurrentHashMap 技术的使用和原理方面对小伙伴们进行 360° 的刁难。

作为一个在互联网公司面一次拿一次 Offer 的面霸,打败了无数竞争对手,每次都只能看到无数落寞的身影失望的离开,略感愧疚(请允许我使用一下夸张的修辞手法)。

于是在一个寂寞难耐的夜晚,暖男我痛定思痛,决定开始写 《吊打面试官》 系列,希望能帮助各位读者以后面试势如破竹,对面试官进行 360° 的反击,吊打问你的面试官,让一同面试的同僚瞠目结舌,疯狂收割大厂 Offer

虽然现在是互联网寒冬,但乾坤未定,你我皆是黑马!

本篇文章大约读取时间十分钟,带你彻底掌握 ConcurrentHashMap魔鬼式的多线程扩容 逻辑,目录如下:

二、扩容流程

我们上一篇文章讲了关于 ConcurrentHashMap 的存储流程:美团二面:聊聊ConcurrentHashMap的存储流程,熟悉了其数据是如何的放到数组、链表、红黑树

今天这篇文章,我们重点讲一下 ConcurrentHashMap 魔鬼式的多线程扩容逻辑

系好安全带,我们发车了!

1、treeifyBin方法触发扩容

// 在链表长度大于等于8时,尝试将链表转为红黑树
private final void treeifyBin(Node<K,V>[] tab, int index) {
    Node<K,V> b; int n, sc;
    // 数组不能为空
    if (tab != null) {
        // 数组的长度n,是否小于64
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            // 如果数组长度小于64,不能将链表转为红黑树,先尝试扩容操作
            // 这里的扩容传入的数据是当前数据的2倍
            tryPresize(n << 1);
        // 省略部分代码……
    }
}

2、tryPreSize方法-针对putAll的初始化操作

private final void tryPresize(int size) {
    // 如果当前传过来的size大于最大值MAXIMUM_CAPACITY >>> 1,则取 MAXIMUM_CAPACITY 
    // 反之取 tableSizeFor(size + (size >>> 1) + 1)= size的1.5倍 + 1
    // tableSizeFor:这个方法我们之前也讲过,找size的最近二次幂
    // 这个地方的c
    int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>> 1) + 1);
    int sc;
    // 当前的sizeCtl大于等于0,证明不属于异常状态
    // static final int MOVED     = -1; // 代表当前hash位置的数据正在扩容!
    // static final int TREEBIN   = -2; // 代表当前hash位置下挂载的是一个红黑树
    // static final int RESERVED  = -3; // 预留当前索引位置……
    while ((sc = sizeCtl) >= 0) {
        Node<K,V>[] tab = table; int n;
        // 如果当前的数组为空,走初始化的逻辑
        // 有人可能会好奇,这里怎么会为空呢?
        // 大家可以去看下putAll的源码(放在下面),源码里面第一句调用的就是tryPresize的方法
        if (tab == null || (n = tab.length) == 0) {
            // 进来执行初始化
            // sc是初始化长度,初始化长度如果比计算出来的c要大的话,直接使用sc,如果没有sc大,
            // 说明sc无法容纳下putAll中传入的map,使用更大的数组长度
            n = (sc > c) ? sc : c;
            // 还是老配方,CAS将SIZECTL尝试从sc变成-1
            if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                try {
                    // Step1:判断引用
                    // Step2:创建数组
                    // Step3:修改sizeCtl数值
                    if (table == tab) {
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = nt;
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
            }
        }
        // 如果计算出来的长度c如果小于等于sc,直接退出循环结束方法
        // 数组长度大于等于最大长度了,直接退出循环结束方法
        // 不需要扩容
        else if (c <= sc || n >= MAXIMUM_CAPACITY)
            break;
        // 省略部分代码
    }
}
// putAll方法
public void putAll(Map<? extends K, ? extends V> m) {
    // 直接走扩容的逻辑
    tryPresize(m.size());
    for (Map.Entry<? extends K, ? extends V> e : m.entrySet())
        putVal(e.getKey(), e.getValue(), false);
}

3、tryPreSize方法-计算扩容戳

private final void tryPresize(int size) {
    int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
        tableSizeFor(size + (size >>> 1) + 1);
    int sc;
    while ((sc = sizeCtl) >= 0) {
        Node<K,V>[] tab = table;
        // 当前数组的长度
        int n; 
        // 判断引用是否一致
        if (tab == table) {
            // 计算扩容标识戳,根据当前数组的长度计算一个16位的扩容戳
            // 1、保证后面的SIZECTL赋值是负数
            // 2、记录当前从什么长度开始扩容
            int rs = resizeStamp(n);
            // 这里存在bug
            // 由于我们上面(sc = sizeCtl) >= 0的判断,sc只能大于0
            // 这里永远大于等于0
            if (sc < 0) {
                // 协助扩容的代码
            }
            // 代表当前没有线程进行扩容,我是第一个扩容的
            else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
        }
    }
}
// 计算扩容标识戳
static final int resizeStamp(int n) {
    return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}

4、tryPreSize方法-对sizeCtl的修改

private final void tryPresize(int size) {
    // sc默认为sizeCtl
    while ((sc = sizeCtl) >= 0) {
        else if (tab == table) {
            // rs:扩容戳  00000000 00000000 10000000 00011010
            int rs = resizeStamp(n);
            if (sc < 0) {
                // 说明有线程正在扩容,过来帮助扩容
                // 我们之前第一个线程扩容的时候,将sizeCtl设置成:10000000 00011010 00000000 00000010
                // 所以:sc = 10000000 00011010 00000000 00000010
                Node<K,V>[] nt;
                // 依然有BUG
                // 当前线程扩容时,老数组长度是否和我当前线程扩容时的老数组长度一致
                // 00000000 00000000 10000000 00011010
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs  
                    // 10000000 00011010 00000000 00000010 
                    // 00000000 00000000 10000000 00011010
                    // 这两个判断都是有问题的,核心问题就应该先将rs左移16位,再追加当前值。
                    // 这两个判断是BUG
                    // 判断当前扩容是否已经即将结束
                    || sc == rs + 1   // sc == rs << 16 + 1 BUG
                    // 判断当前扩容的线程是否达到了最大限度
                    || sc == rs + MAX_RESIZERS   // sc == rs << 16 + MAX_RESIZERS BUG
                    // 扩容已经结束了。
                    || (nt = nextTable) == null 
                    // 记录迁移的索引位置,从高位往低位迁移,也代表扩容即将结束。
                    || transferIndex <= 0)
                    break;
                // 如果线程需要协助扩容,首先就是对sizeCtl进行+1操作,代表当前要进来一个线程协助扩容
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    // 上面的判断没进去的话,nt就代表新数组
                    transfer(tab, nt);
            }
            // 是第一个来扩容的线程
            // 基于CAS将sizeCtl修改为:10000000 00011010 00000000 00000010 
            // 将扩容戳左移16位之后,符号位是1,就代码这个值为负数
            // 低16位在表示当前正在扩容的线程有多少个:低16位为2,代表当前有一个线程正在扩容
            // 为什么低16位值为2时,代表有一个线程正在扩容
            // 每一个线程扩容完毕后,会对低16位进行-1操作,当最后一个线程扩容完毕后,减1的结果还是-1,
            // 当值为-1时,要对老数组进行一波扫描,查看是否有遗漏的数据没有迁移到新数组
            else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))
                // 调用transfer方法,并且将第二个参数设置为null,就代表是第一次来扩容!
                transfer(tab, null);
        }
    }
}

5、transfer方法-计算每个线程迁移的长度

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    // 数组长度 
    int n = tab.length; 
    // 每个线程一次性迁移多少数据到新数组
    int stride;     
    // 基于CPU的线程数来计算:NCPU = Runtime.getRuntime().availableProcessors()
    // MIN_TRANSFER_STRIDE = 16
    // NCPU = 4
    // 举个例子:N = 1024 - 512 - 256 - 128 / 4 = 32
    // 如果算出来每个线程的长度小于16的话,直接使用16
    // 如果大于16的话,则使用N
    // 如果线程数只有1的话,直接就是原数组长度
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE){
        stride = MIN_TRANSFER_STRIDE;
    }
}

6、 transfer方法-构建新数组并查看标识属性

// 以32位长度数组扩容到64位为例子
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    // 数组长度 
    int n = tab.length; 
    // 每个线程一次性迁移多少数据到新数组
    int stride;   
    // 第一个进来扩容的线程需要把新数组构建出来
    if (nextTab == null) {            
        try {
            // 将原数组长度左移一位创建新数组
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            // 赋值操作
            nextTab = nt;
        } catch (Throwable ex) {      
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        // 将成员变量的新数组赋值volatile修饰
        nextTable = nextTab;
        // 迁移数据时用到的标识volatile修饰
        transferIndex = n;
    }
    // 新数组长度
    int nextn = nextTab.length;
    // 老数组迁移完数据后,做的标识
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // 迁移数据时,需要用到的标识
    boolean advance = true;
    boolean finishing = false; 
}

7、transfer方法-线程领取迁移任务

// 以32位长度数组扩容到64位为例子
// 数组长度 
int n = tab.length; 
// 每个线程一次性迁移多少数据到新数组
int stride;
// 迁移数据时用到的标识volatile修饰
transferIndex = n;
// 新数组长度
int nextn = nextTab.length;
// 老数组迁移完数据后,做的标识
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// advance:true,代表当前线程需要接收任务,然后再执行迁移
// 如果为false,代表已经接收完任务
boolean advance = true;
// 标识迁移是否结束
boolean finishing = false;
// 开始循环
for (int i = 0, bound = 0;;) {
    Node<K,V> f; 
    int fh;
    while (advance) {
        int nextIndex;
        int nextBound;
        // 第一次:这里肯定进不去
        // 主要判断当前任务是否执行完毕
        if (--i >= bound || finishing)
            advance = false;
        // 第一次:这里肯定也进不去
        // 判断transferIndex是否小于等于0,代表没有任务可领取,结束了。
        // 在线程领取任务会,会对transferIndex进行修改,修改为transferIndex - stride
        // 在任务都领取完之后,transferIndex肯定是小于等于0的,代表没有迁移数据的任务可以领取
        else if ((nextIndex = transferIndex) <= 0) {
            i = -1;
            advance = false;
        }
        // nextIndex=32
        // stride=16
        // nextIndex > stride ? nextIndex - stride : 0:当前的nextIndex是否大于每个线程切割的
        // 是:nextIndex - stride/否:0
        // 将TRANSFERINDEX从nextIndex变成16
        else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,
                                     nextBound = (nextIndex > stride ?nextIndex - stride : 0))) {
            // 对bound赋值(16)
            bound = nextBound;
            // 对i赋值(31)
            i = nextIndex - 1;
            // 赋值,代表当前领取任务结束
            // 该线程当前领取的任务是(16~31)
            advance = false;
        }
    }
}

线程领取扩容任务的流程:

8、transfer方法-扩容是否已经结束

// 判断扩容是否已经结束!
// i < 0:当前线程没有接收到任务!
// i >= n: 迁移的索引位置,不可能大于数组的长度,不会成立
// i + n >= nextn:因为i最大值就是数组索引的最大值,不会成立
if (i < 0 || i >= n || i + n >= nextn) {
    int sc;
    // 第一次到这里:必定是false
    // 如果再次到这里的时候,最后一个扩容的线程也扫描完了
    if (finishing) {
        nextTable = null;
        table = nextTab;
        // 重新更改当前的sizeCtl的数值:0.75N
        sizeCtl = (n << 1) - (n >>> 1);
        return;
    }
    // 到这里代表当前线程已经不需要去扩容了
    // 那么它要把当前并发扩容的线程数减一
    // SIZECTL ----> sc - 1
    if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
        // 判断当前这个线程是不是最后一个扩容线程
        if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
            // 如果不是的话,直接返回就好了
            return;
        // 如果是的话,当前线程还需要将旧数据全部扫描一遍,校验一下是不是都迁移完了
        finishing = advance = true;
        // 将当前的i重新变成原数组的长度
        // 重新进行一轮校验
        i = n; // recheck before commit 
    }
}

9、 transfer方法-迁移数据(链表)

// 如果当前的i位置上没有数据,直接不需要迁移
  else if ((f = tabAt(tab, i)) == null)
        // 将原来i位置上CAS标记成fwd
        // fwd的Hash为Move
        advance = casTabAt(tab, i, null, fwd);
  // // 拿到当前i位置的hash值,如果为MOVED,证明数据已经迁移过了。
    else if ((fh = f.hash) == MOVED)
        // 那么直接返回即可,主要是最后一个扩容线程扫描的时候使用
        advance = true; // already processed
    else {
        // 锁住
        synchronized (f) {
            // 再次校验
            if (tabAt(tab, i) == f) {
                // ln:null  - lowNode
                Node<K,V> ln;
                 // hn:null  - highNode
                Node<K,V> hn;
                // 如果当前的fh是正常的
                if (fh >= 0) {
                    // 求当前的runBit
                    // 这里只会有两个结果:0 或者 n
                    int runBit = fh & n;
                    // 用当前lastRun指向f
                    Node<K,V> lastRun = f;
                    // 进行链表的遍历
                    for (Node<K,V> p = f.next; p != null; p = p.next) {
                        // 求每一个链表的p.hash & n
                        int b = p.hash & n;
                        // 如果当前的b不等于上一个节点
                        // 需要更新下runBit的数据和lastRun的指针
                        if (b != runBit) {
                            runBit = b;
                            lastRun = p;
                        }
                    }
                    // 如果最后runBit=0的话
                    // lastRun赋值给ln
                    if (runBit == 0) {
                        ln = lastRun;
                        hn = null;
                    }
                    // 如果最后runBit=0的话
                    // lastRun赋值给hn
                    else {
                        hn = lastRun;
                        ln = null;
                    }
                    // 从数组头节点开始一直遍历到lastRun位置
                    for (Node<K,V> p = f; p != lastRun; p = p.next) {
                        // hash/key/value
                        int ph = p.hash; K pk = p.key; V pv = p.val;
                        // 如果当前的是0,则挂到ln下面
                        if ((ph & n) == 0)
                            ln = new Node<K,V>(ph, pk, pv, ln);
                        // 如果当前的是1,则挂到hn下面
                        else
                            hn = new Node<K,V>(ph, pk, pv, hn);
                    }
                    // 将新数组i位置设置成ln
                    setTabAt(nextTab, i, ln);
                    // 将新数组i+N位置设置成hn
                    setTabAt(nextTab, i + n, hn);
                    // 将原来数组的i位置设置成fwd,代表迁移完毕
                    setTabAt(tab, i, fwd);
                    // 将advance设置成true,保证进行上面的while循环
                    // 执行上面的i--,进行下一节点的迁移计划
                    advance = true;
                }
            }
        }
    }
}
9.1 LastRun机制

假如当前我需要将数组从 16 扩容至 32,我们看下原来的结构:

当我们进行下面这一步时:

int runBit = fh & n;
for (Node<K,V> p = f.next; p != null; p = p.next) {
    // 求每一个链表的p.hash & n
    int b = p.hash & n;
    // 如果当前的b不等于上一个节点
    // 需要更新下runBit的数据和lastRun的指针
    if (b != runBit) {
        runBit = b;
        lastRun = p;
    }
}

图片如下:紫色代表 0,粉色代表 1

在迁移之前,我们要掌握一个知识,由于我们的数组是 2 倍扩容,所以原始数据的数据一定会落在新数组的 2N + 2 的位置

最后将我们的数组从原始数组直接迁移过来,由于 lastRun 之后位置的数据都是一样的hash,所以直接全量迁移即可,不需要挨个遍历,这也是实施 lastRun 的原因,减少时间复杂度

最终迁移效果如上,迁移完毕,减少 i 的下标,继续迁移下一个数组的位置。

10、协助扩容

// 如果当前的hash是正在扩容
if ((fh = f.hash) == MOVED){
     // 进行协作式扩容
    tab = helpTransfer(tab, f);
}
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; 
    int sc;
    // 第一个判断:老数组不为null
    // 第二个判断:新数组不为null  (将新数组赋值给nextTab)
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
        // 拿到扩容戳
        int rs = resizeStamp(tab.length);
        // 第一个判断:fwd中的新数组,和当前正在扩容的新数组是否相等。    
        // 相等:可以协助扩容。不相等:要么扩容结束,要么开启了新的扩容
        // 第二个判断:老数组是否改变了。     相等:可以协助扩容。不相等:扩容结束了
        // 第三个判断:如果正在扩容,sizeCtl肯定为负数,并且给sc赋值
        while (nextTab == nextTable && table == tab &&
               (sc = sizeCtl) < 0) {
            // 判断扩容戳是否一致
            // 看下transferIndex是否小于0,如果小于的话,说明扩容的任务已经被领完了
            if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                sc == rs + MAX_RESIZERS || transferIndex <= 0)
                break;
            // 如果还有任务的话,CAS将当前的SIZECTL加一,协助扩容
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                // 协助扩容的代码,上面刚刚分析过
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}

三、流程图

四、总结

鲁迅先生曾说:独行难,众行易,和志同道合的人一起进步。彼此毫无保留的分享经验,才是对抗互联网寒冬的最佳选择。

其实很多时候,并不是我们不够努力,很可能就是自己努力的方向不对,如果有一个人能稍微指点你一下,你真的可能会少走几年弯路。


相关文章
|
4天前
|
设计模式 安全 Java
Java面试题:设计模式如单例模式、工厂模式、观察者模式等在多线程环境下线程安全问题,Java内存模型定义了线程如何与内存交互,包括原子性、可见性、有序性,并发框架提供了更高层次的并发任务处理能力
Java面试题:设计模式如单例模式、工厂模式、观察者模式等在多线程环境下线程安全问题,Java内存模型定义了线程如何与内存交互,包括原子性、可见性、有序性,并发框架提供了更高层次的并发任务处理能力
15 1
|
4天前
|
安全 算法 Java
Java面试题:如何使用并发集合,例如ConcurrentHashMap?
Java面试题:如何使用并发集合,例如ConcurrentHashMap?
14 1
|
4天前
|
设计模式 安全 Java
Java面试题:解释单例模式的实现方式及其优缺点,讨论线程安全性的实现。
Java面试题:解释单例模式的实现方式及其优缺点,讨论线程安全性的实现。
10 0
|
4天前
|
设计模式 存储 缓存
Java面试题:结合设计模式与并发工具包实现高效缓存;多线程与内存管理优化实践;并发框架与设计模式在复杂系统中的应用
Java面试题:结合设计模式与并发工具包实现高效缓存;多线程与内存管理优化实践;并发框架与设计模式在复杂系统中的应用
7 0
|
4天前
|
设计模式 安全 NoSQL
Java面试题:设计一个线程安全的单例模式,并解释其内存占用和垃圾回收机制;使用生产者消费者模式实现一个并发安全的队列;设计一个支持高并发的分布式锁
Java面试题:设计一个线程安全的单例模式,并解释其内存占用和垃圾回收机制;使用生产者消费者模式实现一个并发安全的队列;设计一个支持高并发的分布式锁
8 0
|
4天前
|
设计模式 安全 Java
Java面试题:如何实现一个线程安全的单例模式,并确保其在高并发环境下的内存管理效率?如何使用CyclicBarrier来实现一个多阶段的数据处理任务,确保所有阶段的数据一致性?
Java面试题:如何实现一个线程安全的单例模式,并确保其在高并发环境下的内存管理效率?如何使用CyclicBarrier来实现一个多阶段的数据处理任务,确保所有阶段的数据一致性?
8 0
|
4天前
|
设计模式 缓存 安全
Java面试题:设计模式在并发编程中的创新应用,Java内存管理与多线程工具类的综合应用,Java并发工具包与并发框架的创新应用
Java面试题:设计模式在并发编程中的创新应用,Java内存管理与多线程工具类的综合应用,Java并发工具包与并发框架的创新应用
7 0
|
4天前
|
设计模式 存储 缓存
Java面试题:结合建造者模式与内存优化,设计一个可扩展的高性能对象创建框架?利用多线程工具类与并发框架,实现一个高并发的分布式任务调度系统?设计一个高性能的实时事件通知系统
Java面试题:结合建造者模式与内存优化,设计一个可扩展的高性能对象创建框架?利用多线程工具类与并发框架,实现一个高并发的分布式任务调度系统?设计一个高性能的实时事件通知系统
9 0
|
4天前
|
设计模式 SQL 安全
Java面试题:设计一个线程安全的内存管理器,使用观察者模式来通知所有线程内存使用情况的变化。如何确保在添加和移除内存块时的线程安全?如何确保任务的顺序执行和调度器的线程安全?
Java面试题:设计一个线程安全的内存管理器,使用观察者模式来通知所有线程内存使用情况的变化。如何确保在添加和移除内存块时的线程安全?如何确保任务的顺序执行和调度器的线程安全?
11 0
|
4天前
|
设计模式 并行计算 安全
Java面试题: 如何使用装饰器模式来增强ConcurrentHashMap的功能?在什么情况下应该使用CopyOnWriteArrayList而不是ArrayList?
Java面试题: 如何使用装饰器模式来增强ConcurrentHashMap的功能?在什么情况下应该使用CopyOnWriteArrayList而不是ArrayList?
9 0