線程間共用數據的問題

多線程之間共用數據，最大的問題便是數據競爭導致的異常問題。多個線程操作同一塊資源，如果不做任何限制，那麼一定會發生錯誤。例如：

 1 int g_nResource = 0;
 2 void thread_entry()
 3 {
 4     for (int i = 0; i < 10000000; ++i)
 5         g_nResource++;
 6 }
 7 
 8 int main()
 9 {
10     thread th1(thread_entry);
11     thread th2(thread_entry);
12     th1.join();
13     th2.join();
14     cout << g_nResource << endl;
15     return 0;
16 }

輸出：

10161838

顯然，上面的輸出結果存在問題。出現錯誤的原因可能是：

某一時刻，th1線程獲得CPU時間片，將g_nResource從100增加至200後時間片結束，保存上下文並切換至th2線程。th2將g_nResource增加至300，結束時間片，保存上下文並切換回th1線程。此時，還原上下文，g_nResource會還原成之前保存的200的值。

在併發編程中，操作由兩個或多個線程負責，它們爭先恐後執行各自的操作，而結果取決於它們執行的相對次序，每一種次序都是條件競爭。很多時候，這是良性行為，因為全部可能的結果都可以接受，即便線程變換了相對次序。例如，往容器中添加數據項，不管怎麼添加，只要容器的容量夠，總能將所有數據項填入，我們只關心是否能全部放入，對於元素的次序並不care。

真正讓人煩惱的，是惡性條件競爭。要完成一項操作，需要對共用資源進行修改，當其中一個線程還未完成數據寫入時，另一個線程不期而訪。惡性條件競爭會產生未定義的行為，並且每次產生的結果都不相同，無形中增加故障排除的難度。

歸根結底，多線程共用數據的問題大多數都由線程對數據的修改引發的。如果所有共用數據都是只讀數據，就不會有問題。因為，若數據被某個線程讀取，無論是否存在其他線程也在讀取，該數據都不會受到影響。然而，如果多個線程共用數據，只要一個線程開始改動數據，就會帶來很多隱患，產生麻煩。解決辦法就是使用互斥對數據進行保護。

1 int g_nResource = 0;
2 std::mutex _mutex;　　　　//使用互斥
3 void thread_entry()
4 {
5     _mutex.lock();　　　　//加鎖
6     for (int i = 0; i < 10000000; ++i)
7         g_nResource++;
8     _mutex.unlock();　　//解鎖
9 }

 輸出：

20000000

用互斥保護共用數據

為了達到我們想要效果，C++11引入了互斥（mutual exclusion）。互斥是一把對資源的鎖，線程訪問資源時，先鎖住與該資源相關的互斥，若其他線程試圖再給它加鎖，則須等待，直至最初成功加鎖的線程把該互斥解鎖。這確保了全部線程所見到的共用數據是自洽的（self-consistent），不變數沒有被破壞。

在C++中使用互斥

std::mutex

std::mutex是c++中最基本的互斥量。該類定義在<mutex>頭文件中。

構造函數

1 mutex();
2 
3 //不支持拷貝構造，也不支持移動構造（有定義拷貝，則無移動）
4 mutex(const mutex&) = delete;
5 mutex& operator=(const mutex&) = delete;

剛初始化的互斥處於unlocked狀態。

lock()函數

1 void lock();

用於鎖住該互斥量，有如下3中情況：

• 當前沒有被鎖，則當前線程鎖住互斥量，在未調用unlock()函數前，線程擁有該鎖。
• 被其他線程鎖住，則當前線程被阻塞，一直等待其他線程釋放鎖。
• 被當前線程鎖住，再次加鎖會產生異常。

unlock()函數

1 void unlock();

解鎖，當前線程釋放對互斥量的所有權。在無鎖情況下調用unlock()函數，將導致異常。

try_lock()函數

bool try_lock();

嘗試鎖住互斥量，如果互斥量被其他線程占用，該函數會返回false，並不會阻塞線程。有如下3中情況：

• 當前沒有被鎖，則當前線程鎖住互斥量，並返回true，在未調用unlock函數前，該線程擁有該鎖。
• 被其他線程鎖住，該函數返回false，線程並不會被阻塞。
• 被當前線程鎖住，再次嘗試獲取鎖，返回false。

