問題

（1）什麼是ABA？

（2）ABA的危害？

（3）ABA的解決方法？

（4）AtomicStampedReference是什麼？

（5）AtomicStampedReference是怎麼解決ABA的？

簡介

AtomicStampedReference是java併發包下提供的一個原子類，它能解決其它原子類無法解決的ABA問題。

ABA

ABA問題發生在多執行緒環境中，當某執行緒連續讀取同一塊記憶體地址兩次，兩次得到的值一樣，它簡單地認為“此記憶體地址的值並沒有被修改過”，然而，同時可能存在另一個執行緒在這兩次讀取之間把這個記憶體地址的值從A修改成了B又修改回了A，這時還簡單地認為“沒有修改過”顯然是錯誤的。

比如，兩個執行緒按下面的順序執行：

（1）執行緒1讀取記憶體位置X的值為A；

（2）執行緒1阻塞了；

（3）執行緒2讀取記憶體位置X的值為A；

（4）執行緒2修改記憶體位置X的值為B；

（5）執行緒2修改又記憶體位置X的值為A；

（6）執行緒1恢復，繼續執行，比較發現還是A把記憶體位置X的值設定為C；

可以看到，針對執行緒1來說，第一次的A和第二次的A實際上並不是同一個A。

ABA問題通常發生在無鎖結構中，用程式碼來表示上面的過程大概就是這樣：

public class ABATest {

    public static void main(String[] args) {
        AtomicInteger atomicInteger = new AtomicInteger(1);

        new Thread(()->{
            int value = atomicInteger.get();
            System.out.println("thread 1 read value: " + value);

            // 阻塞1s
            LockSupport.parkNanos(1000000000L);

            if (atomicInteger.compareAndSet(value, 3)) {
                System.out.println("thread 1 update from " + value + " to 3");
            } else {
                System.out.println("thread 1 update fail!");
            }
        }).start();

        new Thread(()->{
            int value = atomicInteger.get();
            System.out.println("thread 2 read value: " + value);
            if (atomicInteger.compareAndSet(value, 2)) {
                System.out.println("thread 2 update from " + value + " to 2");

                // do sth

                value = atomicInteger.get();
                System.out.println("thread 2 read value: " + value);
                if (atomicInteger.compareAndSet(value, 1)) {
                    System.out.println("thread 2 update from " + value + " to 1");
                }
            }
        }).start();
    }
}
 

列印結果為：

thread 1 read value: 1
thread 2 read value: 1
thread 2 update from 1 to 2
thread 2 read value: 2
thread 2 update from 2 to 1
thread 1 update from 1 to 3

ABA的危害

為了更好地理解ABA的危害，我們還是來看一個現實點的例子。

假設我們有一個無鎖的棧結構，如下：

public class ABATest {

    static class Stack {
        // 將top放在原子類中
        private AtomicReference<Node> top = new AtomicReference<>();
        // 棧中節點資訊
        static class Node {
            int value;
            Node next;

            public Node(int value) {
                this.value = value;
            }
        }
        // 出棧操作
        public Node pop() {
            for (;;) {
                // 獲取棧頂節點
                Node t = top.get();
                if (t == null) {
                    return null;
                }
                // 棧頂下一個節點
                Node next = t.next;
                // CAS更新top指向其next節點
                if (top.compareAndSet(t, next)) {
                    // 把棧頂元素彈出，應該把next清空防止外面直接操作棧
                    t.next = null;
                    return t;
                }
            }
        }
        // 入棧操作
        public void push(Node node) {
            for (;;) {
                // 獲取棧頂節點
                Node next = top.get();
                // 設定棧頂節點為新節點的next節點
                node.next = next;
                // CAS更新top指向新節點
                if (top.compareAndSet(next, node)) {
                    return;
                }
            }
        }
    }
}
 

咋一看，這段程式似乎沒有什麼問題，然而試想以下情形。

假如，我們初始化棧結構為 top->1->2->3，然後有兩個執行緒分別做如下操作：

（1）執行緒1執行pop()出棧操作，但是執行到if (top.compareAndSet(t, next)) {這行之前暫停了，所以此時節點1並未出棧；

（2）執行緒2執行pop()出棧操作彈出節點1，此時棧變為 top->2->3；

（3）執行緒2執行pop()出棧操作彈出節點2，此時棧變為 top->3；

（4）執行緒2執行push()入棧操作新增節點1，此時棧變為 top->1->3；

（5）執行緒1恢復執行，比較節點1的引用並沒有改變，執行CAS成功，此時棧變為 top->2；

What？點解變成 top->2 了？不是應該變成 top->3 嗎？

那是因為執行緒1在第一步儲存的next是節點2，所以它執行CAS成功後top節點就指向了節點2了。

測試程式碼如下：

private static void testStack() {
    // 初始化棧為 top->1->2->3
    Stack stack = new Stack();
    stack.push(new Stack.Node(3));
    stack.push(new Stack.Node(2));
    stack.push(new Stack.Node(1));

    new Thread(()->{
        // 執行緒1出棧一個元素
        stack.pop();
    }).start();

    new Thread(()->{
        // 執行緒2出棧兩個元素
        Stack.Node A = stack.pop();
        Stack.Node B = stack.pop();
        // 執行緒2又把A入棧了
        stack.push(A);
    }).start();
}

public static void main(String[] args) {
    testStack();
}

在Stack的pop()方法的if (top.compareAndSet(t, next)) {處打個斷點，執行緒1執行到這裡時阻塞它的執行，讓執行緒2執行完，再執行執行緒1這句，這句執行完可以看到棧的top物件中只有2這個節點了。

記得打斷點的時候一定要打Thread斷點，在IDEA中是右擊選擇Suspend為Thread。

通過這個例子，筆者認為你肯定很清楚ABA的危害了。

ABA的解決方法

ABA的危害我們清楚了，那麼怎麼解決ABA呢？

筆者總結了一下，大概有以下幾種方式：

（1）版本號

比如，上面的棧結構增加一個版本號用於控制，每次CAS的同時檢查版本號有沒有變過。

還有一些資料結構喜歡使用高位儲存一個郵戳來保證CAS的安全。

（2）不重複使用節點的引用

比如，上面的棧結構線上程2執行push()入棧操作的時候新建一個節點傳入，而不是複用節點1的引用；

（3）直接操作元素而不是節點

比如，上面的棧結構push()方法不應該傳入一個節點（Node），而是傳入元素值（int的value）。

好了，扯了這麼多，讓我們來看看java中的AtomicStampedReference是怎麼解決ABA的吧^^

原始碼分析

內部類

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);
    }
}

