大橙子网站建设,新征程启航

为企业提供网站建设、域名注册、服务器等服务

多线程(十六、ConcurrentHashMap原理(3)扩容)

扩容的原理

扩容一般分为2个步骤

1、table数组的扩容,一般是2倍的扩容,这个是单线程操作的。

2、数据的迁移,把旧table中的各个槽中的结点重新分配到新table中。

ConcurrentHashMap在处理rehash的时候,并不会重新计算每个key的hash值,而是利用了一种很巧妙的方法。

1、ConcurrentHashMap内部的table数组的大小必须为2的幂次,原因是让key均匀分布,减少冲突
2、当table数组的大小为2的幂次时,通过key.hash & table.length-1这种方式计算出的索引i,当table扩容后(2倍),新的索引要么在原来的位置i,要么是i+n,n为扩容之前的容量。
3、这种处理方式非常利于扩容时多个线程同时进行的数据迁移操作,因为旧table的各个桶中的结点迁移不会互相影响,将整个table数组划分为很多部分,每一部分包含一定区间的桶,每个数据迁移线程处理各自区间中的结点

什么时候扩容?

链表的结点数目超过一定阈值,就会触发链表 -> 红黑树的转换,执行treeifyBin方法。

/**
 *  链表 -> 红黑树 的转换.
 */
private final void treeifyBin(Node[] tab, int index) {
    Node b;
    int n, sc;
    if (tab != null) {

        // CASE 1: table的容量 < MIN_TREEIFY_CAPACITY(64)时,直接table扩容,不红黑树转换
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            tryPresize(n << 1); //左移一位,就是*2,扩容2倍

            // CASE 2: table的容量 ≥ MIN_TREEIFY_CAPACITY(64)时,链表 -> 红黑树的转换
        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
            synchronized (b) {
                if (tabAt(tab, index) == b) {
                    TreeNode hd = null, tl = null;

                    // 遍历链表,建立红黑树
                    for (Node e = b; e != null; e = e.next) {
                        TreeNode p = new TreeNode(e.hash, e.key, e.val, null, null);
                        if ((p.prev = tl) == null)
                            hd = p;
                        else
                            tl.next = p;
                        tl = p;
                    }
                    // 封装TreeBin,并链接到table[index]中
                    setTabAt(tab, index, new TreeBin(hd));
                }
            }
        }
    }
}

扩容函数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) { //大于等于0代表,初始化或者扩容后的(table*负载因子)的桶数量
            Node[] tab = table; int n;
            //Case1,table没有初始化,先初始化
            if (tab == null || (n = tab.length) == 0) {
                n = (sc > c) ? sc : c;
                //设置SIZECTL为-1代表,初始化或者扩容进行中
                if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                    try {
                        if (table == tab) {
                            Node[] nt = (Node[])new Node[n];//生成新的table
                            table = nt;
                            sc = n - (n >>> 2); //n-(n/4) = 0.75n,也就是负载因子
                        }
                    } finally {
                        sizeCtl = sc; //设置sizeCtl为扩容后的桶*负载因子
                    }
                }
            }
            //Case2, c=MAXIMUM_CAPACITY,说明桶超限了。
            else if (c <= sc || n >= MAXIMUM_CAPACITY)
                break;
            //Case3,开始扩容
            else if (tab == table) {
                int rs = resizeStamp(n);//根据容量n生成一个随机数,唯一标识本次扩容操作
                if (sc < 0) { //sc<0,说明sizeCtl<0,代表有别的线程正在进行扩容
                    Node[] nt;
                    //不能协助扩容,退出
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                            sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                            transferIndex <= 0)
                        break;
                    //加入扩容的队伍,同时sizeCtl+1,此时sizeCtl代表扩容的线程数量
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        transfer(tab, nt);
                }
                //设置sizeCtl为负数,代表自己是第一个扩容的线程,CAS操作保证只有一个线程安全
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                        (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null); //null代表首次扩容
            }
        }
    }