案例

 1 int g_nResource = 0;
 2 std::mutex _mutex;
 3 void thread_entry()
 4 {
 5     while (1)
 6     {
 7         if (_mutex.try_lock())
 8         {
 9             cout << this_thread::get_id() << " get lock\n";
10             for (int i = 0; i < 10000000; ++i)
11                 g_nResource++;
12             _mutex.unlock();
13             return;
14         }
15         else
16         {
17             cout << this_thread::get_id() << " no get lock\n";
18             this_thread::sleep_for(std::chrono::milliseconds(500));
19         }
20     }
21 }
22 
23 int main()
24 {
25     thread th1(thread_entry);
26     thread th2(thread_entry);
27     th1.join();
28     th2.join();
29     cout << "Result = " << g_nResource << endl;
30 }

輸出：

131988 get lock
136260 no get lock
136260 get lock
Result = 20000000

上面代碼有一個缺點，就是需要我們手動調用unlock函數釋放鎖，這是一個安全隱患，並且，在某些情況下（異常），我們根本沒有機會自己手動調用unlock函數。針對上面這種情況，c++引入了lock_guard類。

std::lock_guard

std::lock_guard使用RAII手法，在對象創建時，自動調用lock函數，在對象銷毀時，自動調用unlock()函數，從而保證互斥總能被正確解鎖。該類的實現很簡單，直接貼源碼：

 1 template <class _Mutex>
 2 class _NODISCARD lock_guard { // class with destructor that unlocks a mutex
 3 public:
 4     using mutex_type = _Mutex;
 5 
 6     explicit lock_guard(_Mutex& _Mtx) : _MyMutex(_Mtx) { // construct and lock
 7         _MyMutex.lock();
 8     }
 9 
10     lock_guard(_Mutex& _Mtx, adopt_lock_t) : _MyMutex(_Mtx) {} // construct but don't lock
11 
12     ~lock_guard() noexcept {
13         _MyMutex.unlock();
14     }
15 
16     lock_guard(const lock_guard&) = delete;
17     lock_guard& operator=(const lock_guard&) = delete;
18 
19 private:
20     _Mutex& _MyMutex;
21 };

std::lock_guard僅提供了構造函數和析構函數，並未提供其他成員函數。所以，我們只能用該函數來獲取鎖、釋放鎖。

案例：

1 int g_nResource = 0;
2 std::mutex _mutex;
3 void thread_entry()
4 {
5     lock_guard<mutex> lock(_mutex);
6     for (int i = 0; i < 10000000; ++i)
7         g_nResource++;
8 }

鎖的策略標簽

std::lock_guard在構造時，可以傳入一個策略標簽，用於標識當前鎖的狀態，目前，有如下幾個標簽，含義如下：

• std::defer_lock：表示不獲取互斥的所有權
• std::try_to_lock：嘗試獲得互斥的所有權而不阻塞
• std::adopt_lock：假設調用方線程已擁有互斥的所有權

這幾個標簽可以為 std::lock_guard 、 std::unique_lock 和 std::shared_lock 指定鎖定策略。

用法如下：

1 std::lock(lhs._mutex, rhs._mutex);    //對lhs、rhs上鎖
2 std::lock_guard<mutex> lock_a(lhs._mutex, std::adopt_lock);  //不在上鎖
3 std::lock_guard<mutex> lock_b(rhs._mutex, std::adopt_lock);  //不在上鎖

組織和編排代碼以保護共用數據

使用互斥並不是萬能的，一些情況還是可能會使得共用數據遭受破壞。例如：向調用者返回指針或引用，指向受保護的共用數據，就會危及共用數據安全。或者，在類內部調用其他外部介面，而該介面需要傳遞受保護對象的引用或者指針。例如：

 1 class SomeData
 2 {
 3 public:
 4     void DoSomething() { cout << "do something\n"; }
 5 };
 6 
 7 class Operator
 8 {
 9 public:
10     void process(std::function<void(SomeData&)> func)
11     {
12         std::lock_guard<mutex> lock(_mutex);
13         func(data);     //數據外溢
14     }
15 
16 private:
17     SomeData data;
18     mutex _mutex;
19 };
20 
21 void GetDataPtr(SomeData** pPtr, SomeData& data)
22 {
23     *pPtr = &data;
24 }
25 
26 int main()
27 {
28     Operator opt;
29     SomeData* pUnprotected = nullptr;
30     auto abk = [pUnprotected](SomeData& data) mutable
31     {
32         pUnprotected = &data;
33     };
34     opt.process(abk);
35     pUnprotected->DoSomething();  //以無鎖形式訪問本應該受到保護的數據
36 }

