天天看點

doubleAdder的性能為何比Atomic好

1.引言

       JDK8之前,我們對于簡單類型在高并發下的原子性,多數情況下,都會使用Atomic類型來控制,比如AtomicInteger、AtomicLong等,

       其原理是通過CAS(compare and swap)來進行原子性控制,該方法是調用Unsafe類控制的,是通過JNI調用相關的DLL進行CPU

       級别的操作。但是如果真是在高并發的情況下,針對單一變量的多次自旋,性能的消耗也是很嚴重的。那下面來介紹下面的;

2.DoubleAdder和LongAdder

       但是在JDK8裡有兩個類,DoubleAdder和LongAdder,這兩個類都繼承Striped64類,我們先來看Striped64這個類官方給的解釋:

       Apackage-local class holding common representation and mechanics for classessupporting dynamic striping on 64bit

       values.The class extends Number so that concrete subclasses must publicly do so.

       翻譯:一個包含通用表示和機制的包本地類對于支援64位值的動态條帶的類。類擴充數字,以便具體的子類必須公開這樣做。

       然後這個類下面有個内部類,Cell,源碼如下,

doubleAdder的性能為何比Atomic好

再看上面有個注解@sun.misc.Contended,這個就是對資料進行緩存填充,專業術語

       就是記憶體填充,知名架構Disruptor下也使用了類似的方式,但是當時JDK8還未出來,使用的是8個long,為什麼是64位,自行補充java記憶體模型;

       下面我們直接分析方法,下面是Striped64的方法,類似的還有個是long的,原理類似就不做分析了

   final void doubleAccumulate(double x, DoubleBinaryOperator fn,

                                booleanwasUncontended) {

       int h;

              //初始化指針位址

       if ((h = getProbe()) == 0) {

           ThreadLocalRandom.current(); // force initialization

           h = getProbe();

           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) {//判定存儲數組有值,講cells指派給as

                if ((a = as[(n - 1) & h])== null) {\\目前指針位置資料是否為空

                    if (cellsBusy == 0) {       // Try to attach new Cell

                        Cell r = newCell(Double.doubleToRawLongBits(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) ?

                               Double.doubleToRawLongBits

                               (Double.longBitsToDouble(v) + x) :

                               Double.doubleToRawLongBits

                               (fn.applyAsDouble

                                (Double.longBitsToDouble(v), x)))))

                   break;

                else if (n >= NCPU || cells!= as)

                    collide = false;            // At max size or stale

                else if (!collide)

                    collide = true;

                else if (cellsBusy == 0&& casCellsBusy()) {//指針達到最後,未鎖,搶占鎖

                    try {

                        if (cells == as) {      // 數組擴張一倍

                            Cell[] rs = newCell[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()//加鎖) {

                boolean init = false;

                try {                           // Initialize table

                    if (cells == as) {

                        Cell[] rs = newCell[2];

                                          //資料存入

                        rs[h & 1] = newCell(Double.doubleToRawLongBits(x));

                       cells = rs;

                        init = true;

                    }

                } finally {

                                   //解鎖

                    cellsBusy = 0;

                }

                if (init)

                    break;

           }

                     //高并發下初始情況下,cells還未來得及填充,通過和atomic一樣的方式詢問處理

           else if (casBase(v = base,

                             ((fn == null) ?

                             Double.doubleToRawLongBits

                             (Double.longBitsToDouble(v) + x) :

                              Double.doubleToRawLongBits

                              (fn.applyAsDouble

                              (Double.longBitsToDouble(v), x)))))

                break;                          // Fall back on usingbase

       }

}

3.小結

Strip64通過cell數組和記憶體填充(Contented),分減少多線程針對同一變量的資源并發競争和較少不同線程針對CPU針對記憶體緩存最小顆粒度的記憶體共享問題,這樣就相對于Atomic提高了性能

4.總結

DoubleAdder資料存在一個數組裡,我們看到,在DoubleAdder的sum方法裡,對于cell數組進行了循環周遊累加,doubleValue也是調用的此方法,下面是源碼

doubleAdder的性能為何比Atomic好

5.思維擴充

本身DoubleAdder的設計思想其實在諸多方面都可以應用,比如電商行業的庫存管理,在商品的下單過程中,如果商品很熱門,短時間内該商品下單量劇增,比如雙11,雙12,通過将高競争資源分布式管理,能顯著提升庫存扣減的性能