程式師世界 >> 編程語言 >> C語言 >> C++ >> C++入門知識 >> C++11線程、鎖和條件變量(1)

C++11線程、鎖和條件變量(1)

編輯：C++入門知識

std::thread類代表了一個可執行的線程，它來自頭文件<thread>。與其它創建線程的API比如 Windows API中的CreateThread）不同的是，它可以使用普通函數、lambda函數以及仿函數實現了operator）函數的類）。另外，它還允許向線程函數傳遞任意數量的參數。

#include <thread> void func()  
{ // do some work } int main()  
{  
   std::thread t(func);  
   t.join(); return 0;  
}

在上面的例子中，t是一個線程對象，函數func()運行於該線程之中。調用join函數後，該調用線程本例中指的就是主線程）就會在join進來進行執行的線程t結束執行之前，一直處於阻塞狀態。如果該線程函數執行結束後返回了一個值，該值也將被忽略。不過，該函數可以接受任意數量的參數。

void func(int i, double d, const std::string& s)  
{  
    std::cout << i << ", " << d << ", " << s << std::endl;  
} int main()  
{  
   std::thread t(func, 1, 12.50, "sample");  
   t.join(); return 0;  
}

盡管我們可以向線程函數傳遞任意數量的參數，但是，所有的參數都是按值傳遞的。如果需要將參數按引用進行傳遞，那麼就一定要象下例所示一樣，把該參數封裝到 std::ref或者std::cref之中。

void func(int& a)  
{  
   a++;  
} int main()  
{ int a = 42;  
   std::thread t(func, std::ref(a));  
   t.join();  
   
   std::cout << a << std::endl; return 0;  
}

上面程序打印結果為43，但要不是將a封裝到std::ref之中的話，輸出的將是42。

除join方法之外，這個線程類還提供了另外幾個方法：

swap: 將兩個線程對象的底層句柄進行交換
detatch: 允許執行該方法的線程獨立於本線程對象的執行而繼續執行。脫離後的線程就再也不能執行join了你不能等待到它執行結束了）

<span style="font-family:'Courier New', Arial;font-size:9pt;line-height:1.5;">int</span><span style="font-family:'Courier New', Arial;font-size:9pt;line-height:1.5;"> main()</span> {  
    std::thread t(funct);  
    t.detach(); return 0;  
}

有一點非常重要，值得注意：線程函數中要是拋出了異常的話，使用通常的try-catch方式是捕獲不到該異常的。換句話說，下面這種做法行不通：

try {  
    std::thread t1(func);  
    std::thread t2(func);  
   
    t1.join();  
    t2.join();  
} catch(const std::exception& ex)  
{  
    std::cout << ex.what() << std::endl;  
}

要在線程間傳遞異常，你可以先在線程函數中捕獲它們，然後再將它們保存到一個合適的地方，隨後再讓另外一個線程從這個地方取得這些異常。

std::vector<std::exception_ptr>  g_exceptions; void throw_function()  
{ throw std::exception("something wrong happened");  
} void func()  
{ try {  
      throw_function();  
   } catch(...)  
   {  
      std::lock_guard<std::mutex> lock(g_mutex);  
      g_exceptions.push_back(std::current_exception());  
   }  
} int main()  
{  
   g_exceptions.clear();  
 
   std::thread t(func);  
   t.join(); for(auto& e : g_exceptions)  
   { try { if(e != nullptr)  
         {  
            std::rethrow_exception(e);  
         }  
      } catch(const std::exception& e)  
      {  
         std::cout << e.what() << std::endl;  
      }  
   } return 0;  
}

要獲得更多關於捕獲並傳遞異常的知識，你可以閱讀在主線程中處理工作線程拋出的C++異常以及怎樣才能在線程間傳遞異常？。

在深入討論之前還有一點值得注意，頭文件<thread>裡還在命名空間std::this_thread中提供了一些輔助函數：

get_id: 返回膽怯線程的id
yield: 讓調度器先運行其它的線程，這在忙於等待狀態時很有用
sleep_for: 將當前線程置於阻塞狀態，時間不少於參數所指定的時間段
sleep_util: 在指定的時刻來臨前，一直將當前的線程置於阻塞狀態

鎖

在上一個例子中，我需要對g_exceptions這個vector進行同步訪問，以確保同一個時刻只能有一個線程向其中壓入新元素。為了實現同步，我使用了一個互斥量，並在該互斥量上進行了鎖定。互斥量是一個核心的同步原語，C++11的<mutex>頭文件中包含了四種不同的互斥量。

mutex: 提供了核心的lock()函數和unlock()函數，以及非阻塞式的try_lock()方法，該方法在互斥量不可用時會立即返回。
recursive_mutex: 運行在同一線程中，多次獲得同一個互斥量。
timed_mutex: 同第一條中的mutex類似，但它還帶來了另外兩個方法try_lock_for()和try_lock_until()，分別用於在某個時間段內或在某個時刻到來之前獲得該互斥量。
recursive_timed_mutex: 結合了第二條的timed_mutex和第三條的recusive_mutex。

以下所列就是一個使用std::mutex注意其中get_id()和sleep_for()這兩個前文所述的輔助函數的用法）的例子。

#include <iostream>  
#include <thread>  
#include <mutex>  
#include <chrono>  
   
std::mutex g_lock; void func()  
{  
    g_lock.lock();  
   
    std::cout << "entered thread " << std::this_thread::get_id() << std::endl;  
    std::this_thread::sleep_for(std::chrono::seconds(rand() % 10));  
    std::cout << "leaving thread " << std::this_thread::get_id() << std::endl;  
   
    g_lock.unlock();  
} int main()  
{  
    srand((unsigned int)time(0));  
   
    std::thread t1(func);  
    std::thread t2(func);  
    std::thread t3(func);  
   
    t1.join();  
    t2.join();  
    t3.join(); return 0;  
}

其輸出將類似如下所示：

entered thread 10144 leaving thread 10144 entered thread 4188 leaving thread 4188 entered thread 3424 leaving thread 3424

lock()和unlock()這兩個方法顧名思義，頭一個方法用來對互斥量進行加鎖，如果互斥量不可得便會處於阻塞狀態；第二個方法用來對互斥量進行解鎖。

接下來的這個例子演示的是一個簡單的線程安全的容器內部使用的是std::vector）。這個容器具有添加單個元素的add()方法以及添加一批元素的addrange()方法，addrange()方法內只是簡單的調用了add()方法。

template <typename T> class container   
{  
    std::mutex _lock;  
    std::vector<T> _elements; public: void add(T element)   
    {  
        _lock.lock();  
        _elements.push_back(element);  
        _lock.unlock();  
    } void addrange(int num, ...)  
    {  
        va_list arguments;  
   
        va_start(arguments, num); for (int i = 0; i < num; i++)  
        {  
            _lock.lock();  
            add(va_arg(arguments, T));  
            _lock.unlock();  
        }  
   
        va_end(arguments);   
    } void dump()  
    {  
        _lock.lock(); for(auto e : _elements)  
            std::cout << e << std::endl;  
        _lock.unlock();  
    }  
}; void func(container<int>& cont)  
{  
    cont.addrange(3, rand(), rand(), rand());  
} int main()  
{  
    srand((unsigned int)time(0));  
   
    container<int> cont;  
   
    std::thread t1(func, std::ref(cont));  
    std::thread t2(func, std::ref(cont));  
    std::thread t3(func, std::ref(cont));  
   
    t1.join();  
    t2.join();  
    t3.join();  
   
    cont.dump(); return 0;  
}

這個程序執行起來會進入死鎖狀態。其原因在於，該容器多次嘗試獲取同一個互斥量而之前卻並沒有釋放該互斥量，這麼做是行不通的。這正是std::recursive_mutex的用武之地，它允許同一個線程多次獲得同一個互斥量，可重復獲得的最大次數並未具體說明，但一旦查過一定次數，再對lock進行調用就會拋出std::system錯誤。為了修復上面所列代碼的死鎖問題不通過修改addrange方法的實現，讓它不對lock和unlock方法進行調用)，我們可以將互斥量改為std::recursive_mutex。

