多線程編程學習八(原子操作類).

簡介

原子（atomic）本意是“不能被進一步分割的最小粒子”，而原子操作（atomic operation）意為“不可被中斷的一個或一系列操作”。

Java 在 JDK 1.5 中提供了 java.util.concurrent.atomic 包，這個包中的原子操作類提供了一種用法簡單、性能高效、線程安全地更新一個變量的方式。主要提供了四種類型的原子更新方式，分別是原子更新基本類型、原子更新數組、原子更新引用和原子更新屬性。

Atomic 類基本都是使用 Unsafe 來保證線程安全。

public final class Unsafe {
    ...

    public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5);

    public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);

    public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);

    ...
}

JDK 1.8 中，Doug Lea 又在 atomic 包中新增了 LongAccumulator 等並行累加器，提供了更高效的無鎖解決方案。

原子更新基本數據類型

• AtomicBoolean：原子更新布爾類型
• AtomicInteger：原子更新整型
• AtomicLong：原子更新長整型

public class AtomicIntegerTest {

    private static CountDownLatch countDownLatch = new CountDownLatch(1);

    private static AtomicInteger atomicInteger = new AtomicInteger(1);

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            Thread thread = new Thread(() -> {
                try {
                    countDownLatch.await();
                    // 以原子方式將當前值加 1，並返回之前的值
                    System.out.print(atomicInteger.getAndIncrement() + " ");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            thread.start();
        }
        // 線程同時進行爭搶操作
        countDownLatch.countDown();
        Thread.sleep(2000);
        System.out.println();
        // 以原子方式將輸入的數值與實例中的值相加，並返回結果。
        System.out.println(atomicInteger.addAndGet(10));
        // CAS 操作
        atomicInteger.compareAndSet(21, 30);
        System.out.println(atomicInteger.get());
    }
}

原子更新數組

• AtomicIntegerArray：原子更新整型數組里的元素
• AtomicLongArray：原子更新長整型數組里的元素
• AtomicReferenceArray：原子更新引用類型數組里的元素

public class AtomicReferenceArrayTest {

    // AtomicReferenceArray 會將當前數組（VALUE）復制一份，所以當 AtomicReferenceArray 對內部的數組元素進行修改時，不會影響傳入的數組。
    private static Stu[] VALUE = new Stu[]{new Stu(System.currentTimeMillis(), "張三"),new Stu(System.currentTimeMillis(), "李四")};

    private static AtomicReferenceArray<Stu> REFERENCE_ARRAY = new AtomicReferenceArray<>(VALUE);

    public static void main(String[] args) {
        // 修改指定位置元素的值
        REFERENCE_ARRAY.getAndSet(0, new Stu(System.currentTimeMillis(), "王五"));
        System.out.println(REFERENCE_ARRAY.get(0));
        System.out.println(VALUE[0]);
    }
}

原子更新引用

• AtomicReference：原子更新引用類型
• AtomicMarkableReference：原子更新帶有標記位的引用類型
• AtomicStampedReference：原子更新帶有版本號的引用類型

public class AtomicStampedReferenceTest {

    private static Stu stu = new Stu(System.currentTimeMillis(), "張三");
    /**
     * 更新對象的時候帶一個版本號，可以防止 CAS 中 ABA 問題。原理在於 compare 的時候不僅比較原來的值，還比較版本號。同理更新的時候也需要更新版本號
     */
    private static AtomicStampedReference<Stu> stampedReference = new AtomicStampedReference(stu, 1);

    public static void main(String[] args) {
        System.out.println(stampedReference.getReference());
        Stu newStu = new Stu(System.currentTimeMillis(), "李四");
        int stamp = stampedReference.getStamp();
        stampedReference.compareAndSet(stu, newStu, stamp, stamp++);
        System.out.println(stampedReference.getReference());
    }
}

原子更新屬性

• AtomicIntegerFieldUpdater：原子更新整型的字段的更新器
• AtomicLongFieldUpdater：原子更新長整型字段的更新器
• AtomicReferenceFieldUpdater：原子更新引用類型里的字段

public class AtomicReferenceFieldUpdaterTest {

    // 創建原子更新器，並設置需要更新的對象類和對象的屬性
    private static AtomicReferenceFieldUpdater<Stu, String> atomicUserFieldRef = AtomicReferenceFieldUpdater.newUpdater(Stu.class, String.class, "name");

    public static void main(String[] args) {
        Stu stu = new Stu(System.currentTimeMillis(), "張三");
        atomicUserFieldRef.set(stu, "李四");
        System.out.println(stu.getName());
    }
}

需要注意的是，更新類的屬性必須使用 public volatile 修飾符。以下是 AtomicReferenceFieldUpdater 的源碼內容：

            if (vclass.isPrimitive())
                throw new IllegalArgumentException("Must be reference type");

            if (!Modifier.isVolatile(modifiers))
                throw new IllegalArgumentException("Must be volatile type");

1.8 的並行累加器

AtomicLong 維護一個變量 value，通過 CAS 提供非阻塞的原子性操作。不足的是，CAS 失敗后需要通過無限循環的自旋鎖不斷嘗試，這在高並發N多線程下，將大大浪費 CPU 資源。（這也是其他 Atomic 原子類的通病）

那么如果把一個變量分解為多個變量，讓同樣多的線程去競爭多個資源那么性能問題不就解決了？是的，JDK8提供的 LongAdder 就是這個思路。

LongAdder 的核心思想是分段，它繼承自 Striped64，Striped64 有兩個參數 long base 和 Cell[] cells ,接着來看看 LongAddr 的核心代碼：

public void add(long x) {
        Cell[] as; long b, v; int m; Cell a;
        //想要add一個元素的時候，先看一下 cells 數組是否為空，如果是空的就嘗試去看能不能直接加到 base上面，如果線程競爭很小就加到 base上面了，函數結束
        //如果 cells 是空的，並且競爭很大，cas 失敗，就進入if塊內，創建 cells
        //如果不是空的就進入到 cell 數組中看能加到哪個上面去
        if ((as = cells) != null || !casBase(b = base, b + x)) {
            boolean uncontended = true;
            //如果 cells 是空的，就執行增加操作
            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);
        }
    }

所以想要得到累加的結果，只能調用 LongAdder 的 sum() 方法，即 base + cell[] 數組元素的和。需要注意的是，在計算總和時發生的並發更新可能不會被合並。