隊(duì)列是我們非常常用的數(shù)據(jù)結(jié)構(gòu)，用來(lái)提供數(shù)據(jù)的寫入和讀取功能，而且通常在不同線程之間作為數(shù)據(jù)通信的橋梁。不過(guò)在將無(wú)鎖隊(duì)列的算法之前，需要先了解一下CAS（compare and swap）的原理。由于多個(gè)線程同時(shí)操作同一個(gè)數(shù)據(jù)，其中肯定是存在競(jìng)爭(zhēng)的，那么如何能夠針對(duì)同一個(gè)數(shù)據(jù)進(jìn)行操作，而且又不用加鎖呢？ 這個(gè)就需要從底層，CPU層面支持原子修改操作，比如在X86的計(jì)算機(jī)平臺(tái)，提供了XCHG指令，能夠原子的交互數(shù)值。

　　從開發(fā)語(yǔ)言的層面，比如C++11中，就提供了atomic_compare_exchange_weak函數(shù)，來(lái)實(shí)現(xiàn)CAS。

　　1. lockless queue，enqueue，dequeue操作的算法

　　（1） enqueue算法：

　　

　　enqueue時(shí)，先將需要加入隊(duì)尾的數(shù)據(jù)創(chuàng)建出來(lái)，然后在一個(gè)循環(huán)操作中，將數(shù)據(jù)加入隊(duì)尾，如果加入失敗，那么就更新當(dāng)前的隊(duì)尾指針，直到加入成功，然后循環(huán)結(jié)束。最后調(diào)整隊(duì)尾指針。

　　（2） dequeue算法

　　

　　dequeue時(shí)，在循環(huán)操作中，使用CAS將隊(duì)列頭指針，變成頭指針的下一個(gè)指針，如果失敗，持續(xù)操作直到成功。最后返回頭指針指向的值。

　　2. ABA問(wèn)題，及解決辦法

　 從上面的算法中，可以看出采用CAS方式實(shí)現(xiàn)的無(wú)鎖隊(duì)列的算法過(guò)程，不過(guò)由于CAS操作本身的特殊性（通過(guò)判斷當(dāng)前被變換的值，是否發(fā)生過(guò)變化），可能會(huì)在某些情況下引起ABA問(wèn)題。

　　那么首先什么是ABA問(wèn)題呢？ wiki上有這樣一個(gè)說(shuō)明的例子：

　　Natalie is waiting in her car at a red traffic light with her children. Her children start fighting with each other while waiting, and she leans back to scold them. Once their fighting stops, Natalie checks the light again and notices that it's still red. However, while she was focusing on her children, the light had changed to green, and then back again. Natalie doesn't think the light ever changed, but the people waiting behind her are very mad and honking their horns now.

　　意思就是說(shuō)，Natalie在等紅燈的時(shí)候，由于回頭管孩子，錯(cuò)過(guò)了綠燈，等她再回過(guò)頭看信號(hào)燈的時(shí)候，又是紅燈了。

　　這其實(shí)就是一個(gè)ABA問(wèn)題，雖然中間信號(hào)燈發(fā)生了變化，但是Natalie卻不知道。

　　

　　用C++中的一個(gè)stack來(lái)說(shuō)明，stack代碼如下：

/* Naive lock-free stack which suffers from ABA problem.*/
  class Stack {
    std::atomic<Obj*> top_ptr;
    //
    // Pops the top object and returns a pointer to it.
    //
    Obj* Pop() {
      while(1) {
        Obj* ret_ptr = top_ptr;
        if (!ret_ptr) return std::nullptr;
        // For simplicity, suppose that we can ensure that this dereference is safe
        // (i.e., that no other thread has popped the stack in the meantime).
        Obj* next_ptr = ret_ptr->next;
        // If the top node is still ret, then assume no one has changed the stack.
        // (That statement is not always true because of the ABA problem)
        // Atomically replace top with next.
        if (top_ptr.compare_exchange_weak(ret_ptr, next_ptr)) {
          return ret_ptr;
        }
        // The stack has changed, start over.
      }
    }
    //
    // Pushes the object specified by obj_ptr to stack.
    //
    void Push(Obj* obj_ptr) {
      while(1) {
        Obj* next_ptr = top_ptr;
        obj_ptr->next = next_ptr;
        // If the top node is still next, then assume no one has changed the stack.
        // (That statement is not always true because of the ABA problem)
        // Atomically replace top with obj.
        if (top_ptr.compare_exchange_weak(next_ptr, obj_ptr)) {
          return;
        }
        // The stack has changed, start over.
      }
    }
  };

　　假設(shè)，stack初始化為top→ A → B → C

　　線程1先執(zhí)行　

ret = A;
next = B;

　　然后在線程1執(zhí)行compare_exchange_weak之前被中斷，換成線程2執(zhí)行。

{ // 線程2先pop出A
    ret = A;
    next = B;
    compare_exchange_weak(A, B)  // Success, top = B
    return A;
  } // Now the stack is top → B → C
  { // 線程2再pop出B
    ret = B;
    next = C;
    compare_exchange_weak(B, C)  // Success, top = C
    return B;
  } // Now the stack is top → C
  delete B;
  { // 最后線程2將A再push進(jìn)stack
    A->next = C;
    compare_exchange_weak(C, A)  // Success, top = A
  }

　　當(dāng)線程2執(zhí)行完所有這些操作之后，換成線程1執(zhí)行，線程1的compare_exchange_weak是會(huì)成功執(zhí)行的，因?yàn)樗恢纓op_ptr已經(jīng)被修改過(guò)了。

　　通常針對(duì)ABA問(wèn)題的解決辦法，就是針對(duì)操作的數(shù)據(jù)加上一個(gè)原子操作的使用計(jì)數(shù)，在CAS執(zhí)行之前，先獲取一下計(jì)數(shù)是否和之前一樣，如果不一樣，就說(shuō)明數(shù)據(jù)已經(jīng)被修改過(guò)了。

總結(jié)

以上是生活随笔為你收集整理的无锁队列以及ABA问题的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。