你好，C++（83）12.3.3 利用mutex處理線程之間的共享資源競爭
想要搶先看後面的章節？打賞本文10元，即可獲得帶插圖全本下載地址！打賞完成記得私信我哦 :p
12.3.3 利用mutex處理線程之間的共享資源競爭
通過前面的學習，我們學會了利用thread創建線程來執行線程函數，學會了利用函數參數向線程函數傳遞數據，也學會了利用future和promise來獲得線程函數的結果數據，似乎我們只用了20%的技術就已經完成了多線程開發中80%的任務。可是，別高興得太早，二八原則告訴我們，用來完成那剩下的20%的任務，我們不得不用上那剩下的80%的技術。
前面我們所遇到的多線程應用場景，都只是各個線程各自獨立地訪問自己的資源，各個線程之間並沒有共享資源，也就不存在共享資源的競爭。然而在更多時候，多個線程之間往往需要共享資源，比如它們都需要訪問某個共享的容器，這時就存在一個共享資源競爭的問題。當多個線程各自獨立地同時訪問某個共享資源時，將導致未定義的結果。比如，某個線程正在向某個容器插入數據的同時，另外一個線程正在從容器中讀取數據，兩個線程將在誰先完成對容器的操作上進行競爭。其最後的結果是不可預料的，有可能讀操作失敗，也可能寫操作失敗，更有可能讀寫都失敗。這種不確定的行為對總是按照條條框框做事的程序來說是絕不容許的。為此，C++11專門提出了多種措施來處理線程之間共享資源的競爭，以保證在任何時刻都只有唯一的一個線程對共享資源進行操作，從而確保其操作行為結果的確定性和唯一性。在這些措施當中，最簡單也是最常用的就是互斥機制。互斥機制主要通過mutex類來實現。我們首先在程序中創建一個全局的mutex對象，然後通過在線程函數中先後調用這個對象的lock()成員函數和unlock()成員函數來形成一個代碼區域，通常稱為臨界區。而互斥機制保證了在任何時刻，最多只能有一個線程進入到lock()函數和unlock()函數之間的臨界區執行其中的代碼。當第一個線程正在臨界區執行時，臨界區處於鎖定狀態。如果後續有執行到lock()函數臨界區開始位置的線程將會被阻塞，進入線程隊列等待，直到第一個線程執行到unlock()函數臨界區結束離開臨界區，解除了臨界區的鎖定，其他處於線程隊列中等待的線程才會按照先進先出（FIFO，First In，First Out）的規則進入臨界區執行，而一旦有線程進入臨界區，臨界區又會被再次鎖定。其他未進入臨界區的線程只有在線程隊列中繼續等待，直到它的機會到來。由於互斥對象所確定的臨界區每次只能有一個線程進入執行，如果我們將對共享資源的訪問放到臨界區來進行，這樣就能保證每次只有一個線程在臨界區對共享資源進行訪問，也就避免了共享資源被多個線程同時訪問的問題。例如，在前面的餐館中有兩個廚子，他們都會將炒好的菜交給一個服務員讓她端給客人，這裡，服務員就成了共享資源，為了避免兩個廚子同時讓服務員端菜，我們用mutex對象為她設立一個臨界區，讓她在臨界區工作：
#include <mutex> // 引入mutex所在的頭文件
#include <queue> // 引入queue容器所在的頭文件
// 全局的互斥對象
mutex m;
// 全局的queue容器對象quFoods
// 線程函數會將炒好的菜push()到quFoods容器，所以它表示服務員
queue<Food> quFoods;
// 線程函數，創建臨界區訪問共享資源quFoods
void Cook(string strName)
{
// 炒菜…
// 這些不涉及共享資源的動作是可以放在臨界區之外多個線程同時進行的
Food food(strName);
m.lock(); // 臨界區開始
// 對共享資源的操作
quFoods.push(food);// 將food對象添加到共享的容器中
m.unlock(); // 臨界區結束
}
int main()
{
thread coWang(Cook,"回鍋肉"); // 王廚子炒回鍋肉
thread coChen(Cook,"鹽煎肉"); // 陳廚子炒鹽煎肉
// 等待廚子炒完菜…
coWang.join();
coChen.join();
// 輸出結果
cout<<"兩位廚子炒出了"<<endl;
// 輸出quFoods容器中所有Food對象的名字
// 這裡只有主線程會執行，所以對共享資源的訪問不需要放在臨界區
while(0 != quFoods.size() )
{
cout<<quFoods.front().GetName()<<endl;
quFoods.pop(); // 從容器中彈出最先進入隊列的Food對象
}
return 0;
}
在這裡，我們首先創建了一個全局的mutex對象m以及一個共享資源quFoods容器，然後在線程函數Cook()中，我們用m的lock()函數和unlock()函數形成了一個臨界區。因為Food對象的創建不涉及共享資源，各個線程可以各自獨立地進行，所以我們把Food對象的創建工作放在臨界區之外進行。當Food對象創建完成需要添加到quFoods容器時，就涉及到了對共享資源quFoods的操作，就需要放到臨界區來進行以保證任何時刻只有唯一的線程對quFoods進行操作。當多個線程在執行線程函數Cook()時，其流程如下圖12-6所示：
圖12-6 臨界區的執行流程
從這裡我們可以看到，互斥對象mutex的使用非常簡單。我們只需要創建一個全局的mutex對象，然後利用其lock()函數和unlock()函數劃定一個臨界區，同時將那些對共享資源的訪問放到臨界區就可以保證同一時刻只有唯一的一個線程對共享資源進行訪問了。互斥對象的使用是一件簡單的事情，但是用好互斥對象卻沒那麼簡單。如果我們錯誤地使用了互斥，比如，一個線程只執行了lock()函數鎖定了臨界區，但因為某種原因（出現異常或者是長時間被阻塞）而沒有執行相應的unlock()函數解除臨界區的鎖定，那麼其他線程將再也無法進入臨界區，從而整個程序都會被阻塞而失去響應。所以，我們在使用mutex對象的時候必須小心謹慎，要不然就很容易造成線程死鎖而給程序帶來災難性的後果。同時，這種錯誤也難以發現，程序員往往因此而陷入水深火熱之中。
為了挽救程序員於水火，C++11提供了多種措施來避免這種災難的發生。其中最簡單的是mutex對象的try_lock()函數，利用它，我們可以在鎖定臨界區之前進行一定的嘗試，如果當前臨界區可以鎖定，則鎖定臨界區而進入臨界區執行。如果當前臨界區已經被其他線程鎖定而無法再次鎖定，則可以採取一定的措施來避免線程一直等待而形成線程的死鎖。例如：
void Cook(string strName)
{
Food food(strName);
// 嘗試鎖定臨界區
if(m.try_lock())
{
quFoods.push(food);
m.unlock(); // 解除鎖定
}
else
{
// 在無法鎖定臨界區時採取的補救措施
cout<<"服務員這會兒太忙了，炒好的菜先放放"<<endl;
}
}
除了try_lock()函數，C++11還提供了recursive_mutex互斥對象，它可以讓同一線程多次進入某一臨界區，從而巧妙地解決了函數遞歸調用中對同一個互斥對象多次執行lock()操作的問題。除此之外，C++11還提供了timed_mutex互斥對象，利用它的try_lock_for()函數和try_lock_until()函數，可以讓線程只是在某段時間之內或某個固定時間點之內嘗試鎖定，這樣就對線程等待鎖定臨界區的時間做了限制，從而避免了線程在異常情況下無法鎖定臨界區時還長時間地等待，也就有機會採取措施解決問題。與timed_mutex相對應地，C++11中也有recursive_timed_mutex對象，其使用方法與前兩者類似。這些輔助性的互斥對象各有各的用途，根據我們的應用場景而選擇合適的互斥對象，是利用互斥對象解決共享資源競爭問題的前提。
藉助於C++11所提供的這些輔助性策略，我們可以很好地避免線程因長時間等待進入臨界區而形成線程死鎖。但是，這些方法都只是亡羊補牢的方法，讓我們在臨界區無法鎖定的情況下有機會採取措施解決問題。而且，臨界區無法鎖定的情況往往是程序員自己造成的，很多時候，我們只是調用lock()函數鎖定了臨界區，但忘了調用unlock()函數來解除臨界區的鎖定，或者是因為程序邏輯的問題而跳過了unlock()函數的調用，最終導致lock()和unlock()的不匹配，才造成了臨界區無法鎖定的情況。《黃帝內經》上說，聖人「不治已病治未病」，與其在出現臨界區無法鎖定的情況下採取補救措施解決問題，不如事先就管理好互斥對象，讓它的lock()和unlock()完全配對，也就不會出現臨界區無法鎖定的情況了。為此，C++11的標準庫中專門地提供了鎖（lock）對象用於互斥對象的管理，其中最簡單也最常用的就是lock_guard類。
lock_guard類實際上是一個類模板，我們只需要在合適的地方（需要mutex互斥對象鎖定的地方），首先使用需要管理的互斥對象類型（比如，mutex或timed_mutex等）作為其類型參數特化這個類模板而得到一個特定類型的模板類，然後使用一個互斥對象作為其構造函數參數而創建一個lock對象，從此這個lock對象與這個互斥對象建立聯繫，lock對象開始對互斥對象進行管理。當以互斥對象為參數創建lock對象的時候，它的構造函數會自動調用互斥對象的lock()函數，鎖定它所管理的互斥對象。而當代碼執行離開lock對象的作用域時，作為局部變數的lock對象會被自動釋放，而它的析構函數則會自動調用互斥對象的unlock()函數，自動解除它所管理的互斥對象的鎖定。通過lock對象的幫助，藉助其構造函數和析構函數的嚴格匹配，互斥對象的鎖定（lock()）和解鎖（unlock()）也同樣做到了自動地嚴格匹配，從此讓程序員們再也不用為錯過調用互斥對象的unlock()函數造成線程死鎖而頭疼了。利用lock對象，我們可以將上一小節的例子改寫為：
// 使用lock_guard來管理mutex對象
void Cook(string strName)
{
Food food(strName);
// 使用需要管理的mutex對象作為構造函數創建lock對象
// 構造函數會調用mutex對象的lock()函數鎖定臨界區
lock_guard<mutex> lock(m);
// m.lock(); // 不用直接調用mutex對象lock()函數
// 對共享資源的訪問
quFoods.push(food);
// lock對象被釋放，其析構函數被自動調用，
// 在其析構函數中，會調用mutex對象的unlock()函數解除臨界區的鎖定
// m.unlock();
}
在這裡，我們創建了一個lock_guard對象lock對mutex對象m進行管理，當lock對象被創建的時候，其構造函數會調用m的lock()函數鎖定臨界區，而當線程函數執行完畢後，作為局部變數的lock對象會被自動釋放，其析構函數也會被自動調用，而在其析構函數中，m對象的unlock()函數會被調用，從而隨著lock對象的析構自動地解除了臨界區的鎖定。這樣，利用局部變數lock對象構造函數和析構函數相互匹配的特性，就自動完成了mutex對象的lock()與unlock()的匹配，很大程度上避免了因lock()和unlock()無法匹配而形成的線程死鎖問題。
除了只提供構造函數和析構函數的lock_guard類之外，C++11標準庫還提供了擁有其他成員函數的unique_lock類，從而讓我們可以對鎖對象進行更多的控制。比如，我們可以利用它的try_lock()函數嘗試鎖定它所管理的互斥對象，也可以用try_lock_for()函數在某一段時間內嘗試鎖定等等，其使用方法與互斥對象相似。
另外需要注意的是，無論是互斥對象還是鎖對象，某種意義上它們都代表著某種共享資源的所有權，它們往往是一一對應的。另外對於互斥對象而言，只有當它對至少兩個線程可見時，它才是有意義的，所以它往往是全局的。而對於鎖對象，我們需要利用它的構造函數和析構函數來完成臨界區的鎖定與解鎖，所以它往往是局部的。因為共享資源的唯一性，也同樣決定了互斥對象和鎖對象的唯一性。所以它們都是不可以被複制的，因為共享資源只有一個，如果它們被複制，我們就無法確定到底哪一個副本應該擁有這唯一的共享資源，從而造成所屬關係上的混亂。但是，它們是可以被移動的，也就是相當於我們將這個資源的所有權從一個局部環境轉移或共享到了另外一個局部環境。
到這裡，我們學習了C++11中關於多線程開發的大部分技術，我們知道了如何利用thread對象創建線程執行某個線程函數，知道了如何利用future和promise對象在線程之間傳遞數據，也知道了如何利用mutex對象來管理線程之間共享資源的訪問競爭。可以說，利用C++11所提供的這些技術創建一個多線程的程序是十分容易的，也可以再次吃到那份美味的免費午餐。但是，因為多線程程序在邏輯上的複雜性，搞不好就會出現線程死鎖等嚴重影響性能的問題，所以要想把多線程程序做好，卻卻又沒有那麼容易。C++11所提供的這些支持多線程開發的技術，就像一把鋒利的刀子，用好了削鐵如泥，沒用好也容易傷到自己。所以我們唯一的辦法就是不斷地去實踐，從實踐中積累經驗。當我們手上的傷疤足夠多的時候，我們自然也就能把手上的鋒利刀子運用自如了。
知道更多：OpenMP——thread對象之外的另一種選擇
程序員大約是這個世界上最懶的一類人了。雖然利用thread對象可以輕鬆簡便地將一個程序並行化，可程序員們仍不滿足。他們覺得，雖然有了thread對象，創建線程是簡單了很多，可是依然需要他們去創建thread對象，依然需要他們去利用mutex對象處理線程之間共享資源的競爭，在將一些單線程程序改寫為多線程程序時，甚至還需要他們對演算法進行重新設計。他們在想，有沒有一種方法可以自動將一個單線程程序並行化而無需程序員做什麼額外的工作？
這個世界的發展一定是被懶人推動的。正是因為程序員們有這種懶人想法，才有了OpemMP（open multi-processing）這種懶人多線程方案。它是一套支持跨平台的、用於共享內存並行系統的多線程程序設計的編譯指令、函數和一些能夠影響運行行為的環境變數，目前已經受到主流編譯器的支持。OpenMP提供對並行演算法的高層抽象的描述，程序員只需要通過在原始的串列代碼中加入專用的編譯指令來指明他們的意圖，編譯器就會根據這些指令自動地創建線程、分配線程任務、處理共享資源競爭，從而幾乎是全自動地完成程序的並行化。當選擇忽略代碼中的這些編譯指令時，或者是編譯器不支持OpenMP時，程序又可退化為原始的串列代碼。這樣就做到了進可攻退可守，極大地增加了代碼的靈活性。
要想在程序中使用OpenMP非常簡單，只需要在編譯器選項中啟用對OpenMP的支持（例如，gcc編譯器使用-fopenmp，Visual C++編譯器使用/openmp），並在代碼中引入OpenMP的頭文件，然後就可以在原始代碼中那些可以並行執行的代碼（比如，某個for循環，某個對數組的操作等）前加入相應的OpenMP編譯指令來將程序並行化。下面看一個簡單的例子：// 引入OpenMP的頭文件
#include <omp.h>
using namespace std;
void foo()
{
// …
}
int main()
{
// 用pragma指令指明這是一個可以並行執行的for循環
// 編譯器會根據這些指令自動創建多個線程，
// 對for循環進行相應的並行處理
#pragma omp parallel for
for (int i = 0; i < 100; ++i)
foo();
return 0;
}
在這裡，我們只是簡單地用一個pragma編譯指令告訴編譯器接下來的for循環是一個可以並行處理的for循環，編譯器就會根據程序員的這個意圖表達自動地創建多個線程並行地執行這個for循環。在整個過程中，程序員只需要使用編譯指令告訴編譯器「嘿，下面這個for循環需要並行執行」，然後編譯器就會自動為我們創建線程來完成for循環的並行執行，根本不用我們操心。想幹啥就有人去幫你干，這恐怕是懶人的最高境界了。
推薦閱讀：