template <typename T> class container   
{  
    std::recursive_mutex _lock; // ...   
};

經過修改之後，該程序的輸出會同如下所示類似：

6334 18467 41 6334 18467 41 6334 18467 41

明眼的讀者可能已經發現了，每次調用func()所產生的數字序列都完全相同。這是因為對srad的初始化是要分線程進行的，對srand()的調用只是在主線程中進行了初始化。在其它的工作線程中，srand並沒有得到初始化，所以每次產生的數字序列就是完全相同的了。

顯式的加鎖和解鎖可能會導致一定的問題，比如忘了解鎖或者加鎖的順序不對都有可能導致死鎖。本標准提供了幾個類和函數用於幫助解決這類問題。使用這些封裝類就能夠以相互一致的、RAII風格的方式使用互斥量了，它們可以在相應的代碼塊的范圍內進行自動的加鎖和解鎖動作。這些封裝類包括：

lock_guard: 該類的對象在構造之時會試圖獲得互斥量的擁有權通過調用lock()實現），而在析構之時會自動釋放它所獲得的互斥量通過調用unlock()實現）。這是一個不可復制的類。
unique_lock: 是一個通用的互斥量封裝類。與lock_quard不同，它還支持延遲加鎖、時間鎖、遞歸鎖、鎖所有權的轉移並且還支持使用條件變量。這也是一個不可復制的類，但它是可以移動的類。

使用這些封裝類，我們可以象這樣來改寫我們的容器：

template <typename T> class container   
{  
    std::recursive_mutex _lock;  
    std::vector<T> _elements; public: void add(T element)   
    {  
        std::lock_guard<std::recursive_mutex> locker(_lock);  
        _elements.push_back(element);  
    } void addrange(int num, ...)  
    {  
        va_list arguments;  
   
        va_start(arguments, num); for (int i = 0; i < num; i++)  
        {  
            std::lock_guard<std::recursive_mutex> locker(_lock);  
            add(va_arg(arguments, T));  
        }  
   
        va_end(arguments);   
    } void dump()  
    {  
        std::lock_guard<std::recursive_mutex> locker(_lock); for(auto e : _elements)  
            std::cout << e << std::endl;  
    }  
};

