:: :: :: ::

::

公告

Atomic原子操作類介紹

在并發編程中很容易出現并發安全的問題，有一個很簡單的例子就是多線程更新變量i=1,比如多個線程執行i++操作，就有可能獲取不到正確的值，而這個問題，最常用的方法是通過Synchronized進行控制來達到線程安全的目的。但是由于synchronized是采用的是悲觀鎖策略，并不是特別高效的一種解決方案。實際上，在J.U.C下的atomic包提供了一系列的操作簡單，性能高效，并能保證線程安全的類去更新基本類型變量，數組元素，引用類型以及更新對象中的字段類型。atomic包下的這些類都是采用的是樂觀鎖策略去原子更新數據，在java中則是使用CAS操作具體實現。

在java.util.concurrent.atomic包里提供了一組原子操作類：

基本類型：AtomicInteger、AtomicLong、AtomicBoolean；

引用類型：AtomicReference、AtomicStampedRerence、AtomicMarkableReference；

數組類型：AtomicIntegerArray、AtomicLongArray、AtomicReferenceArray

對象屬性原子修改器：AtomicIntegerFieldUpdater、AtomicLongFieldUpdater、AtomicReferenceFieldUpdater

原子類型累加器（jdk1.8增加的類）：DoubleAccumulator、DoubleAdder、LongAccumulator、LongAdder、Striped64

原子更新基本類型

AtomicInteger

以AtomicInteger為例總結常用的方法

//以原子的方式將實例中的原值加1，返回的是自增前的舊值；
public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}
 
//getAndSet(int newValue)：將實例中的值更新為新值，并返回舊值；
public final boolean getAndSet(boolean newValue) {
    boolean prev;
    do {
        prev = get();
    } while (!compareAndSet(prev, newValue));
    return prev;
}
 
//incrementAndGet() ：以原子的方式將實例中的原值進行加1操作，并返回最終相加后的結果；
public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}
 
//addAndGet(int delta) ：以原子方式將輸入的數值與實例中原本的值相加，并返回最后的結果；
public final int addAndGet(int delta) {
    return unsafe.getAndAddInt(this, valueOffset, delta) + delta;
}

測試

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicIntegerTest {

    static AtomicInteger sum = new AtomicInteger(0);

    public static void main(String[] args) {

        for (int i = 0; i < 10; i++) {
            Thread thread = new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    // 原子自增  CAS
                    sum.incrementAndGet();
                    //count++;

                }
            });
            thread.start();
        }

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(sum.get());

    }

}

incrementAndGet()方法通過CAS自增實現，如果CAS失敗，自旋直到成功+1。

public final int incrementAndGet() {
   return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}

