Java并發(fā)編程之LongAdder執(zhí)行情況解析

更新時(shí)間：2023年04月28日 14:01:47 作者：bright丶

這篇文章主要為大家介紹了Java并發(fā)編程之LongAdder執(zhí)行情況解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

正文

上篇文章 Java并發(fā)編程之LongAdder源碼（一）中最后寫(xiě)到了有三種情況會(huì)執(zhí)行longAccumulate方法，下面就根據(jù)這三種情況來(lái)進(jìn)行分析

當(dāng)Cell數(shù)組為null時(shí)，傳入的三個(gè)參數(shù)為1，null，true
隨機(jī)找到Cell數(shù)組某個(gè)索引位置的值為null時(shí)，傳入的三個(gè)參數(shù)為1，null，true
對(duì)Cell數(shù)組某個(gè)索引位置的值進(jìn)行累加失敗時(shí)，傳入的三個(gè)參數(shù)為1，null，false

longAccumulate方法

final void longAccumulate(long x, LongBinaryOperator fn,
                          boolean wasUncontended) {
    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) {
            if ((a = as[(n - 1) & h]) == null) {
                if (cellsBusy == 0) {       // Try to attach new Cell
                    Cell r = new Cell(x);   // Optimistically create
                    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))))
                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) {      // 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()) {
            boolean init = false;
            try {                           // Initialize table
                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))))
            break;                          // Fall back on using base
    }
}

代碼較長(zhǎng)，我們分段來(lái)分析，首先介紹一下各部分的內(nèi)容

第一部分：for循環(huán)之前的代碼，主要是獲取線程的hash值，如果是0的話就強(qiáng)制初始化
第二部分：for循環(huán)中第一個(gè)if語(yǔ)句，在Cell數(shù)組中進(jìn)行累加、擴(kuò)容
第三部分：for循環(huán)中第一個(gè)else if語(yǔ)句，這部分的作用是創(chuàng)建Cell數(shù)組并初始化
第四部分：for循環(huán)中第二個(gè)else if語(yǔ)句，當(dāng)Cell數(shù)組競(jìng)爭(zhēng)激烈時(shí)嘗試在base上進(jìn)行累加

線程hash值

int h; 
if ((h = getProbe()) == 0) { 
    ThreadLocalRandom.current(); // force initialization 
    h = getProbe(); 
    wasUncontended = true;   // true表示沒(méi)有競(jìng)爭(zhēng)
} 
boolean collide = false; // True if last slot nonempty 可以理解為是否需要擴(kuò)容

這部分的核心代碼是getProbe方法，這個(gè)方法的作用就是獲取線程的hash值，方便后面通過(guò)位運(yùn)算定位到Cell數(shù)組中某個(gè)位置，如果是0的話就會(huì)進(jìn)行強(qiáng)制初始化

初始化Cell數(shù)組

final void longAccumulate(long x, LongBinaryOperator fn,
                          boolean wasUncontended) {
    // 省略...
    for (;;) {
        Cell[] as; Cell a; int n; long v;
        if ((as = cells) != null && (n = as.length) > 0) {
            // 省略...
        }
        else if (cellsBusy == 0 && cells == as && casCellsBusy()) {  // 獲取鎖
            boolean init = false;  // 初始化標(biāo)志
            try {                           // Initialize table
                if (cells == as) {
                    Cell[] rs = new Cell[2];  // 創(chuàng)建Cell數(shù)組
                    rs[h & 1] = new Cell(x);  // 索引1位置創(chuàng)建Cell元素，值為x=1
                    cells = rs;   // cells指向新數(shù)組
                    init = true;  // 初始化完成
                }
            } finally {
                cellsBusy = 0;  // 釋放鎖
            }
            if (init)
                break;  // 跳出循環(huán)
        }
        else if (casBase(v = base, ((fn == null) ? v + x :
                                    fn.applyAsLong(v, x))))
            break;                          // Fall back on using base
    }
}

第一種情況下Cell數(shù)組為null，所以會(huì)進(jìn)入第一個(gè)else if語(yǔ)句，并且沒(méi)有其他線程進(jìn)行操作，所以cellsBusy==0，cells==as也是true，casCellsBusy()嘗試對(duì)cellsBusy進(jìn)行cas操作改成1也是true。

首先創(chuàng)建了一個(gè)有兩個(gè)元素的Cell數(shù)組，然后通過(guò)線程h & 1 的位運(yùn)算在索引1的位置設(shè)置一個(gè)value為1的Cell，然后重新賦值給cells，標(biāo)記初始化成功，修改cellsBusy為0表示釋放鎖，最后跳出循環(huán)，初始化操作就完成了。

對(duì)base進(jìn)行累加

