Java高並發之無鎖與Atomic源碼分析

目錄

無鎖即無障礙的運行, 所有線程都可以到達臨界區, 接近於無等待.

無鎖采用CAS(compare and swap)算法來處理線程沖突, 其原理如下

CAS原理

CAS包含3個參數CAS(V,E,N).V表示要更新的變量, E表示預期值, N表示新值.

僅當V值等於E值時, 才會將V的值設為N, 如果V值和E值不同, 則說明已經有其他線程做了更新, 則當前線程什么

都不做. 最后, CAS返回當前V的真實值. CAS操作是抱着樂觀的態度進行的, 它總是認為自己可以成功完成操作.

當多個線程同時使用CAS操作一個變量時, 只有一個會勝出, 並成功更新, 其余均會失敗.失敗的線程不會被掛起,

僅是被告知失敗, 並且允許再次嘗試, 當然也允許失敗的線程放棄操作.基於這樣的原理, CAS操作即時沒有鎖,

也可以發現其他線程對當前線程的干擾, 並進行恰當的處理.

CPU指令

另外, 雖然上述步驟繁多, 實際上CAS整一個操作過程是一個原子操作, 它是由一條CPU指令完成的,

從指令層保證操作可靠, 不會被多線程干擾.

無鎖與volatile

無鎖可以通過cas來保證原子性與線程安全, 他與volatile什么區別呢?

當給變量加了volatile關鍵字, 表示該變量對所有線程可見, 但不保證原子性.

以volatile i, i++為例, 分為以下四步:

• 加載i
• 對i進行+1
• 回寫i的值
• 用內存屏障通知其他線程i的值

其中前三步是線程不安全的, 可能其他線程會對i進行讀寫.

因此任何依賴於之前值的操作, 如i++, i = i *10使用volatile都不安全.

而諸如get/set, boolean這類可以使用volatile.

AtomicInteger

主要接口

// 取得當前值
public final int get() 
// 設置當前值
public final void set(int newValue)
// 設置新值，並返回舊值
public final int getAndSet(int newValue)
// 如果當前值為expect，則設置為u
public final boolean compareAndSet(int expect, int u)
// 當前值加1，返回舊值
public final int getAndIncrement()
// 當前值減1，返回舊值
public final int getAndDecrement() 
// 當前值增加delta，返回舊值
public final int getAndAdd(int delta)
// 當前值加1，返回新值
public final int incrementAndGet() 
// 當前值減1，返回新值
public final int decrementAndGet() 
// 當前值增加delta，返回新值
public final int addAndGet(int delta) 

源碼實現

    // 封裝了一個int對其加減
    private volatile int value;
    .......
    public final boolean compareAndSet(int expect, int update) {
    // 通過unsafe 基於CPU的CAS指令來實現, 可以認為無阻塞.
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    }
    .......
    public final int getAndIncrement() {
        for (;;) {
        // 當前值
            int current = get();
        // 預期值
            int next = current + 1;
            if (compareAndSet(current, next)) {
        // 如果加成功了, 則返回當前值
                return current;
        }
        // 如果加失敗了, 說明其他線程已經修改了數據, 與期望不相符,
        // 則繼續無限循環, 直到成功. 這種樂觀鎖, 理論上只要等兩三個時鍾周期就可以設值成功
        // 相比於直接通過synchronized獨占鎖的方式操作int, 要大大節約等待時間.
        }
    }

Demo

