LongAdder及AtomicLong

秒速五厘米 2024-02-19 21:11 42阅读 0赞

**AtomicLong原理**  
  就像我们所知道的那样,AtomicLong的原理是依靠底层的cas来保障原子性的更新数据，在要添加或者减少的时候，会使用死循环不断地cas到特定的值，从而达到更新数据的目的。那么LongAdder又是使用到了什么原理?难道有比cas更加快速的方式？  
**LongAdder原理**  
这里让我困惑的一个问题是LongAdder中没有类似于AtomicLong中getAndIncrement()或者incrementAndGet()这样的原子操作，所以只能通过increment()方法和longValue()方法的组合来实现更新和获取的动作，然而这样不能保证这个组合操作的原子性，猜想也许LongAdder就是不具备这样的机制吧。那么就主要看一下increment()和longValue()方法。  
**increment方法：**  
分析increment()方法，可以看到该方法就是对add(long x)的封装，那么具体来分析一下这个方法

public void add(long x) {
            Cell[] as; long b, v; int m; Cell a;
            if ((as = cells) != null || !casBase(b = base, b + x)) {
                boolean uncontended = true;
                if (as == null || (m = as.length - 1) < 0 ||
                    (a = as[getProbe() & m]) == null ||
                    !(uncontended = a.cas(v = a.value, v + x)))
                    longAccumulate(x, null, uncontended);
            }
        }

这里重点是casBase(b = base, b + x)，来看一下它做了什么

final boolean casBase(long cmp, long val) {
            return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
        }

这里的四个条件其实并不是并列的，而是递进式的，1和2判断cells数组是否为空，3取cells数组中的任意一个元素a判断是否为空，4是对a进行cas操作并将执行结果赋值标志位uncontended。从这里可以给出第二个结论，当竞争激烈到一定程度无法对base进行累加操作时，会对cells数组中某个元素进行更新。  
最后来看一下当上述条件无法全部满足时调用的longAccumulate(x, null, uncontended)方法

final void longAccumulate(long x, LongBinaryOperator fn,
                                  boolean wasUncontended) {
            int h;
            if ((h = getProbe()) == 0) {
                ThreadLocalRandom.current(); // force initialization
                h = getProbe();  //返回当前线程的threadLocalRandomProbe值
                wasUncontended = true;
            }
            boolean collide = false;                // True if last slot nonempty
            for (;;) {
                Cell[] as; Cell a; int n; long v;
                if ((as = cells) != null && (n = as.length) > 0) {
                    if ((a = as[(n - 1) & h]) == null) {
                        if (cellsBusy == 0) {       // cells数组中对应位置没有数据则插入新对象
                            Cell r = new Cell(x);   
                            if (cellsBusy == 0 && casCellsBusy()) {
                                boolean created = false;
                                try {               // Recheck under lock
                                    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;           // Slot is now non-empty
                            }
                        }
                        collide = false;
                    }
                    else if (!wasUncontended)       // CAS already known to fail
                        wasUncontended = true;      // Continue after rehash
                    else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                                 fn.applyAsLong(v, x))))         //对该位置的cell元素进行累加
                        break;
                    else if (n >= NCPU || cells != as)
                        collide = false;            // At max size or stale   //判断数组大小是否大于核数
                    else if (!collide)
                        collide = true;
                    else if (cellsBusy == 0 && casCellsBusy()) {         //对cells数组进行扩容，直接扩容为2倍
                        try {
                            if (cells == as) {      // Expand table unless stale
                                Cell[] rs = new 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);
                }
                else if (cellsBusy == 0 && cells == as && casCellsBusy()) {  //cellsBusy这里是做为一个自旋锁来使用的
                    boolean init = false;
                    try {                           // 初始化cells数组大小为2
                        if (cells == as) {
                            Cell[] rs = new Cell[2];
                            rs[h & 1] = new Cell(x);
                            cells = rs;
                            init = true;
                        }
                    } finally {
                        cellsBusy = 0;
                    }
                    if (init)
                        break;
                }
                else if (casBase(v = base, ((fn == null) ? v + x :
                                            fn.applyAsLong(v, x))))      //对base进行CAS操作
                    break;                          // Fall back on using base
            }
        }

这个方法比较长，大致对几个关键点做了注释，该方法主要是用一个死循环对cells数组中的元素进行操作，当要更新的位置的元素为空时插入新的cell元素，否则在该位置进行CAS的累加操作，如果CAS操作失败并且数组大小没有超过核数就扩容cells数组。

**总结**  
LongAdder类与AtomicLong类的区别在于高并发时前者将对单一变量的CAS操作分散为对数组cells中多个元素的CAS操作，取值时进行求和；而在并发较低时仅对base变量进行CAS操作，与AtomicLong类原理相同。不得不说这种分布式的设计还是很巧妙的。