final void longAccumulate(long x, LongBinaryOperator fn,
                          boolean wasUncontended) {
    // 省略...
    for (;;) {
        Cell[] as; Cell a; int n; long v;
        if ((as = cells) != null && (n = as.length) > 0) {
            // 省略...
        }
        else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
            // 省略...
        }
        else if (casBase(v = base, ((fn == null) ? v + x :
                                    fn.applyAsLong(v, x))))
            break;                          // Fall back on using base
    }
}

第二個(gè)else if語(yǔ)句的意思是當(dāng)Cell數(shù)組中所有位置競(jìng)爭(zhēng)都很激烈時(shí)，就嘗試在base上進(jìn)行累加，可以理解為最后的保障

Cell數(shù)組初始化之后

final void longAccumulate(long x, LongBinaryOperator fn,
                          boolean wasUncontended) {
    // 省略...
    for (;;) {
        Cell[] as; Cell a; int n; long v;
        if ((as = cells) != null && (n = as.length) > 0) {  // as初始化之后滿足條件
            if ((a = as[(n - 1) & h]) == null) {  // as中某個(gè)位置的值為null
                if (cellsBusy == 0) {       // Try to attach new Cell 是否加鎖
                    Cell r = new Cell(x);   // Optimistically create 創(chuàng)建新Cell
                    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) {  // 重新檢查該位置是否為null
                                rs[j] = r;  // 該位置添加Cell元素
                                created = true;  // 新Cell創(chuàng)建成功
                            }
                        } finally {
                            cellsBusy = 0;  // 釋放鎖
                        }
                        if (created)
                            break;  // 創(chuàng)建成功，跳出循環(huán)
                        continue;           // Slot is now non-empty
                    }
                }
                collide = false;  // 擴(kuò)容標(biāo)志
            }
            else if (!wasUncontended)       // 上面定位到的索引位置的值不為null
                wasUncontended = true;      // 重新計(jì)算hash，重新定位其他索引位置重試
            else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                         fn.applyAsLong(v, x))))  // 嘗試在該索引位置進(jìn)行累加
                break;
            else if (n >= NCPU || cells != as)  // 如果數(shù)組長(zhǎng)度大于等于CPU核心數(shù)，就不能在擴(kuò)容
                collide = false;            // At max size or stale
            else if (!collide)  // 數(shù)組長(zhǎng)度沒(méi)有達(dá)到最大值，修改擴(kuò)容標(biāo)志可以擴(kuò)容
                collide = true;
            else if (cellsBusy == 0 && casCellsBusy()) { // 嘗試加鎖
                try {
                    if (cells == as) {      // Expand table unless stale
                        Cell[] rs = new Cell[n << 1];  // 創(chuàng)建一個(gè)原來(lái)長(zhǎng)度2倍的數(shù)組
                        for (int i = 0; i < n; ++i)
                            rs[i] = as[i];  // 把原來(lái)的元素拷貝到新數(shù)組中
                        cells = rs;  // cells指向新數(shù)組
                    }
                } finally {
                    cellsBusy = 0;  // 釋放鎖
                }
                collide = false;  // 已經(jīng)擴(kuò)容完成，修改標(biāo)志不用再擴(kuò)容
                continue;                   // Retry with expanded table
            }
            h = advanceProbe(h);  // 重新獲取hash值
        }
        // 省略...
}

根據(jù)代碼中的注釋分析一遍整體邏輯

首先如果找到數(shù)組某個(gè)位置上的值為null，說(shuō)明可以在這個(gè)位置進(jìn)行操作，就創(chuàng)建一個(gè)新的Cell并初始化值為1放到這個(gè)位置，如果失敗了就重新計(jì)算hash值再重試
定位到的位置已經(jīng)有值了，說(shuō)明線程之間產(chǎn)生了競(jìng)爭(zhēng)，如果wasUncontended是false就修改為true，并重新計(jì)算hash重試
定位的位置有值并且wasUncontended已經(jīng)是true，就嘗試在該位置進(jìn)行累加
當(dāng)累加失敗時(shí)，判斷數(shù)組容量是否已經(jīng)達(dá)到最大，如果是就不能進(jìn)行擴(kuò)容，只能rehash并重試
如果前面條件都不滿足，并且擴(kuò)容標(biāo)志collide標(biāo)記為false的話就修改為true，表示可以進(jìn)行擴(kuò)容，然后rehash重試
首先嘗試加鎖，成功了就進(jìn)行擴(kuò)容操作，每次擴(kuò)容長(zhǎng)度是之前的2倍，然后把原來(lái)數(shù)組內(nèi)容拷貝到新數(shù)組，擴(kuò)容就完成了。