一、自旋鎖

自旋鎖是一種基礎的同步原語，用于保障對共享數據的互斥訪問。與互斥鎖的相比，在獲取鎖失敗的時候不會使得線程阻塞而是一直自旋嘗試獲取鎖。當線程等待自旋鎖的時候，CPU不能做其他事情，而是一直處于輪詢忙等的狀態。自旋鎖主要適用于被持有時間短，線程不希望在重新調度上花過多時間的情況。實際上許多其他類型的鎖在底層使用了自旋鎖實現，例如多數互斥鎖在試圖獲取鎖的時候會先自旋一小段時間，然后才會休眠。如果在持鎖時間很長的場景下使用自旋鎖，則會導致CPU在這個線程的時間片用盡之前一直消耗在無意義的忙等上，造成計算資源的浪費。

使用自旋鎖時要注意：

由于自旋時不釋放CPU，因而持有自旋鎖的線程應該盡快釋放自旋鎖，否則等待該自旋鎖的線程會一直在哪里自旋，這就會浪費CPU時間。

持有自旋鎖的線程在sleep之前應該釋放自旋鎖以便其他咸亨可以獲得該自旋鎖

二、CAS操作實現自旋鎖

CAS（Compare and Swap），即比較并替換，實現并發算法時常用到的一種技術，這種操作提供了硬件級別的原子操作（通過鎖總線的方式）。CAS操作的原型可以認為是：

bool CAS(V, A, B)

其中V代表內存中的變量，A代表期待的值，B表示新值。當V的值與A相等時，將V與B的值交換。邏輯上可以用下面的偽代碼表示：

bool CAS(V, A, B){ if (V == A) { swap(V, B); return true; }  return false;}

需要強調的是上面的操作是原子的，要么不做，要么全部完成。

那么已經擁有CAS操作的情況下如何實現一個自旋鎖呢？首先回憶自旋鎖的用途，本質上我們是希望能夠讓一個線程在不滿足進入臨界區的條件時，不停的忙等輪詢，直到可以運行的時候再繼續（進入臨界區）執行。那么，我們可能自然的想到使用一個bool變量來表示是否可以進入臨界區，例如以下面的偽代碼的邏輯：

while(flag == true);flag = true;/*do something ...*/flag = false; ...

這樣做的直觀想法是當flag為true的時候表示已經有線程處于臨界區內，只有當flag為fasle時才能進入，而在進入的時候立即將flag置為true。但是這樣做明顯存在一個問題，判斷flag為false和設置flag為true并不是一個不可分割的整體，有可能出現類似下面這樣的時序, 假設最初flag為false：

step是虛構的步驟，do something為一系列指令，這里寫在一起表示并發執行。這里可以看出由于thread1讀取判斷flag的值與修改flag的值是兩個獨立的操作，中間插入了thread2的判斷操作，最終使得有兩個線程同時進入了臨界區，這與我們的期望相悖。那么如何解決呢？如果能將讀取判斷與修改的操作合二為一，變成一個不可分割的整體，那么自然就不可能出現這種交錯的場景。對于這樣一個整體操作，我們希望它能讀取內存中變量的值，并且當其等于特定值的時候，修改它為我們需要的另一個值。嗯......沒錯，這樣我們就得到了CAS操作。

現在可以重新修改我們的同步方式，不停的進行期望flag為false的CAS操作 CAS(flag, flase, b) (這里b為true)，直到其返回成功為止，再進行臨界區中的操作，離開臨界區時將flag置為false。

b = true;while(!CAS(flag, false, b));//do somethingflag = false;

現在，判斷操作與寫入操作已經成為了一個整體，當一個線程的CAS操作成功的時候會阻止其他線程進入臨界區，到達互斥訪問的目的。

現在我們已經可以使用CAS操作來解決臨界區的互斥訪問的問題了，但是如果每次都這樣寫一遍實在太過麻煩，因此可以進行一些封裝使得使用更加方便，也就是說...可以封裝成自旋鎖。我們可以用一個類來表示，將一個bool值作為類的數據成員，同時將CAS操作和賦值操作作為其成員函數，CAS操作其實就是加鎖操作，而后面的賦值操作就是解鎖操作。

三、用C++原子量實現

按照上面的思路，接下來用 C++ 11 引入標準庫的原子量來實現一個自旋鎖并且進行測試。