public final int getAndAddInt(Object var1, long var2, int var4) {
    int var5;
    do {
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

    return var5;
}

原子更新數組類型

AtomicIntegerArray

AtomicIntegerArray為例總結常用的方法

//addAndGet(int i, int delta)：以原子更新的方式將數組中索引為i的元素與輸入值相加；
public final int addAndGet(int i, int delta) {
    return getAndAdd(i, delta) + delta;
}
 
//getAndIncrement(int i)：以原子更新的方式將數組中索引為i的元素自增加1；
public final int getAndIncrement(int i) {
    return getAndAdd(i, 1);
}
 
//compareAndSet(int i, int expect, int update)：將數組中索引為i的位置的元素進行更新
public final boolean compareAndSet(int i, int expect, int update) {
    return compareAndSetRaw(checkedByteOffset(i), expect, update);
}

測試

import java.util.concurrent.atomic.AtomicIntegerArray;

public class AtomicIntegerArrayTest {

    static int[] value = new int[]{ 1, 2, 3, 4, 5 };
    static AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(value);


    public static void main(String[] args) throws InterruptedException {

        //設置索引0的元素為100
        atomicIntegerArray.set(0, 100);
        System.out.println(atomicIntegerArray.get(0));
        //以原子更新的方式將數組中索引為1的元素與輸入值相加
        atomicIntegerArray.getAndAdd(1,5);

        System.out.println(atomicIntegerArray);
    }
}

原子更新引用類型

AtomicReference

AtomicReference作用是對普通對象的封裝，它可以保證你在修改對象引用時的線程安全性。

import java.util.concurrent.atomic.AtomicReference;
import lombok.AllArgsConstructor;
import lombok.Data;

public class AtomicReferenceTest {

    public static void main( String[] args ) {
        User user1 = new User("張三", 23);
        User user2 = new User("李四", 25);
        User user3 = new User("王五", 20);

        //初始化為 user1
        AtomicReference<User> atomicReference = new AtomicReference<>();
        atomicReference.set(user1);

        //把 user2 賦給 atomicReference
        atomicReference.compareAndSet(user1, user2);
        System.out.println(atomicReference.get());

        //把 user3 賦給 atomicReference
        atomicReference.compareAndSet(user1, user3);
        System.out.println(atomicReference.get());

    }

}


@Data
@AllArgsConstructor
class User {
    private String name;
    private Integer age;
}

對象屬性原子修改器

AtomicIntegerFieldUpdater

AtomicIntegerFieldUpdater可以線程安全地更新對象中的整型變量。

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

public class AtomicIntegerFieldUpdaterTest {


    public static class Candidate {
        //字段必須是volatile類型
        volatile int score = 0;

        AtomicInteger score2 = new AtomicInteger();
    }

    public static final AtomicIntegerFieldUpdater<Candidate> scoreUpdater =
            AtomicIntegerFieldUpdater.newUpdater(Candidate.class, "score");

    public static AtomicInteger realScore = new AtomicInteger(0);

    public static void main(String[] args) throws InterruptedException {

        final Candidate candidate = new Candidate();

        Thread[] t = new Thread[10000];
        for (int i = 0; i < 10000; i++) {
            t[i] = new Thread(new Runnable() {
                @Override
                public void run() {
                    if (Math.random() > 0.4) {
                        candidate.score2.incrementAndGet();
                        scoreUpdater.incrementAndGet(candidate);
                        realScore.incrementAndGet();
                    }
                }
            });
            t[i].start();
        }
        for (int i = 0; i < 10000; i++) {
            t[i].join();
        }
        System.out.println("AtomicIntegerFieldUpdater Score=" + candidate.score);
        System.out.println("AtomicInteger Score=" + candidate.score2.get());
        System.out.println("realScore=" + realScore.get());

    }
}

對于AtomicIntegerFieldUpdater 的使用稍微有一些限制和約束，約束如下：

（1）字段必須是volatile類型的，在線程之間共享變量時保證立即可見.eg:volatile int value = 3
（2）字段的描述類型（修飾符public/protected/default/private）與調用者與操作對象字段的關系一致。也就是說調用者能夠直接操作對象字段，那么就可以反射進行原子操作。但是對于父類的字段，子類是不能直接操作的，盡管子類可以訪問父類的字段。
（3）只能是實例變量，不能是類變量，也就是說不能加static關鍵字。
（4）只能是可修改變量，不能使final變量，因為final的語義就是不可修改。實際上final的語義和volatile是有沖突的，這兩個關鍵字不能同時存在。
（5）對于AtomicIntegerFieldUpdater和AtomicLongFieldUpdater只能修改int/long類型的字段，不能修改其包裝類型（Integer/Long）。如果要修改包裝類型就需要使用AtomicReferenceFieldUpdater。

LongAdder/DoubleAdder詳解

AtomicLong是利用了底層的CAS操作來提供并發性的，比如addAndGet方法：

public final long getAndAdd(long delta) {
        return unsafe.getAndAddLong(this, valueOffset, delta);
    }

public final long getAndAddLong(Object var1, long var2, long var4) {
        long var6;
        do {
            var6 = this.getLongVolatile(var1, var2);
        } while(!this.compareAndSwapLong(var1, var2, var6, var6 + var4));

        return var6;
    }

上述方法調用了Unsafe類的getAndAddLong方法，該方法內部是個native方法，它的邏輯是采用自旋的方式不斷更新目標值，直到更新成功。

在并發量較低的環境下，線程沖突的概率比較小，自旋的次數不會很多。但是，高并發環境下，N個線程同時進行自旋操作，會出現大量失敗并不斷自旋的情況，此時AtomicLong的自旋會成為瓶頸。

這就是LongAdder引入的初衷——解決高并發環境下AtomicInteger，AtomicLong的自旋瓶頸問題。

LongAdder

性能測試

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

public class LongAdderTest {

    public static void main(String[] args) {
        LongAdder longAdder=new LongAdder();
        longAdder.add(1);
        testAtomicLongVSLongAdder(10, 10000);
        System.out.println("==================");
        testAtomicLongVSLongAdder(10, 200000);
        System.out.println("==================");
        testAtomicLongVSLongAdder(100, 200000);
    }

    static void testAtomicLongVSLongAdder(final int threadCount, final int times) {
        try {
            long start = System.currentTimeMillis();
            testLongAdder(threadCount, times);
            long end = System.currentTimeMillis() - start;
            System.out.println("條件>>>>>>線程數:" + threadCount + ", 單線程操作計數" + times);
            System.out.println("結果>>>>>>LongAdder方式增加計數" + (threadCount * times) + "次,共計耗時:" + end);

            long start2 = System.currentTimeMillis();
            testAtomicLong(threadCount, times);
            long end2 = System.currentTimeMillis() - start2;
            System.out.println("條件>>>>>>線程數:" + threadCount + ", 單線程操作計數" + times);
            System.out.println("結果>>>>>>AtomicLong方式增加計數" + (threadCount * times) + "次,共計耗時:" + end2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static void testAtomicLong(final int threadCount, final int times) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        AtomicLong atomicLong = new AtomicLong();
        for (int i = 0; i < threadCount; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int j = 0; j < times; j++) {
                        atomicLong.incrementAndGet();
                    }
                    countDownLatch.countDown();
                }
            }, "my-thread" + i).start();
        }
        countDownLatch.await();
    }

    static void testLongAdder(final int threadCount, final int times) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        LongAdder longAdder = new LongAdder();
        for (int i = 0; i < threadCount; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int j = 0; j < times; j++) {
                        longAdder.add(1);
                    }
                    countDownLatch.countDown();
                }
            }, "my-thread" + i).start();
        }

        countDownLatch.await();
    }
}

