数据结构之-concurrentHashMap源码分析

太过爱你忘了你带给我的痛 2023-09-24 22:21 62阅读 0赞

1.  initTable（）的时候，通过sizeCtl属性值判断是否有其它线程在进行初始化，有的话就很灵性的出让cpu使用权，懂事。没有的话就自己干，这里使用了while循环，初始化完成后才放你出来。。

private final Node<K,V>[] initTable() {
            Node<K,V>[] tab; int sc;
            while ((tab = table) == null || tab.length == 0) {
                //<0，表示数组正在进行扩容
                if ((sc = sizeCtl) < 0)
                   //主动出让cpu使用权
                    Thread.yield(); 
                    //没有线程来初始化，就自己来，自己尝试来获取初始化数组任务
                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;
                            sc = n - (n >>> 2);
                        }
                    } finally {
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            return tab;
        }

1.  putVal（K，V），这里使用了CAS在为空节点时创建新节点，当已存在值的情况下，使用synchronized锁住当前节点（使用了最小的锁粒度）并进行是链表还是红黑树的判断，再进行值的修改，最后再来个分组不同位置计数

//onlyIfAbsent是否覆盖旧值
        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)
                    tab = helpTransfer(tab, f);
                    //当前节点上已有值
                else {
                    V oldVal = null;
                    synchronized (f) {
                    //在判断在节点位置有没有变化，被其它线程修改
                        if (tabAt(tab, i) == f) {
                             //表示是链表结构
                            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;
                                    }
                                }
                            }
                            //红黑树结构
                            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;
                                }
                            }
                        }
                    }
                    //查看是否需要升级为红黑树
                    if (binCount != 0) {
                        if (binCount >= TREEIFY_THRESHOLD)
                            treeifyBin(tab, i);
                        if (oldVal != null)
                            return oldVal;
                        break;
                    }
                }
            }
            修改的计数器+1
            addCount(1L, binCount);
            return null;
        }

1.  get(Object key)，这里面并没有加锁，因为**table**在定义是就是Volatile 保证了可见性，在查找节点的时候需要考虑多种情况：红黑树、链表、forwarding节点不同形式对应下的查找

public V get(Object key) {
            Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
            int h = spread(key.hashCode());
            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;
                }
                <0 红黑树结构或者forwarding节点，不同的节点方式查找
                else if (eh < 0)
                    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;
        }

1.  扩容（transfer(Node<K,V>\[\] tab, Node<K,V>\[\] nextTab)，首先就是分配复制任务给多线程，第一个线程进行新Node初始化，扩容为原来的2倍，线程扩容完成后就进行数据的复制迁移，这里有低位的ln跟高位的hn分别存储，最后进行汇总到新的Node中，将旧的位置就标记为fwd节点

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
            //单线程初始化，调用transfer扩容方法都是在CAS中
            if (nextTab == null) {            // initiating
                try {
                    @SuppressWarnings("unchecked")
                    //扩容为原来的2倍
                    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
            ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
            //判断扩容完成的属性值
            boolean advance = true;
            boolean finishing = false; // to ensure sweep before committing nextTab
            //bound表示当前线程可以处理的当前桶区间最小下标
            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))) {
                        //当前线程能处理的最小区间小标
                        bound = nextBound;
                        //当前线程能处理的最大区间下标
                        i = nextIndex - 1;
                        advance = false;
                    }
                }
                
                if (i < 0 || i >= n || i + n >= nextn) {
                    int sc;
                    if (finishing) {//扩容完成
                        nextTable = null;//删除成员表里
                        table = nextTab;//更新table
                        sizeCtl = (n << 1) - (n >>> 1);//更新阈值
                        return;
                    }
                    //sc-1，表示当前线程结束帮助扩容
                    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)
                    advance = casTabAt(tab, i, null, fwd);
                else if ((fh = f.hash) == MOVED)
                    advance = true; // already processed
                else {
                    //加锁防止扩容的时候put数据
                    synchronized (f) {
                        if (tabAt(tab, i) == f) {     
                            Node<K,V> ln, hn;//ln是低位节点，hn是高位节点
                            if (fh >= 0) { //链表结构
                                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);//低位就放在新链表的i位置
                                setTabAt(nextTab, i + n, hn);//高位就放在新链表的i+n位置
                                setTabAt(tab, i, fwd);//将旧的链表设置为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);
                                    if ((h & n) == 0) { //==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;
                                    }
                                }
                                //如果节点个数小于等于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);//旧的设置为fwd节点
                                advance = true;
                            }
                        }
                    }
                }
            }
        }

小结：当删除，增加元素时，碰到扩容过程是要进行协助扩容的。

多线程下的技术点：volatile,cas,synchronized

cas:用来创建新节点，更改一些属性值、状态值

synchronized：用来锁住数组的头节点

数据结构:数组,链表,红黑树

1.  为什么concurrentHashMap不能存null值

因为有歧义，concurrentHashMap是针对多线程的场景，假设能存入null值，那么在并发的情况下无法判断这个值是为空的null还是赋值为null

1.  扩容长度为8

泊松分布，长度为8已经是千万分之几的概率了，这是测试出来最合适的数值，正常情况下是达不到的，只是为了开发过程中的不当使用。当节点个数小于等于6时就退化为链表，也是防止链表跟红黑树的频繁转换。

1.  怎么解决死循环

使用尾插法，避免了在扩容时形成的死循环