c++並未提供任何方法解決上面問題，歸根結底這是我們代碼設計的問題，需要牢記：不得向鎖所在的作用域之外傳遞指針和引用，指向受保護的共用數據，無論是通過函數返回值將它們保存到對外可見的記憶體，還是將它們作為參數傳遞給使用者提供的函數。

發現介面固有的條件競爭

 1 void func()
 2 {
 3     stack<int> s;
 4     if (!s.empty())
 5     {
 6         int nValue = s.top();
 7         s.pop();
 8         do_something(nValue);
 9     }
10 }

在空棧上調用top()會導致未定義行為，上面的代碼已做好數據防備。對單線程而言，它既安全，又符合預期。可是，只要涉及共用，這一連串調用便不再安全。因為，在empty()和top()之間，可能有另一個線程調用pop()，彈出棧頂元素。毫無疑問，這正是典型的條件競爭。它的根本原因在於函數介面，即使在內部使用互斥保護棧容器中的元素，也無法防範。

消除返回值導致的條件競爭的方法

方法一：傳入引用接收數據

template<typename T>
class myStack
{
public:
    myStack();
    ~myStack();

    void pop(T& data);        //傳入引用接收數據

};

int main()
{
    myStack<DataRes> s;
    DataRes result;
    s.pop(result);
}

這在許多情況下行之有效，但還是有明顯短處。如果代碼要調用pop()，則須先依據棧容器中的元素類型構造一個實例，將其充當接收目標傳入函數內。對於某些類型，構建實例的時間代價高昂或耗費資源過多，所以不太實用。並且，該類型必須支持拷貝賦值運算符。

方法二：提供不拋出異常的拷貝構造函數，或不拋出異常的移動構造函數

假設某個介面是按值返回，若它拋出異常，則牽涉異常安全的問題只會在這裡出現。那麼，只要確保構造函數不會出現異常，該問題就可以解決。解決辦法是：讓該介面只允許哪些安全的類型返回。

方法三：返回指針，指向待返回元素

返回指針，指向彈出的元素，而不是返回它的值，其優點是指針可以自由地複製，不會拋出異常。可以採用std::shared_ptr托管記憶體資源。

方法四：結合方法一和方法二，或結合方法一和方法三

將上面幾種方法結合起來一起使用。

死鎖問題

線程在互斥上爭搶鎖，有兩個線程，都需要同時鎖住兩個互斥，可它們偏偏都只鎖住了一個，都在等待另一把鎖，上述情況被稱為死鎖。

防範死鎖的建議是：始終按相同順序對互斥加鎖。

 1 class A
 2 {
 3 public:
 4     A(int nValue) : m_nValue(nValue) {}
 5     friend void Swap(A& lhs, A& rhs)
 6     {
 7         if (&lhs == &rhs) return;
 8         lock_guard<mutex> lock_a(lhs._mutex);
 9         lock_guard<mutex> lock_b(rhs._mutex);
10         std::swap(lhs.m_nValue, rhs.m_nValue);
11     }
12 private:
13     int m_nValue;
14     mutex _mutex;
15 };
16 
17 void func(A& lhs, A& rhs)
18 {
19     Swap(lhs, rhs);
20 }
21 
22 int main()
23 {
24     A a1(10);
25     A a2(20);
26     thread th1(func, std::ref(a1), std::ref(a2));  //傳入參數順序不同
27     thread th2(func, std::ref(a2), std::ref(a1));  //傳入參數順序不同
28     th1.join();
29     th2.join();
30 }

上述代碼存在死鎖發生的可能。原因是在調用Swap時，加鎖順序不一致，並且，上述例子出錯更加的隱蔽，故障排除更困難。為此，c++提供了std::lock()函數。

std::lock()函數