有人會說，既然dump()方法並不會對容器的狀態做出任何修改，所以它應該定義為congst的方法。但要是你真的這麼改了之後，編譯器就會報告出如下的錯誤：

‘std::lock_guard<_Mutex>::lock_guard(_Mutex &)' : cannot convert parameter 1 from ‘const std::recursive_mutex' to ‘std::recursive_mutex &'

互斥量無論使用的是哪一種實現）必須要獲得和釋放，這就意味著要調用非常量型的lock()和unlock()方法。所以，從邏輯上講，lock_guard不能在定義中添加const因為該方法定義為const的話，互斥量也就必需是const的了）這個問題有個解決辦法，可以讓 mutex變為mutable的。成為 mutable之後就可以在const函數中對狀態進行修改了。不過，這種用法應該只用於隱藏的或者“元”狀態比如，對計算結果或者查詢到的數據進行緩存，以供下次調用時直接使用而無需再次計算或查詢；再比如，對只是對對象的實際狀態起著輔助作用的互斥量中的位進行修改）。

template <typename T> class container   
{  
   mutable std::recursive_mutex _lock;  
   std::vector<T> _elements; public: void dump() const {  
      std::lock_guard<std::recursive_mutex> locker(_lock); for(auto e : _elements)  
         std::cout << e << std::endl;  
   }  
};

這些封裝類都具有可以接受一個用來指導加鎖策略的參數的構造器，可用的加鎖策略有：

defer_lockof typedefer_lock_t: 不要取得互斥量的擁有權
try_to_lockof typetry_to_lock_t: 在不會被阻塞的情況下嘗試獲得互斥量的擁有權
adopt_lockof typeadopt_lock_t: 假設調用線程已經獲得了互斥量的擁有權

這些策略的定義如下所示：

struct defer_lock_t { };   
struct try_to_lock_t { };   
struct adopt_lock_t { };   
constexpr std::defer_lock_t defer_lock = std::defer_lock_t();   
constexpr std::try_to_lock_t try_to_lock = std::try_to_lock_t();   
constexpr std::adopt_lock_t adopt_lock = std::adopt_lock_t();

除了這些互斥量的封裝類，本標准還提供了幾個用來對一個或多個互斥量進行加鎖的方法。

lock: 使用一種可避免死鎖的算法對互斥量進行加鎖通過調用tolock()、try_lock()以及unlock()）。
try_lock: 通過調用try_lock()i按照參數裡指定的互斥量的順序對多個互斥量進行加鎖。

這裡舉一個造成死鎖的例子：我們有一個保存元素的容器，還有一個叫做exchange()的方法，用來將一個元素從一個容器中取出來放入另外一個容器。為了成為線程安全的函數，這個函數通過獲得每個容器的互斥量，對兩個容器的訪問進行了同步處理。

template <typename T> class container   
{ public:  
    std::mutex _lock;  
    std::set<T> _elements; void add(T element)   
    {  
        _elements.insert(element);  
    } void remove(T element)   
    {  
        _elements.erase(element);  
    }  
}; void exchange(container<int>& cont1, container<int>& cont2, int value)  
{  
    cont1._lock.lock();  
    std::this_thread::sleep_for(std::chrono::seconds(1)); // <-- forces context switch to simulate the deadlock  cont2._lock.lock();      
   
    cont1.remove(value);  
    cont2.add(value);  
   
    cont1._lock.unlock();  
    cont2._lock.unlock();  
}

假設這個函數是從兩個不同的線程中進行調用的，在第一個線程中有一個元素從第一個容器中取出來，放到了第二個容器中，在第二個線程中該元素又從第二個容器中取出來放回到了第一個容器中。這樣會導致死鎖如果線程上下文正好在獲得第一個鎖的時候從一個線程切換到了另一個線程的時候就會發生死鎖）。

int main()  
{  
    srand((unsigned int)time(NULL));  
   
    container<int> cont1;   
    cont1.add(1);  
    cont1.add(2);  
    cont1.add(3);  
   
    container<int> cont2;   
    cont2.add(4);  
    cont2.add(5);  
    cont2.add(6);  
   
    std::thread t1(exchange, std::ref(cont1), std::ref(cont2), 3);  
    std::thread t2(exchange, std::ref(cont2), std::ref(cont1), 6);  
   
    t1.join();  
    t2.join(); return 0;  
}

要解決該問題，你可以使用以能夠避免死鎖的方式獲得鎖的std::lock：

void exchange(container<int>& cont1, container<int>& cont2, int value)  
{  
    std::lock(cont1._lock, cont2._lock);   
   
    cont1.remove(value);  
    cont2.add(value);  
   
    cont1._lock.unlock();  
    cont2._lock.unlock();  
}