使用10個線程打印0-10000, 最終得到結果10w.

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicIntegerDemo {
    static AtomicInteger i = new AtomicInteger();

    public static class AddThread implements Runnable {
        public void run() {
            for (int k = 0; k < 10000; k++) {
                i.incrementAndGet();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Thread[] ts = new Thread[10];
        for (int k = 0; k < 10; k++) {
            ts[k] = new Thread(new AddThread());
        }
        for (int k = 0; k < 10; k++) {
            ts[k].start();
        }
        for (int k = 0; k < 10; k++) {
            ts[k].join();
        }
        System.out.println(i);
    }
}

Unsafe

Unsafe類是在sun.misc包下, 可以用於一些非安全的操作，比如：

根據偏移量設置值, 線程park(), 底層的CAS操作等等.

// 獲取類實例中變量的偏移量
valueOffset = unsafe.objectFieldOffset(AtomicInteger.class.getDeclaredField("value"));
// 基於偏移量對值進行操作
unsafe.compareAndSwapInt(this, valueOffset, expect, update);

主要接口

// 獲得給定對象偏移量上的int值
public native int getInt(Object o, long offset);
// 設置給定對象偏移量上的int值
public native void putInt(Object o, long offset, int x);
// 獲得字段在對象中的偏移量
public native long objectFieldOffset(Field f);
// 設置給定對象的int值，使用volatile語義
public native void putIntVolatile(Object o, long offset, int x);
// 獲得給定對象對象的int值，使用volatile語義
public native int getIntVolatile(Object o, long offset);
// 和putIntVolatile()一樣，但是它要求被操作字段就是volatile類型的
public native void putOrderedInt(Object o, long offset, int x);

AtomicReference

與AtomicInteger類似, 只是里面封裝了一個對象, 而不是int, 對引用進行修改

主要接口

get()
set(V)
compareAndSet()
getAndSet(V)

Demo

使用10個線程, 同時嘗試修改AtomicReference中的String, 最終只有一個線程可以成功.

import java.util.concurrent.atomic.AtomicReference;

public class AtomicReferenceTest {
    public final static AtomicReference<String> attxnicStr = new AtomicReference<String>("abc");

    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            new Thread() {
                public void run() {
                    try {
                        Thread.sleep(Math.abs((int) (Math.random() * 100)));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if (attxnicStr.compareAndSet("abc", "def")) {
                        System.out.println("Thread:" + Thread.currentThread().getId() + " change value to " + attxnicStr.get());
                    } else {
                        System.out.println("Thread:" + Thread.currentThread().getId() + " change failed!");
                    }
                }
            }.start();
        }
    }
}

AtomicStampedReference

也是封裝了一個引用, 主要解決ABA問題.

ABA問題

線程一准備用CAS將變量的值由A替換為B, 在此之前線程二將變量的值由A替換為C, 線程三又將C替換為A, 然后線程一執行CAS時發現變量的值仍然為A, 所以線程一CAS成功.

主要接口

// 比較設置 參數依次為：期望值 寫入新值 期望時間戳 新時間戳
public boolean compareAndSet(V expectedReference,V newReference,int expectedStamp,int newStamp)
// 獲得當前對象引用
public V getReference()
// 獲得當前時間戳
public int getStamp()
// 設置當前對象引用和時間戳
public void set(V newReference, int newStamp)

源碼分析

    // 內部封裝了一個Pair對象, 每次對對象操作的時候, stamp + 1
    private static class Pair<T> {
        final T reference;
        final int stamp;
        private Pair(T reference, int stamp) {
            this.reference = reference;
            this.stamp = stamp;
        }
        static <T> Pair<T> of(T reference, int stamp) {
            return new Pair<T>(reference, stamp);
        }
    }

    private volatile Pair<V> pair;

    // 進行cas操作的時候, 會對比stamp的值
    public boolean compareAndSet(V   expectedReference,
                                 V   newReference,
                                 int expectedStamp,
                                 int newStamp) {
        Pair<V> current = pair;
        return
            expectedReference == current.reference &&
            expectedStamp == current.stamp &&
            ((newReference == current.reference &&
              newStamp == current.stamp) ||
             casPair(current, Pair.of(newReference, newStamp)));
    }

Demo

后台使用多個線程對用戶充值, 要求只能充值一次

public class AtomicStampedReferenceDemo {
 static AtomicStampedReference<Integer> money=new AtomicStampedReference<Integer>(19,0);
    public staticvoid main(String[] args) {
        //模擬多個線程同時更新后台數據庫，為用戶充值
        for(int i = 0 ; i < 3 ; i++) {
            final int timestamp=money.getStamp();
            newThread() {  
                public void run() { 
                    while(true){
                       while(true){
                           Integerm=money.getReference();
                            if(m<20){
                         if(money.compareAndSet(m,m+20,timestamp,timestamp+1)){
          　　　　　　　　　　　　　　　 System.out.println("余額小於20元，充值成功，余額:"+money.getReference()+"元");
                                    break;
                                }
                            }else{
                               //System.out.println("余額大於20元，無需充值");
                                break ;
                             }
                       }
                    }
                } 
            }.start();
         }
        
       //用戶消費線程，模擬消費行為
        new Thread() { 
             publicvoid run() { 
                for(int i=0;i<100;i++){
                   while(true){
                        int timestamp=money.getStamp();
                        Integer m=money.getReference();
                        if(m>10){
                             System.out.println("大於10元");
                          　　if(money.compareAndSet(m, m-10,timestamp,timestamp+1)){
                      　　　　　　 System.out.println("成功消費10元，余額:"+money.getReference());
                                 break;
                             }
                        }else{
                           System.out.println("沒有足夠的金額");
                             break;
                        }
                    }
                    try {Thread.sleep(100);} catch (InterruptedException e) {}
                 }
            } 
        }.start(); 
    }
 }

AtomicIntegerArray

支持無鎖的數組

主要接口

// 獲得數組第i個下標的元素
public final int get(int i)
// 獲得數組的長度
public final int length()
// 將數組第i個下標設置為newValue，並返回舊的值
public final int getAndSet(int i, int newValue)
// 進行CAS操作，如果第i個下標的元素等於expect，則設置為update，設置成功返回true
public final boolean compareAndSet(int i, int expect, int update)
// 將第i個下標的元素加1
public final int getAndIncrement(int i)
// 將第i個下標的元素減1
public final int getAndDecrement(int i)
// 將第i個下標的元素增加delta（delta可以是負數）
public final int getAndAdd(int i, int delta)

源碼分析

    // 數組本身基地址
    private static final int base = unsafe.arrayBaseOffset(int[].class);

    // 封裝了一個數組
    private final int[] array;

    static {
        // 數組中對象的寬度, int類型, 4個字節, scale = 4;
        int scale = unsafe.arrayIndexScale(int[].class);
        if ((scale & (scale - 1)) != 0)
            throw new Error("data type scale not a power of two");
        // 前導0 : 一個數字轉為二進制后, 他前面0的個數
        // 對於4來講, 他就是00000000 00000000 00000000 00000100, 他的前導0 就是29
        // 所以shift = 2
        shift = 31 - Integer.numberOfLeadingZeros(scale);
    }

    // 獲取第i個元素
    public final int get(int i) {
        return getRaw(checkedByteOffset(i));
    }

    // 第i個元素, 在數組中的偏移量是多少
    private long checkedByteOffset(int i) {
        if (i < 0 || i >= array.length)
            throw new IndexOutOfBoundsException("index " + i);

        return byteOffset(i);
    }

    // base : 數組基地址, i << shift, 其實就是i * 4, 因為這邊是int array.
    private static long byteOffset(int i) {
        // i * 4 + base
        return ((long) i << shift) + base;
    }

    // 根據偏移量從數組中獲取數據
    private int getRaw(long offset) {
        return unsafe.getIntVolatile(array, offset);
    }

Demo

import java.util.concurrent.atomic.AtomicIntegerArray;

public class AtomicArrayDemo {
    static AtomicIntegerArray arr = new AtomicIntegerArray(10);

    public static class AddThread implements Runnable {
        public void run() {
            for (int k = 0; k < 10000; k++) {
                arr.incrementAndGet(k % arr.length());
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Thread[] ts = new Thread[10];
        for (int k = 0; k < 10; k++) {
            ts[k] = new Thread(new AddThread());
        }
        for (int k = 0; k < 10; k++) {
            ts[k].start();
        }
        for (int k = 0; k < 10; k++) {
            ts[k].join();
        }
        System.out.println(arr);
    }
}

AtomicIntegerFieldUpdater

讓普通變量也享受原子操作

主要接口

AtomicIntegerFieldUpdater.newUpdater()
incrementAndGet()

• Updater只能修改它可見范圍內的變量。因為Updater使用反射得到這個變量。如果變量不可見，就會出錯。比如如果score申明為private，就是不可行的。
• 為了確保變量被正確的讀取，它必須是volatile類型的。如果我們原有代碼中未申明這個類型，那么簡單得申明一下就行，這不會引起什么問題。
• 由於CAS操作會通過對象實例中的偏移量直接進行賦值，因此，它不支持static字段（Unsafe.objectFieldOffset()不支持靜態變量）。

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

public class AtomicIntegerFieldUpdaterDemo {
    public static class Candidate {
        int id;
        // 如果直接把int改成atomicinteger, 可能對代碼破壞比較大
        // 因此使用AtomicIntegerFieldUpdater對score進行封裝
        volatile int score;
    }

    // 通過反射實現
    public final static AtomicIntegerFieldUpdater<Candidate> scoreUpdater = AtomicIntegerFieldUpdater.newUpdater(Candidate.class, "score");
    // 檢查Updater是否工作正確, allScore的結果應該跟score一致
    public static AtomicInteger allScore = new AtomicInteger(0);

    public static void main(String[] args) throws InterruptedException {
        final Candidate stu = new Candidate();
        Thread[] t = new Thread[10000];
        for (int i = 0; i < 10000; i++) {
            t[i] = new Thread() {
                public void run() {
                    if (Math.random() > 0.4) {
                        scoreUpdater.incrementAndGet(stu);
                        allScore.incrementAndGet();
                    }
                }
            };
            t[i].start();
        }
        for (int i = 0; i < 10000; i++) {
            t[i].join();
        }

        System.out.println("score=" + stu.score);
        System.out.println("allScore=" + allScore);
    }
}

無鎖的Vector

jdk中Vector是加鎖的, 網上找的一個無鎖Vector LockFreeVector, 給他添加了源碼中文注釋.

主要關注push_back, 添加元素的函數

import java.util.AbstractList;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceArray;

/**
 * It is a thread safe and lock-free vector.
 * This class implement algorithm from:<br>
 *
 * Lock-free Dynamically Resizable Arrays <br>
 *
 * @param <E> type of element in the vector
 *
 */
public class LockFreeVector<E> extends AbstractList<E> {
    private static final boolean debug = false;
    /**
     * Size of the first bucket. sizeof(bucket[i+1])=2*sizeof(bucket[i])
     */
    private static final int FIRST_BUCKET_SIZE = 8;

    /**
     * number of buckets. 30 will allow 8*(2^30-1) elements
     */
    private static final int N_BUCKET = 30;

    /**
     * We will have at most N_BUCKET number of buckets. And we have
     * sizeof(buckets.get(i))=FIRST_BUCKET_SIZE**(i+1)
     *
     * 為什么AtomicReferenceArray里再套一個AtomicReferenceArray呢, 類似一個籃子(buckets)里放了很多籃子
     * 為了在容量擴展時希望盡可能少的改動原有數據, 因此把一維數組擴展成二維數組.
     * 該二維數組並非均衡的分布. 可能第一個數組8個元素, 第二個數組16個元素, 第三個數組32個......
     */
    private final AtomicReferenceArray<AtomicReferenceArray<E>> buckets;

    /**
     * @param <E>
     */
    static class WriteDescriptor<E> {
        public E oldV;
        public E newV;
        public AtomicReferenceArray<E> addr;
        public int addr_ind;

        /**
         * Creating a new descriptor.
         *
         * @param addr Operation address 對哪個數組進行寫
         * @param addr_ind  Index of address 指定index
         * @param oldV old operand
         * @param newV new operand
         */
        public WriteDescriptor(AtomicReferenceArray<E> addr, int addr_ind,
                               E oldV, E newV) {
            this.addr = addr;
            this.addr_ind = addr_ind;
            this.oldV = oldV;
            this.newV = newV;
        }

        /**
         * set newV.
         */
        public void doIt() {
            // 這邊失敗后重試的邏輯在另外的代碼里.
            addr.compareAndSet(addr_ind, oldV, newV);
        }
    }

    /**
     * @param <E>
     */
    static class Descriptor<E> {
        public int size;
        volatile WriteDescriptor<E> writeop;

        /**
         * Create a new descriptor.
         *
         * @param size Size of the vector
         * @param writeop Executor write operation
         */
        public Descriptor(int size, WriteDescriptor<E> writeop) {
            this.size = size;
            this.writeop = writeop;
        }

        /**
         *
         */
        public void completeWrite() {
            WriteDescriptor<E> tmpOp = writeop;
            if (tmpOp != null) {
                tmpOp.doIt();
                writeop = null; // this is safe since all write to writeop use
                // null as r_value.
            }
        }
    }

    private AtomicReference<Descriptor<E>> descriptor;
    private static final int zeroNumFirst = Integer
            .numberOfLeadingZeros(FIRST_BUCKET_SIZE);

    /**
     * Constructor.
     */
    public LockFreeVector() {
        buckets = new AtomicReferenceArray<AtomicReferenceArray<E>>(N_BUCKET);
        buckets.set(0, new AtomicReferenceArray<E>(FIRST_BUCKET_SIZE));
        descriptor = new AtomicReference<Descriptor<E>>(new Descriptor<E>(0,
                null));
    }

    /**
     * add e at the end of vector.
     * 把元素e加到vector中
     *
     * @param e
     *            element added
     */
    public void push_back(E e) {
        Descriptor<E> desc;
        Descriptor<E> newd;
        do {
            desc = descriptor.get();
            desc.completeWrite();
            // desc.size   Vector 本身的大小
            // FIRST_BUCKET_SIZE  第一個一維數組的大小
            int pos = desc.size + FIRST_BUCKET_SIZE;
            // 取出pos 的前導0
            int zeroNumPos = Integer.numberOfLeadingZeros(pos);
            // zeroNumFirst  為FIRST_BUCKET_SIZE 的前導0
            // bucketInd 數據應該放到哪一個一維數組(籃子)里的
            int bucketInd = zeroNumFirst - zeroNumPos;
            // 00000000 00000000 00000000 00001000 第一個籃子滿 8
            // 00000000 00000000 00000000 00011000 第二個籃子滿 8 + 16
            // 00000000 00000000 00000000 00111000 第三個籃子滿 8 + 16 + 32
            // ... bucketInd其實通過前導0相減, 就是為了得出來當前第幾個籃子是空的.

            // 判斷這個一維數組是否已經啟用, 可能是第一次初始化
            if (buckets.get(bucketInd) == null) {
                //newLen  一維數組的長度, 取前一個數組長度 * 2
                int newLen = 2 * buckets.get(bucketInd - 1).length();
                // 設置失敗也沒關系, 只要有人初始化成功就行
                buckets.compareAndSet(bucketInd, null,
                        new AtomicReferenceArray<E>(newLen));
            }

            // 在這個一位數組中，我在哪個位置
            // 0x80000000是 10000000 00000000 00000000 00000000
            // 這句話就是把上述111000, 第一個1變成了0, 得到011000, 即新值的位置.
            int idx = (0x80000000>>>zeroNumPos) ^ pos;
            // 通過bucketInd與idx來確定元素在二維數組中的位置
            // 期望寫入的時候, 該位置值是null, 如果非null, 說明其他線程已經寫了, 則繼續循環.
            newd = new Descriptor<E>(desc.size + 1, new WriteDescriptor<E>(
                    buckets.get(bucketInd), idx, null, e));
            // 循環cas設值
        } while (!descriptor.compareAndSet(desc, newd));
        descriptor.get().completeWrite();
    }

    /**
     * Remove the last element in the vector.
     *
     * @return element removed
     */
    public E pop_back() {
        Descriptor<E> desc;
        Descriptor<E> newd;
        E elem;
        do {
            desc = descriptor.get();
            desc.completeWrite();

            int pos = desc.size + FIRST_BUCKET_SIZE - 1;
            int bucketInd = Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE)
                    - Integer.numberOfLeadingZeros(pos);
            int idx = Integer.highestOneBit(pos) ^ pos;
            elem = buckets.get(bucketInd).get(idx);
            newd = new Descriptor<E>(desc.size - 1, null);
        } while (!descriptor.compareAndSet(desc, newd));

        return elem;
    }

    /**
     * Get element with the index.
     *
     * @param index
     *            index
     * @return element with the index
     */
    @Override
    public E get(int index) {
        int pos = index + FIRST_BUCKET_SIZE;
        int zeroNumPos = Integer.numberOfLeadingZeros(pos);
        int bucketInd = zeroNumFirst - zeroNumPos;
        int idx = (0x80000000>>>zeroNumPos) ^ pos;
        return buckets.get(bucketInd).get(idx);
    }

    /**
     * Set the element with index to e.
     *
     * @param index
     *            index of element to be reset
     * @param e
     *            element to set
     */
    /**
     * {@inheritDoc}
     */
    public E set(int index, E e) {
        int pos = index + FIRST_BUCKET_SIZE;
        int bucketInd = Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE)
                - Integer.numberOfLeadingZeros(pos);
        int idx = Integer.highestOneBit(pos) ^ pos;
        AtomicReferenceArray<E> bucket = buckets.get(bucketInd);
        while (true) {
            E oldV = bucket.get(idx);
            if (bucket.compareAndSet(idx, oldV, e))
                return oldV;
        }
    }

    /**
     * reserve more space.
     *
     * @param newSize
     *            new size be reserved
     */
    public void reserve(int newSize) {
        int size = descriptor.get().size;
        int pos = size + FIRST_BUCKET_SIZE - 1;
        int i = Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE)
                - Integer.numberOfLeadingZeros(pos);
        if (i < 1)
            i = 1;

        int initialSize = buckets.get(i - 1).length();
        while (i < Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE)
                - Integer.numberOfLeadingZeros(newSize + FIRST_BUCKET_SIZE - 1)) {
            i++;
            initialSize *= FIRST_BUCKET_SIZE;
            buckets.compareAndSet(i, null, new AtomicReferenceArray<E>(
                    initialSize));
        }
    }

    /**
     * size of vector.
     *
     * @return size of vector
     */
    public int size() {
        return descriptor.get().size;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean add(E object) {
        push_back(object);
        return true;
    }
}