該函數可以一次鎖住兩個或者兩個以上的互斥量。由於內部演算法的特性，它能避免因為多個線程加鎖順序不同導致死鎖的問題。用法如下：

 1 class A
 2 {
 3 public:
 4     A(int nValue) : m_nValue(nValue) {}
 5 
 6     friend void Swap(A& lhs, A& rhs)
 7     {
 8         if (&lhs == &rhs) return;
 9         std::lock(lhs._mutex, rhs._mutex);
10         std::lock_guard<mutex> lock_a(lhs._mutex, std::adopt_lock);  //已經上鎖，不再加鎖
11         std::lock_guard<mutex> lock_b(rhs._mutex, std::adopt_lock);  //已經上鎖，不再加鎖
12         std::swap(lhs.m_nValue, rhs.m_nValue);
13     }
14 
15 private:
16     int m_nValue;
17     mutex _mutex;
18 };

std::scoped_lock類

c++17提供了scoped_lock類，該類的用法和std::lock_guard類相似，也是用於托管互斥量。二者區別在於scoped_lock類可以同時托管多個互斥。例如：

1 scoped_lock<mutex, mutex> lock(lhs._mutex, rhs._mutex);

由於c++17自帶類模板參數推導，因此，上面代碼可以改寫為：

1 scoped_lock lock(lhs._mutex, rhs._mutex);

防範死鎖的補充準則

雖然死鎖最常見的誘因之一是互斥操作，但即使沒有牽涉互斥，也會發生死鎖現象。例如：有兩個線程，各自關聯了std::thread實例，若它們同時在對方的std::thread實例上調用join()，就能製造出死鎖現象卻不涉及鎖操作。如果線程甲正等待線程乙完成某一動作，同時線程乙卻在等待線程甲完成某一動作，便會構成簡單的迴圈等待。防範死鎖的準則最終可歸納成一個思想：只要另一線程有可能正在等待當前線程，那麼當前線程千萬不能反過來等待它。

準則1：避免嵌套鎖

假如已經持有鎖，就不要試圖獲取第二個鎖，若每個線程最多只持有唯一一個鎖，那麼對鎖的操作不會導致死鎖。萬一確有需要獲取多個鎖，我們應採用std::lock()函數，借單獨的調用動作一次獲取全部鎖來避免死鎖。

準則2：一旦持鎖，就須避免調用由用戶提供的程式介面

若程式介面由用戶自行實現，則我們無從得知它到底會做什麼，它可能會隨意操作，包括試圖獲取鎖。一旦我們已經持鎖，若再調用由用戶提供的程式介面，而它恰好也要獲取鎖，此時就會導致死鎖。

準則3：依次從固定順序獲取鎖

如果多個鎖是絕對必要的，卻無法通過std::lock()在一步操作中獲取全部的鎖，我們只能退而求其次，在每個線程內部都依照固定順序獲取這些鎖，並確保所有線程都遵從。

準則4：按層級加鎖

依照固定次序加鎖可能在實際中並不好執行，那麼，我們可以自己構建一個層級鎖，根據鎖的層級結構來進行加鎖。但線程已經獲取一個較低層的互斥鎖，那麼，所有高於該層的互斥鎖全部不允許加鎖。

運用std::unique_lock類靈活加鎖

std::unique_lock類同樣可以用來托管互斥量，但它比std::lock_guard類更加靈活，不一定始終占有與之關聯的互斥。

構造函數

unique_lock();
unique_lock(_Mutex&);     //構造並調用lock上鎖
~unique_lock();                //析構並調用unlock解鎖

//構造，_Mtx已經被鎖，構造函數不在調用lock
unique_lock(_Mutex&, adopt_lock_t);    

//構造，但不對_Mtx上鎖，需後續手動調用
unique_lock(_Mutex&, defer_lock_t)

//構造，嘗試獲取鎖，不會造成阻塞
unique_lock(_Mutex&, try_to_lock_t)

//構造 + try_lock_shared_for
unique_lock(_Mutex&, const chrono::duration<_Rep, _Period>&);

//構造 + try_lock_shared_until
unique_lock(_Mutex&, const chrono::time_point<_Clock, _Duration>&);

unique_lock(unique_lock&& _Other)；    //移動構造

//若占有則解鎖互斥，並取得另一者的所有權
unique_lock& operator=(unique_lock&& _Other)；

//無拷貝構造
unique_lock(const unique_lock&) = delete;
unique_lock& operator=(const unique_lock&) = delete;