首先，我們需要一個bool值來表示鎖的狀態，這里直接使用標準庫中的原子量 atomic<bool> (C++ 11的原子量可以參考：http://m.survivalescaperooms.com/article/141896.htm ，在我的平臺（Cygwin64、GCC7.3）上 atomic<bool> 的成員函數is_lock_free（）返回值為true，是無鎖的實現（如果內部使用了鎖來實現的話那還叫什么自旋鎖 = =）。實際上在大多數平臺上 atomic<bool> 都是無鎖的，如果不確定的話也可以使用C++標準規定必須為無鎖實現的atomic_flag。

接下來，我們需要兩個原子操作，CAS和賦值，C++11標準庫在原子量的成員函數中直接提供了這兩個操作。

//CASstd::atomic::compare_exchange_weak( T& expected, T desired,     std::memory_order order =     std::memory_order_seq_cst ),     std::atomic::compare_exchange_strong( T& expected, T desired,     std::memory_order order =     std::memory_order_seq_cst )//賦值void store( T desired, std::memory_order order = std::memory_order_seq_cst )

compare_exchange_weak 與 compare_exchange_strong 主要的區別在于內存中的值與expected相等的時候，CAS操作是否一定能成功，compare_exchange_weak有概率會返回失敗，而compare_exchange_strong則一定會成功。因此，compare_exchange_weak必須與循環搭配使用來保證在失敗的時候重試CAS操作。得到的好處是在某些平臺上compare_exchange_weak性能更好。按照上面的模型，我們本來就要和while搭配使用，可以使用compare_exchange_weak。最后內存序的選擇沒有特殊需求直接使用默認的std::memory_order_seq_cst。而賦值操作非常簡單直接，這個調用一定會成功（只是賦值而已 = =），沒有返回值。

實現代碼非常短，下面是源代碼：

#include <atomic>class SpinLock {public: SpinLock() : flag_(false) {} void lock() { bool expect = false; while (!flag_.compare_exchange_weak(expect, true)) {  //這里一定要將expect復原，執行失敗時expect結果是未定的  expect = false; } } void unlock() { flag_.store(false); }private: std::atomic<bool> flag_;};

如上面所說，lock操作不停的嘗試CAS操作直到成功為止，unlock操作則將bool標志位復原。使用方式如下：

SpinLock myLock;myLock.lock();//do somethingmyLock.unlock();

接下來，我們進行正確性測試，以經典的i++ 問題為例：

#include <atomic>#include <thread>#include <vector>//自旋鎖類定義class SpinLock {public: SpinLock() : flag_(false) {} void lock() { bool expect = false; while (!flag_.compare_exchange_weak(expect, true)) {  expect = false; } } void unlock() { flag_.store(false); }private: std::atomic<bool> flag_;};//每個線程自增次數const int kIncNum = 1000000;//線程數const int kWorkerNum = 10;//自增計數器int count = 0;//自旋鎖SpinLock spinLock;//每個線程的工作函數void IncCounter(){ for (int i = 0; i < kIncNum; ++i) { spinLock.lock(); count++; spinLock.unlock(); }}int main(){ std::vector<std::thread> workers; std::cout << "SpinLock inc MyTest start" << std::endl; count = 0; std::cout << "start " << kWorkerNum << " workers_" << "every worker inc " << kIncNum << std::endl; std::cout << "count_: " << count << std::endl; //創建10個工作線程進行自增操作 for (int i = 0; i < kWorkerNum; ++i) workers.push_back(std::move(std::thread(IncCounter))); for (auto it = workers.begin(); it != workers.end(); it++) it->join(); std::cout << "workers_ end" << std::endl; std::cout << "count_: " << count << std::endl; //驗證結果 if (count == kIncNum * kWorkerNum) { std::cout << "SpinLock inc MyTest passed" << std::endl; return true; } else { std::cout << "SpinLock inc MyTest failed" << std::endl; return false; } return 0;}

上面的代碼中創建了10個線程對共享的全局變量count分別進行一百萬次++操作，然后驗證結果是否正確，最終執行的輸出為：

SpinLock inc MyTest startstart 10 workers_every worker inc 1000000count_: 0workers_ endcount_: 10000000SpinLock inc MyTest passed

從結果中可以看出我們實現的自旋鎖起到了保護臨界區（這里就是i++ ）的作用，count最后的值等于每個線程執行自增的數目之和。作為對比，可以去掉IncCounter中的加鎖解鎖操作：

void IncCounter(){ for (int i = 0; i < kIncNum; ++i) { //spinLock.lock(); count++; //spinLock.unlock(); }}

執行后的輸出為：

SpinLock inc MyTest startstart 10 workers_every worker inc 1000000count_: 0workers_ endcount_: 7254522SpinLock inc MyTest failed

結果由于多個線程同時執行 i++ 造成結果錯誤。

到這里，我們就通過 C++ 11的原子量實現了一個簡單的自旋鎖。這里只是對C++原子量的一個小使用，無論是自旋鎖本身還是原子量都還有許多值得探究的地方。

總結

以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對武林網的支持。