Striped64类的实现

介绍:

这个类是个抽象类，里面主要实现了两个方法longAccumulate，doubleAccumulate 分别long 和double的数据进行一个累计。主要提供给LongAdder ，LongAccumulator,DoubleAdder,DoubleAccumulator 进行一个调用。可以理解为用cells的方式来减少并发时产生的冲突。

详解：

    final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
        int h;
        //获取probe 值，如果为0，进行初始化处理
        if ((h = getProbe()) == 0) {
            ThreadLocalRandom.current(); // force initialization
            h = getProbe();
            //标志未竞争为true
            wasUncontended = true;
        }
        //设置没有cell 碰撞
        boolean collide = false;                // True if last slot nonempty
        for (;;) {
            Striped64.Cell[] as; Striped64.Cell a; int n; long v;
            //cells中有值
            if ((as = cells) != null && (n = as.length) > 0) {
                //对应这个hash 值对应的cell中没有值
                if ((a = as[(n - 1) & h]) == null) {
                    //创建cell的时候的锁
                    if (cellsBusy == 0) {       // Try to attach new Cell
                        Striped64.Cell r = new Striped64.Cell(x);   // Optimistically create
                        //加锁
                        if (cellsBusy == 0 && casCellsBusy()) {
                            boolean created = false;
                            try {               // Recheck under lock
                                Striped64.Cell[] rs; int m, j;
                                if ((rs = cells) != null &&
                                        (m = rs.length) > 0 &&
                                        rs[j = (m - 1) & h] == null) {
                                    rs[j] = r;
                                    created = true;
                                }
                            } finally {
                                cellsBusy = 0;
                            }
                            if (created)
                                break;
                            //走到continue 表示这cell 已经创建了，重试吧，这种情况一般不会出现
                            continue;           // Slot is now non-empty
                        }
                    }
                    collide = false;
                }
                //发生了cas竞争的失败
                else if (!wasUncontended)       // CAS already known to fail
                //表示此cell肯定有值了，就是a，设置为true ，最后会重新计算 probe值，这个值的false，只会在当前线程probe 已经存在的情况下出现
                    wasUncontended = true;      // Continue after rehash
                // 知晓cell 就是a 了，对a 这个cell 里面的值进行一个操作
                else if (a.cas(v = a.value, ((fn == null) ? v + x :
                        fn.applyAsLong(v, x))))
                    break;
                //如果table长度已经是最大值，或as 已经不是原来的了,即竞争超时
                else if (n >= NCPU || cells != as)
                    //设置碰撞为false，之后会重新计算probe值
                    collide = false;            // At max size or stale
                else if (!collide)
                    collide = true; //重新计算probe值时候还是碰撞，并且对cell更新还是一个失败
                else if (cellsBusy == 0 && casCellsBusy()) { //竞争太多了，开始扩容了
                    try {
                        if (cells == as) {      // Expand table unless stale
                            Striped64.Cell[] rs = new Striped64.Cell[n << 1];
                            for (int i = 0; i < n; ++i)
                                rs[i] = as[i];
                            cells = rs;
                        }
                    } finally {
                        cellsBusy = 0;
                    }
                    collide = false;
                    //扩容后的重试
                    continue;                   // Retry with expanded table
                }
                h = advanceProbe(h);//重新计算probe 值
            }//表示为第一次遇到竞争，此时基本上是刚刚开始创建cells数组
            else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
                boolean init = false;
                try {        //初始化cells并写入第一个值                  // Initialize table
                    if (cells == as) {
                        Striped64.Cell[] rs = new Striped64.Cell[2];
                        //初始化cell，并将值写入cell中
                        rs[h & 1] = new Striped64.Cell(x);
                        cells = rs;
                        init = true;
                    }
                } finally {
                    cellsBusy = 0;
                }
                if (init)
                    break;
            }// cellsBusy锁竞争失败 试试能不能修改base的值
            else if (casBase(v = base, ((fn == null) ? v + x :
                    fn.applyAsLong(v, x))))
                break;                          // Fall back on using base
        }
    }

cells的长度为2的幂次方
配合上LongAdder 类中的add方法来看,这个是实现是尽可能的减少并发时产生的冲突,实在不行了才进行下一个冲突。

1. 更新base值，不行
   2.如果没有初始化cells，初始化cells更新
   3.如果对于hash的cell上没有值，写入，写入不成功，重新hash
   4.如果之前在cell上有竞争的，重新hash
   5.在对应cell值上对cas 进行更新。
   6.如果cells的长度还没有到最大cpu数，设置碰撞为false，重新hash
2. 如果没有碰撞，设置为有碰撞，重新hash
   8.如果上面都没有命中，那就开始扩容一倍，从第三步开始

   
    public void add(long x) {
        Cell[] as; long b, v; int m; Cell a;
// 如果cell为空，尝试对base值修改
        if ((as = cells) != null || !casBase(b = base, b + x)) {
            boolean uncontended = true;
//  如果cell上对应的值不为空，先试试更新cell中的值        
      if (as == null || (m = as.length - 1) < 0 ||
                (a = as[getProbe() & m]) == null ||
                !(uncontended = a.cas(v = a.value, v + x))
       //不行的话 ，进行上不striped64 中的方式进行
                longAccumulate(x, null, uncontended);
        }
    }