測試結果：線程數越多，并發操作數越大，LongAdder的優勢越明顯

條件>>>>>>線程數:10, 單線程操作計數10000
結果>>>>>>LongAdder方式增加計數100000次,共計耗時:10
條件>>>>>>線程數:10, 單線程操作計數10000
結果>>>>>>AtomicLong方式增加計數100000次,共計耗時:5
==================
條件>>>>>>線程數:10, 單線程操作計數200000
結果>>>>>>LongAdder方式增加計數2000000次,共計耗時:18
條件>>>>>>線程數:10, 單線程操作計數200000
結果>>>>>>AtomicLong方式增加計數2000000次,共計耗時:41
==================
條件>>>>>>線程數:100, 單線程操作計數200000
結果>>>>>>LongAdder方式增加計數20000000次,共計耗時:67
條件>>>>>>線程數:100, 單線程操作計數200000
結果>>>>>>AtomicLong方式增加計數20000000次,共計耗時:364

低并發、一般的業務場景下AtomicLong是足夠了。如果并發量很多，存在大量寫多讀少的情況，那LongAdder可能更合適。

LongAdder原理

設計思路

AtomicLong中有個內部變量value保存著實際的long值，所有的操作都是針對該變量進行。也就是說，高并發環境下，value變量其實是一個熱點，也就是N個線程競爭一個熱點。LongAdder的基本思路就是分散熱點，將value值分散到一個數組中，不同線程會命中到數組的不同槽中，各個線程只對自己槽中的那個值進行CAS操作，這樣熱點就被分散了，沖突的概率就小很多。如果要獲取真正的long值，只要將各個槽中的變量值累加返回。

LongAdder的內部結構

LongAdder內部有一個base變量，一個Cell[]數組：

base變量：非競態條件下，直接累加到該變量上

Cell[]數組：競態條件下，累加個各個線程自己的槽Cell[i]中

/** Number of CPUS, to place bound on table size */
// CPU核數，用來決定槽數組的大小
static final int NCPU = Runtime.getRuntime().availableProcessors();

/**
 * Table of cells. When non-null, size is a power of 2.
 */
 // 數組槽，大小為2的次冪
transient volatile Cell[] cells;

/**
 * Base value, used mainly when there is no contention, but also as
 * a fallback during table initialization races. Updated via CAS.
 */
 /**
 *  基數，在兩種情況下會使用：
 *  1. 沒有遇到并發競爭時，直接使用base累加數值
 *  2. 初始化cells數組時，必須要保證cells數組只能被初始化一次（即只有一個線程能對cells初始化），
 *  其他競爭失敗的線程會講數值累加到base上
 */