將元素值和版本號繫結在一起，儲存在Pair的reference和stamp（郵票、戳的意思）中。

屬性

private volatile Pair<V> pair;
private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();
private static final long pairOffset =
    objectFieldOffset(UNSAFE, "pair", AtomicStampedReference.class);

宣告一個Pair型別的變數並使用Unsfae獲取其偏移量，儲存到pairOffset中。

構造方法

public AtomicStampedReference(V initialRef, int initialStamp) {
    pair = Pair.of(initialRef, initialStamp);
}

構造方法需要傳入初始值及初始版本號。

compareAndSet()方法

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) ||
          // 構造新的Pair物件並CAS更新
         casPair(current, Pair.of(newReference, newStamp)));
}

private boolean casPair(Pair<V> cmp, Pair<V> val) {
    // 呼叫Unsafe的compareAndSwapObject()方法CAS更新pair的引用為新引用
    return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
}

（1）如果元素值和版本號都沒有變化，並且和新的也相同，返回true；

（2）如果元素值和版本號都沒有變化，並且和新的不完全相同，就構造一個新的Pair物件並執行CAS更新pair。

可以看到，java中的實現跟我們上面講的ABA的解決方法是一致的。

首先，使用版本號控制；

其次，不重複使用節點（Pair）的引用，每次都新建一個新的Pair來作為CAS比較的物件，而不是複用舊的；

最後，外部傳入元素值及版本號，而不是節點（Pair）的引用。

案例

讓我們來使用AtomicStampedReference解決開篇那個AtomicInteger帶來的ABA問題。

public class ABATest {

    public static void main(String[] args) {
        testStamp();
    }

    private static void testStamp() {
        AtomicStampedReference<Integer> atomicStampedReference = new AtomicStampedReference<>(1, 1);

        new Thread(()->{
            int[] stampHolder = new int[1];
            int value = atomicStampedReference.get(stampHolder);
            int stamp = stampHolder[0];
            System.out.println("thread 1 read value: " + value + ", stamp: " + stamp);

            // 阻塞1s
            LockSupport.parkNanos(1000000000L);

            if (atomicStampedReference.compareAndSet(value, 3, stamp, stamp + 1)) {
                System.out.println("thread 1 update from " + value + " to 3");
            } else {
                System.out.println("thread 1 update fail!");
            }
        }).start();

        new Thread(()->{
            int[] stampHolder = new int[1];
            int value = atomicStampedReference.get(stampHolder);
            int stamp = stampHolder[0];
            System.out.println("thread 2 read value: " + value + ", stamp: " + stamp);
            if (atomicStampedReference.compareAndSet(value, 2, stamp, stamp + 1)) {
                System.out.println("thread 2 update from " + value + " to 2");

                // do sth

                value = atomicStampedReference.get(stampHolder);
                stamp = stampHolder[0];
                System.out.println("thread 2 read value: " + value + ", stamp: " + stamp);
                if (atomicStampedReference.compareAndSet(value, 1, stamp, stamp + 1)) {
                    System.out.println("thread 2 update from " + value + " to 1");
                }
            }
        }).start();
    }
}

執行結果為：

thread 1 read value: 1, stamp: 1
thread 2 read value: 1, stamp: 1
thread 2 update from 1 to 2
thread 2 read value: 2, stamp: 2
thread 2 update from 2 to 1
thread 1 update fail!

可以看到執行緒1最後更新1到3時失敗了，因為這時版本號也變了，成功解決了ABA的問題。

總結

（1）在多執行緒環境下使用無鎖結構要注意ABA問題；

（2）ABA的解決一般使用版本號來控制，並保證資料結構使用元素值來傳遞，且每次新增元素都新建節點承載元素值；

（3）AtomicStampedReference內部使用Pair來儲存元素值及其版本號；

彩蛋

（1）java中還有哪些類可以解決ABA的問題？

AtomicMarkableReference，它不是維護一個版本號，而是維護一個boolean型別的標記，標記值有修改，瞭解一下。

（2）實際工作中遇到過ABA問題嗎？

筆者還真遇到過，以前做棋牌遊戲的時候，ABCD四個玩家，A玩家出了一張牌，然後他這個請求遲遲沒到伺服器，也就是超時了，伺服器就幫他自動出了一張牌。

然後，轉了一圈，又輪到A玩家出牌了，說巧不巧，正好這時之前那個請求到了伺服器，伺服器檢測到現在正好是A出牌，而且請求的也是出牌，就把這張牌打出去了。

然後呢，A玩家的牌就不對了。

最後，我們是通過給每個請求增加一個序列號來處理的，檢測到過期的序列號請求直接拋棄掉。

你有沒有遇到過ABA問題呢？

---

歡迎關注我的公眾號“彤哥讀原始碼”，檢視更多原始碼系列文章, 與彤哥一起暢遊原始碼的海洋。

相關推薦