数据迁移方法transfer

/**
     * tab,    旧table
     * nextTAB 扩容后的table
     */
    private final void transfer(Node[] tab, Node[] nextTab) {
        int n = tab.length, stride;
        //每个线程负责迁移table一个区间段的桶的个数,最少是16个
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        //首次扩容
        if (nextTab == null) {            // initiating
            try {
                //创建新的table,默认n*2
                Node[] nt = (Node[])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;
        //创建扩容节点,当某个桶迁移完成后,放入table[i],标记桶扩容完成
        ForwardingNode fwd = new ForwardingNode(nextTab);
        boolean advance = true; //为true,表示当前桶迁移完成,可以继续处理下一个桶
        boolean finishing = false; // 最后一个数据迁移的线程将该值置为true,进行扩容的收尾工作
        //i桶索引,bound就是线程要处理的另一个区间边界
        for (int i = 0, bound = 0;;) {
            Node f; int fh;
            //定位本轮处理区间【transferIndex-1,transferIndex-stride】
            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))) {
                    bound = nextBound;
                    i = nextIndex - 1;
                    advance = false;
                }
            }
            //Case1,最后一个迁移线程或者是线程出现了冲突,导致了i<0
            if (i < 0 || i >= n || i + n >= nextn) {
                int sc;
                if (finishing) { //迁移完成
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                }
                //扩容线程数减1,当前线程任务已执行完成
                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                    //判断是否最后一个迁移线程,不是则退出
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    finishing = advance = true;
                    //最后一个线程,还原i值,重新进行检查,是否全部迁移完成,应该所有桶都是ForwardingNode
                    i = n; // recheck before commit
                }
            }
            // CASE2:旧桶本身为null,不用迁移,放一个ForwardingNode
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            //CASE3:该旧桶已经迁移完成,直接跳过,hash==moved 代表ForwardingNode
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
                // CASE4:该旧桶未迁移完成,进行数据迁移,加锁
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        Node ln, hn;
                        if (fh >= 0) { //桶是链表,迁移链表

                            /**
                             * 下面的过程会将旧桶中的链表分成两部分:ln链和hn链
                             * ln链会插入到新table的槽i中,hn链会插入到新table的槽i+n中
                             */
                            int runBit = fh & n;
                            Node lastRun = f;
                            for (Node 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 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(ph, pk, pv, ln);
                                else
                                    hn = new Node(ph, pk, pv, hn);
                            }
                            setTabAt(nextTab, i, ln);  // ln链表存入新桶的索引i位置
                            setTabAt(nextTab, i + n, hn); // hn链表存入新桶的索引i+n位置
                            setTabAt(tab, i, fwd); // 设置ForwardingNode占位
                            advance = true; // 表示当前旧桶的结点已迁移完毕
                        }
                        else if (f instanceof TreeBin) { //红黑树迁移
                            TreeBin t = (TreeBin)f;
                            TreeNode lo = null, loTail = null;
                            TreeNode hi = null, hiTail = null;
                            /**
                             * 先以链表方式遍历,复制所有结点,然后根据高低位组装成两个链表;
                             * 然后看下是否需要进行红黑树转换,最后放到新table对应的桶中
                             */
                            int lc = 0, hc = 0;
                            for (Node e = t.first; e != null; e = e.next) {
                                int h = e.hash;
                                TreeNode p = new TreeNode
                                        (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(lo) : t;
                            hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                                    (lc != 0) ? new TreeBin(hi) : t;
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);// 设置ForwardingNode占位
                            advance = true; // 表示当前旧桶的结点已迁移完毕
                        }
                    }
                }
            }
        }
    }

当前文章:多线程(十六、ConcurrentHashMap原理(3)扩容)
网站地址:http://dzwzjz.com/article/jgdheh.html
在线咨询
服务热线
服务热线:028-86922220
TOP