transient volatile long base;

/**
 * Spinlock (locked via CAS) used when resizing and/or creating Cells.
 */
 transient volatile int cellsBusy;

Cell

@sun.misc.Contended是jdk提供的注解用于填充緩存行

定義了一個內部Cell類，這就是我們之前所說的槽，每個Cell對象存有一個value值，可以通過Unsafe來CAS操作它的值：

@sun.misc.Contended static final class Cell {
    volatile long value;
    Cell(long x) { value = x; }
    final boolean cas(long cmp, long val) {
        return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
    }

    // Unsafe mechanics
    private static final sun.misc.Unsafe UNSAFE;
    private static final long valueOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> ak = Cell.class;
            valueOffset = UNSAFE.objectFieldOffset
                (ak.getDeclaredField("value"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }
}

LongAdder#add方法

LongAdder#add方法的邏輯如下圖：

只有從未出現過并發沖突的時候，base基數才會使用到，一旦出現了并發沖突，之后所有的操作都只針對Cell[]數組中的單元Cell。

如果Cell[]數組未初始化，會調用父類的longAccumelate去初始化Cell[]，如果Cell[]已經初始化但是沖突發生在Cell單元內，則也調用父類的longAccumelate，此時可能就需要對Cell[]擴容了。

這也是LongAdder設計的精妙之處：盡量減少熱點沖突，不到最后萬不得已，盡量將CAS操作延遲。

Striped64#longAccumulate方法

整個Striped64#longAccumulate的流程圖如下：

LongAdder#sum方法

/**
* 返回累加的和，也就是"當前時刻"的計數值
* 注意： 高并發時，除非全局加鎖，否則得不到程序運行中某個時刻絕對準確的值
*  此返回值可能不是絕對準確的，因為調用這個方法時還有其他線程可能正在進行計數累加,
*  方法的返回時刻和調用時刻不是同一個點，在有并發的情況下，這個值只是近似準確的計數值
*/
public long sum() {
    Cell[] as = cells; Cell a;
    long sum = base;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

由于計算總和時沒有對Cell數組進行加鎖，所以在累加過程中可能有其他線程對Cell中的值進行了修改，也有可能對數組進行了擴容，所以sum返回的值并不是非常精確的，其返回值并不是一個調用sum方法時的原子快照值。

LongAccumulator

LongAccumulator是LongAdder的增強版。LongAdder只能針對數值的進行加減運算，而LongAccumulator提供了自定義的函數操作。其構造函數如下：

public class LongAccumulator extends Striped64 implements Serializable {
    private static final long serialVersionUID = 7249069246863182397L;

    private final LongBinaryOperator function;
    private final long identity;

    /**
     * Creates a new instance using the given accumulator function
     * and identity element.
     * @param accumulatorFunction a side-effect-free function of two arguments
     * @param identity identity (initial value) for the accumulator function
     */
    public LongAccumulator(LongBinaryOperator accumulatorFunction,
                           long identity) {
        this.function = accumulatorFunction;
        base = this.identity = identity;
    }
 
}

通過LongBinaryOperator，可以自定義對入參的任意操作，并返回結果（LongBinaryOperator接收2個long作為參數，并返回1個long）。

@FunctionalInterface
public interface LongBinaryOperator {

    /**
     * Applies this operator to the given operands.
     *
     * @param left the first operand
     * @param right the second operand
     * @return the operator result
     */
    long applyAsLong(long left, long right);
}

LongAccumulator內部原理和LongAdder幾乎完全一樣，都是利用了父類Striped64的longAccumulate方法。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.LongAccumulator;
import java.util.stream.IntStream;

public class LongAccumulatorTest {

    public static void main(String[] args) throws InterruptedException {
        // 累加 x+y
        LongAccumulator accumulator = new LongAccumulator((x, y) -> x + y, 0);

        ExecutorService executor = Executors.newFixedThreadPool(8);
        // 1到9累加
        IntStream.range(1, 10).forEach(i -> executor.submit(() -> accumulator.accumulate(i)));

        Thread.sleep(2000);
        System.out.println(accumulator.getThenReset());

    }
}

posted on 2022-03-09 00:06 路仁甲閱讀(267) 評論(0) 收藏舉報

刷新頁面返回頂部