当前位置 博文首页 > =凌晨=:ConcurrentHashMap源码解读二

    =凌晨=:ConcurrentHashMap源码解读二

    作者:=凌晨= 时间:2021-05-11 18:20

    接下来就讲解put里面的三个方法,分别是

    1、数组初始化方法initTable()
    2、线程协助扩容方法helpTransfer()
    3、计数方法addCount()

    首先是数组初始化,再将源码之前,先得搞懂里面的一个重要参数,那就是sizeCtl。

    sizeCtl默认为0,代表数组未初始化。

    sizeCtl为正数,如果数组未初始化,那么其记录的是数组的初始容量,如果数组已经初始化,那么其记录的是数组的扩容阈值。

    sizeCtl为-1,表示数组正在进行初始化。

    sizeCtl小于0,并且不是-1,表示数组正在扩容,-(1+n),表示此时有n个线程共同完成数组的扩容操作。

    接下来讲解initTable()方法

    private final Node<K,V>[] initTable() {
            Node<K,V>[] tab; int sc;
        //第一次put的时候,table还没被初始化,所以能进入while。sizeCtl默认值是0,当有两个线程都想进行初始化时,线程A CAS成功,也就是else if为true,继续执行下面,而另一个线程cas就是false,就重新进行while循环,而这时sizeCtl为-1.所以这个线程就
        放弃cpu的控制权,说白了就是在多线程下保证初始化只执行一次。
    while ((tab = table) == null || tab.length == 0) {//tab在这里赋值,是table的引用 if ((sc = sizeCtl) < 0)//sc在这里赋值。是sizeCtl的引用。 Thread.yield(); // lost initialization race; just spin//线程放弃cpu的控制权。
            //SIZECTL:表示当前对象的内存偏移量,sc表示期望值,-1表示要替换的值,设定为-1表示要初始化表了
    else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//SIZECTL是地址偏移量,如果SIZECTL对应地址的值与sc相等,说明当前的线程是第一个到达这条语句的线程,那么就会将SIZECTL地址所对应的值替换成-1,而SIZECTL地址偏移量对应的对象就是sizeCtl try { if ((tab = table) == null || tab.length == 0) {//再次进行判断,防止在进行U.compareAndSwapInt(this, SIZECTL, sc, -1)的时候有其他线程并发进入方法,导致出错,使用双重锁 int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//sc在前边就赋值了,如果有初值,那么这里n就是设定的初始值,否则n就是默认容量。16 @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//基于初始长度,构建数组对象 table = tab = nt; sc = n - (n >>> 2);//这里就是计算扩容阈值,并赋值给sc。也就是n-0.25n = 0.75n } } finally { sizeCtl = sc;//将扩容阈值,赋值给sizeCtl,第一次初始化后。 } break; } } return tab; }

    所以在这个方法中,sizeCtl为正数,如果数组未初始化,那么其记录的是数组的初始容量,如果数组已经初始化,那么其记录的是数组的扩容阈值。就是这个含义。

    第二个要讲解的方法是addCount方法。

     这段代码分为两个部分,一个是计数部分,另一个是扩容部分。

    <用两种方法进行计数,一个是用cas对baseCount<baseCount是一个全局属性,volatile的>进行加法计数,另一个是用CounterCell数组,其实CounterCell对象就有一个属性是value,用CounterCell构建一个数组,然后哪个线程要进行加法,就用这个线程产生一个hash值,并用这个数与CounterCell数组的长度减一做与运算,得到的结果就是CounterCell数组的index,然后对这个index对应的CounterCell对象的value做加法。在最后统计计数的时候是用:baseCount+每个CounterCell的value>
    countcell就是通过分散计算来减少竞争。其内部有一个基础值和一个哈希表。当没有竞争的时候在基础值上计数。有竞争的时候通过哈希表计数(每个线程有一个哈希值,通过哈希值确定在哈希表的位置,在这个位置进行计数,不同位置互不影响)

    计数部分:通过baseCount和CounterCell数组,二选一的参与计数
    扩容部分:当大于扩容阈值的时候进行扩容,当满足扩容条件时才能扩容
    如果有别的线程正在进行扩容,那么就存在nextTable(一个全局属性,表示正在扩容的线程),就把nextTable作为参数传入transfer方法,就是让这个线程帮忙扩容nextTable
    如果没有别的线程正在扩容,那么就把null传入transfer方法,让transfer方法创建一个nextTable

    private final void addCount(long x, int check) {
            CounterCell[] as; long b, s;//as表示counterCells引用, b表示获取的baseCount值, s应该是表示元素个数。
     //当CounterCell数组不为空,则优先利用数组中的CounterCell记录数量
        //或者当baseCount的累加操作失败,会利用数组中的CounterCell记录数量
        //条件一:as!=null true:表示counterCells以及初始化过了,当前线程应该将数据写入到对应的counterCell中。
                //as==null 也就是条件1为false,那么表示counterCells未初始化,当前所有线程应该将数据写到baseCount中。
    if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {//条件二:true:表示当前线程cas替换数据成功  false表示发生竞争了,可能需要重试或者扩容。//因为这里有个!,所以应该false才能进入
        //什么时候会进入到if判断里面
             //1.条件一:as!=null true:表示counterCells以及初始化过了,当前线程应该将数据写入到对应的counterCell中。
            //2.条件二:false表示发生竞争了,可能需要重试或者扩容 CounterCell a;
    long v; int m;//a表示当前线程命中的CounterCell单元格 v表示期望值,m表示as数组的长度。 boolean uncontended = true; //标识是否有多线程竞争 ,//true表示未发生竞争,false表示发生竞争。
          //当as数组为空
            //或者当as长度为0
            //或者当前线程对应的as数组桶位的元素为空
            //或者当前线程对应的as数组桶位不为空,但是累加失败
            //条件一:as==null 为true,说明counterCells未初始化,那么上面就是根据多线程写base发生竞争进入到这里的。
                    //as!=null为false,说明counterCells已经初始化了。当前线程应该是找自己的counterCells写值。
    if (as == null || (m = as.length - 1) < 0 ||
            //ThreadLocalRandom.getProbe()表示当前线程的hash值。 m是CountCell长度-1,as.length一定是2的次方数,比如长度为16,那么减一就是15
                //也就是b1111,那么与上之后肯定是小于等于当前长度的值,也就是下标。
                //如果a==null,也就是条件为true,说明当前线程对应下标的CountCell为空,那么就需要创建
                //如果是false,不为空,说明下一步想要将x的值添加到CountCell中。 (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
             //条件三:true:表示cas失败,意味着当前线程对应的CountCell有竞争,
                        //false,表示cas成功 !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
           //都有哪些情况会调用下面的方法:
                //1.as==null 为true,说明counterCells未初始化,那么上面就是根据多线程写base发生竞争进入到这里的,那么初始化。
                //2.如果a==null,也就是条件为true,说明当前线程对应下标的CountCell为空,那么就需要创建
                //3.true:表示cas失败,意味着当前线程对应的CountCell有竞争,那么就可能是重试或者扩容。
                //以上任何一种情况成立,都会进入该方法,传入的uncontended是false fullAddCount(x, uncontended);//也就是countCells是一个全局得volatile的CountCell数组,创建实在fullAddCount方法中。
    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) { //sc小于0,说明有线程正在扩容,那么会协助扩容
          //扩容结束或者扩容线程数达到最大值或者扩容后的数组为null或者没有更多的桶位需要转移,结束操作
    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break;
                //扩容线程加1,成功后,进行协助扩容操作
    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt);//协助扩容,newTable不为null }
              //否则sc>=0,说明是首个扩容的线程,所以transfer传入的参数是null
    else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount(); } } }

    接下来就是讲解上面的fullAddCount方法

    ① 当CounterCell数组不为空,优先对CounterCell数组中的CounterCell的value累加

    ② 当CounterCell数组为空,会去创建CounterCell数组,默认长度为2,并对数组中的CounterCell的value累加

    ③ 当数组为空,并且此时有别的线程正在创建数组,那么尝试对baseCount做累加,成功即返回,否则自旋

    private final void fullAddCount(long x, boolean wasUncontended) {//wasUncontended只有counterCells初始化之后,并且当前线程竞争修改失败,才会是false。其余情况都是true。
            int h;//h表示线程的hash值。
        //获取当前线程的hash值
    if ((h = ThreadLocalRandom.getProbe()) == 0) {//条件成立:说明当前线程还未分配hash值。 ThreadLocalRandom.localInit();//给当前线程分配hash值 // force initialization h = ThreadLocalRandom.getProbe();//取出当前线程的hash值,赋值给h wasUncontended = true;//为啥这里强制设为true,当前线程肯定是写入到了countCells[0]位置,不把它当作一次真正的竞争。 } boolean collide = false; //标识是否有冲突,如果最后一个桶不是null,那么为true //表示扩容意向 false一定不会扩容,true可能会扩容  // True if last slot nonempty for (;;) {//自旋 CounterCell[] as; CounterCell a; int n; long v;//as表示counterCells引用,a表示当前线程命中的CounterCell,n表示counterCells数组长度,v表示期望值
           //数组不为空,优先对数组中CouterCell的value累加
            //CASE1:表示countCells已经初始化了,当前线程应该将数据写入到对应的CounterCell中。
    if ((as = counterCells) != null && (n = as.length) > 0) {
          //2.如果a==null,也就是条件为true,说明当前线程对应下标的CountCell为空,那么就需要创建
                //3.true:表示cas失败,意味着当前线程对应的CountCell有竞争,那么就可能是重试或者扩容。这两种情况会进入到这个if判断中
                //线程对应的桶位为null
             //CASE1.1:a = as[(n - 1) & h]) == null true表示当前线程对应的下标位置的CounterCell为null,需要创建new CounterCell
    if ((a = as[(n - 1) & h]) == null) { if (cellsBusy == 0) { //true:表示当前锁未被占用,false表示锁被占用 // Try to attach new Cell CounterCell r = new CounterCell(x); // Optimistic create//创建CounterCell对象
                    //利用CAS修改cellBusy状态为1,成功则将刚才创建的CounterCell对象放入数组中
                        //条件一:true:表示当前锁未被占用,false表示锁被占用
                        //条件二:true:表示当前线程获取锁成功,false表示当前线程获取锁失败。
    if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean created = false;//是否创建成功的标记 try { // Recheck under lock CounterCell[] rs; int m, j;//rs表示当前countCells引用,m表示rs的长度,j表示当前线程命中的下标。
                      //桶位为空, 将CounterCell对象放入数组
                                //条件一条件二恒成立
                                // rs[j = (m - 1) & h] == null为了防止其它线程初始化过该位置,然后当前线程再次初始化该位置,导致数据丢失。
    if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) { rs[j] = r; created = true;//表示放入成功 } } finally { cellsBusy = 0; } if (created)//成功退出循环 break; continue; //桶位已经被别的线程放置了已给CounterCell对象,继续循环 // Slot is now non-empty } } collide = false;//将扩容意向改为false,原因是因为再CASE1.1中当前CounterCell都是为null,不可能不让写。因此不需要扩容。 }
              //桶位不为空,重新计算线程hash值,然后继续循环
                //CASE1.2:只有一种情况会来到这里,wasUncontended只有counterCells初始化之后,并且当前线程竞争修改失败,才会是false
    else if (!wasUncontended) // CAS already known to fail wasUncontended = true; // Continue after rehash
            //重新计算了hash值后,对应的桶位依然不为空,对value累加
                //成功则结束循环
                //失败则继续下面判断
                //CASE1.3:当前线程rehash过hash值,然后新命中的CounterCell不为空。则来到这里。
                //true:写成功,退出循环,
                //false:表示rehash之后命中的新的cell也有竞争,重试1次  再重试一次
    else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)) break;
             //数组被别的线程改变了,或者数组长度超过了可用cpu大小,重新计算线程hash值,否则继续下一个判断
                //CASE1.4:条件一:n >= NCPU true->表示扩容意向改为false,表示不扩容了。false说明counterCells数组还可以扩容
                //条件二:counterCells != as true表示其它线程已经扩容过了,当前线程rehash之后重试即可。
    else if (counterCells != as || n >= NCPU) collide = false; // At max size or stale
            //当没有冲突,修改为有冲突,并重新计算线程hash,继续循环
                //CASE1.5  !collide=true,设置扩容意向为true,但是不一定真的发生扩容。因为一旦进入这里,那么又会rehash一下,又会重来。
    else if (!collide) collide = true;
             //如果CounterCell的数组长度没有超过cpu核数,对数组进行两倍扩容
                //并继续循环
                //CASE1.6:真正扩容的逻辑
                //条件一:cellsBusy == 0 true表示当前无锁状态,当前线程可以去竞争这把锁。
                //条件二:U.compareAndSwapInt(this, CELLSBUSY, 0, 1)  true表示当前线程获取锁成功,可以执行扩容逻辑。
                                                                //false表示当前时刻有其它线程正在做扩容相关的操作。
    else if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { try { if (counterCells == as) {// Expand table unless stale //这里又有双重检测,就是为了防止扩容过了又再次扩容。 CounterCell[] rs = new CounterCell[n << 1];//扩容为两倍扩容 for (int i = 0; i < n; ++i) rs[i] = as[i];//将旧数组的值放到新数组。 counterCells = rs; } } finally { cellsBusy = 0; //释放锁 } collide = false; continue; // Retry with expanded table } h = ThreadLocalRandom.advanceProbe(h);//重置当前hash值 rehash }
         //CounterCells数组为空,并且没有线程在创建数组,修改标记,并创建数组,因为前面CASE1是数组不为空,所以这里是数组为空
            //CASE2:当前条件countCells还未初始化,as为null
            //条件一:cellsBusy == 0  true表示当前未加锁
            //条件二:counterCells == as,为什么又重新比较一遍,明明前面CASE1已经赋值未null了。因为其它线程可能会在你给as赋值之后修改了counterCells。
            //条件三:如果未true,表示获取锁成功,会把cellsBusy改成1。false表示其它线程正在持有这把锁,那么当前线程就进不了这里面了。
    else if (cellsBusy == 0 && counterCells == as && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean init = false; try { // Initialize table if (counterCells == as) {//为什么这里又判断了一下counterCells == as,防止其它线程已经初始化了,当前线程再次初始化,就会覆盖掉其它线程初始化的CountCell,导致丢失数据。 CounterCell[] rs = new CounterCell[2];//初始容量为2 rs[h & 1] = new CounterCell(x); counterCells = rs; init = true; } } finally { cellsBusy
    下一篇:没有了