構造函數提供了靈活的加鎖策略。

成員函數

//鎖定關聯互斥
void lock();

//解鎖關聯互斥
void unlock();

//嘗試鎖定關聯互斥，若互斥不可用則返回
bool try_lock();

//試圖鎖定關聯的可定時鎖定 (TimedLockable) 互斥，若互斥在給定時長中不可用則返回
bool try_lock_for(const chrono::duration<_Rep, _Period>&);

//嘗試鎖定關聯可定時鎖定 (TimedLockable) 互斥，若抵達指定時間點互斥仍不可用則返回
bool try_lock_until(const chrono::time_point<_Clock, _Duration>&);

//與另一 std::unique_lock 交換狀態
void swap(unique_lock& _Other);

//將關聯互斥解關聯而不解鎖它
 _Mutex* release();

//測試是否占有其關聯互斥
bool owns_lock();

//同owns_lock
operator bool();

//返回指向關聯互斥的指針
_Mutex* mutex();

提供了lock()、unlock()等介面，可以隨時解鎖或者上鎖。

在不同的作用域之間轉移互斥歸屬權

因為std::unique_lock實例不占有與之關聯的互斥，所以隨著其實例的轉移，互斥的歸屬權可以在多個std::unique_lock實例之間轉移。通過移動語義完成，註意區分左值和右值。

轉移有一種用途：准許函數鎖定互斥，然後把互斥的歸屬權轉移給函數調用者，好讓他在同一個鎖的保護下執行其他操作。代碼如下：

 1 std::mutex _Mtx;
 2 
 3 void PrepareData() {}
 4 
 5 void DoSomething() {}
 6 
 7 std::unique_lock<std::mutex> get_lock()
 8 {
 9     std::unique_lock<std::mutex> lock(_Mtx);
10     PrepareData();
11     return lock;
12 }
13 
14 void ProcessData()
15 {
16     std::unique_lock<std::mutex> lock(get_lock());
17     DoSomething();
18 }

按適合的粒度加鎖

“鎖粒度”該術語描述一個鎖所保護的數據量。粒度精細的鎖保護少量數據，而粒度粗大的鎖保護大量數據。鎖操作有兩個要點：一是選擇足夠粗大的鎖粒度，確保目標數據都受到保護；二是限制範圍，務求只在必要的操作過程中持鎖。只要條件允許，我們僅僅在訪問共用數據期間才鎖住互斥，讓數據處理儘可能不用鎖保護。持鎖期間應避免任何耗時的操作，如讀寫文件。這種情況可用std::unique_lock處理：假如代碼不再需要訪問共用數據，那我們就調用unlock()解鎖；若以後需重新訪問，則調用lock()加鎖。

 1 std::mutex _Mtx;
 2 bool GetAndProcessData()
 3 {
 4     std::unique_lock<std::mutex> lock(_Mtx);
 5     DataResource data = GetData();
 6     lock.unlock();
 7     bool bResult = WirteToFile(data);    //非常耗時
 8     lock.lock();
 9     SaveResult(bResult);
10     return bResult;
11 }

一般地，若要執行某項操作，那我們應該只在所需的最短時間內持鎖。換言之，除非絕對必要，否則不得在持鎖期間進行耗時的操作，如等待I/O完成或獲取另一個鎖（即便我們知道不會死鎖）。例如，在比較運算的過程中，每次只鎖住一個互斥：

 1 class Y
 2 {
 3 private:
 4     int some_detail;
 5     mutable std::mutex m;
 6     int get_detail() const
 7     {
 8         std::lock_guard<std::mutex> lock_a(m);
 9         return some_detail;
10     }
11 public:
12     Y(int sd):some_detail(sd){}
13     friend bool operator==(Y const& lhs, Y const& rhs)
14     {
15         if(&lhs==&rhs)
16             return true;
17         int const lhs_value=lhs.get_detail();    
18         int const rhs_value=rhs.get_detail();   
19         return lhs_value==rhs_value;    ⇽---  ④
20     }
21 };

為了縮短持鎖定的時間，我們一次只持有一個鎖。

保護共用數據的其他工具

互斥是保護共用數據的最普遍的方式之一，但它並非唯一方式。

在初始化過程中保護共用數據