本文翻譯自 C++ 之父 Bjarne Stroustrup 的 C++ 之旅（A Tour of C++）一書的第 13 章 Concurrency，作者用短短數十頁，帶你一窺現代 C++ 對并發/多執行緒的支持，原文地址：現代 C++ 對多執行緒/并發的支持（上） -- 節選自 C++ 之父的《A Tour of C++》水平有限，有條件的建議直接閱讀原版書籍，

13 并發

13 并發
- 13.1 介紹
- 13.2 任務和執行緒
- 13.3 傳遞引數
- 13.4 回傳結果
- 13.5 共享資料
- 13.6 等待事件
- 13.7 通信任務

13.1 介紹

并發，即同時執行多個任務，常用來提高吞吐量（通過利用多處理器進行同一個計算）或者改善回應性（等待回復的時候，允許程式的其他部分繼續執行），所有現代語言都支持并發，C++ 標準庫提供了可移植、型別安全的并發支持，經過 20 多年的發展，幾乎被所有現代硬體所支持，標準庫提供的主要是系統級的并發支持，而非復雜的、更高層次的并發模型；其他庫可以基于標準庫，提供更高級別的并發支持，

C++ 提供了適當的記憶體模型（memory model）和一組原子操作（atomic operation），以支持在同一地址空間內并發執行多個執行緒，原子操作使得無鎖編程成為可能，記憶體模型保證了在避免資料競爭（data races，不受控地同時訪問可變資料）的前提下，一切按照預期作業，

本章將給出標準庫對并發的主要支持示例：thread、mutex、lock()、packaged_task 以及 future，這些特征直接基于作業系統構建，相較于作業系統原生支持，不會帶來性能損失，也不保證會有顯著的性能提升，

那為什么要用標準庫而非作業系統的并發？可移植性，

不要把并發當作靈丹妙藥：如果順序執行可以搞定，通常順序會比并發更簡單、更快速！

13.2 任務和執行緒

如果一個計算有可能（potentially）和另一個計算并發執行，我們稱之為任務（task），執行緒是任務的系統級表示，任務可以通過構造一個 std::thread 來啟動，任務作為引數，

任務是一個函式或者函式物件，
任務是一個函式或者函式物件，
任務是一個函式或者函式物件，

void f();              // 函式

struct F {             // 函式物件
    void operator()()  // F 的呼叫運算子
};

void user()
{
    thread t1 {f};     // f() 在另一個執行緒中執行
    thread t2 {F()};   // F()() 在另一個執行緒中執行

    t1.join();  // 等待 t1
    t2.join();  // 等待 t2
}

join() 確保執行緒完成后才退出 user()，“join 執行緒”的意思是“等待執行緒結束”，

一個程式的執行緒共享同一地址空間，執行緒不同于行程，行程通常不直接共享資料，執行緒間可以通過共享物件（shared object）通信，這類通信一般用鎖或其他機制控制，以避免資料競爭，

撰寫并發任務可能會非常棘手，假如上述例子中的 f 和 F 實作如下：

void f() {cout << "Hello ";}

struct F {
    void operator()() {cout << "Parallel World!\n";}
};

這里有個嚴重的錯誤：f 和 F() 都用到了 cout 物件，卻沒有任何形式的同步，這會導致輸出的結果不可預測，多次執行的結果可能會得到不同的結果：因為兩個任務的執行順序是未定義的，程式可能產生詭異的輸出，比如：

PaHerallllel o World!

定義一個并發程式中的任務時，我們的目標是保持任務之間完全獨立，最簡單的方法就是把并發任務看作是一個恰巧可以和呼叫者同時運行的函式：我們只要傳遞引數、取回結果，保證該程序中沒有使用共享資料（沒有資料競爭）即可，

13.3 傳遞引數

一般來說，任務需要處理一些資料，我們可以通過引數傳遞資料（或者資料的指標或參考），

void f(vector<double>& v); // 處理 v 的函式

struct F {                 // 處理 v 的函式物件
    vector<double>& v;
    F(vector<double>& vv) : v(vv) {}
    void operator()();
};

int main()
{
    vector<double> some_vec{1,2,3,4,5,6,7,8,9};
    vector<double> vec2{10,11,12,13,14};

    thread t1{f,ref(some_vec)}; // f(some_vec) 在另一個執行緒中執行
    thread t2{F{vec2}};         // F{vec2}() 在另一個執行緒中執行

    t1.join();
    t2.join();
}

F{vec2} 在 F 中保存了引數 vector 的參考，F 現在可以使用這個 vector，但愿在 F 執行時，沒有其他任務訪問 vec2，如果通過值傳遞 vec2 則可以消除這個隱患，

t1 通過 {f,ref(some_vec)} 初始化，用到了 thread 的可變引數模板構造，可以接受任意序列的引數，ref() 是來自 <functional> 的型別函式，為了讓可變引數模板把 some_vec 當作一個參考而非物件，ref() 不能省略，編譯器檢查第一個引數可以通過其后面的引數呼叫，并構建必要的函式物件，傳遞給執行緒，如果 F::operator()() 和 f() 執行了相同的演算法，兩個任務的處理幾乎是等同的：兩種情況下，都各自構建了一個函式物件，讓 thread 去執行，

可變引數模板需要用 ref()、cref() 傳遞參考

13.4 回傳結果

13.3 的例子中，我傳了一個非 const 的參考，只有在希望任務修改參考資料時我才這么做，這是一種很常見的獲取回傳結果的方式，但這么做并不能清晰、明確地向他人傳達你的意圖，稍好一點的方式是通過 const 參考傳遞輸入資料，通過另外單獨的引數傳遞儲存結果的指標，

void f(const vector<double>& v, double *res); // 從 v 獲取輸入; 結果存入 *res

class F {
public:
    F(const vector<double>& vv, double *p) : v(vv), res(p) {}
    void operator()();  // 結果保存到 *res

private:
    const vector<double>& v;  // 輸入源
    double *p;                // 輸出地址
};

int main()
{
    vector<double> some_vec;
    vector<double> vec2;

    double res1;
    double res2;

    thread t1{f,cref(some_vec),&res1}; // f(some_vec,&res1) 在另一個執行緒中執行
    thread t2{F{vec2,&res2}};          // F{vec2,&res2}() 在另一個執行緒中執行

    t1.join();
    t2.join();
}

這么做沒問題，也很常見，但我不覺得通過引數傳遞回傳結果有多優雅，我會在 13.7.1 節再次討論這個話題，

通過引數（出參）傳遞結果并不優雅

13.5 共享資料

有時任務需要共享資料，這種情況下，對共享資料的訪問需要進行同步，同一時刻只能有一個任務訪問資料（但是多任務同時讀取不變數是沒有問題的），我們要考慮如何保證在同一時刻最多只有一個任務能夠訪問一組物件，

解決這個問題需要通過 mutex（mutual exclusion object，互斥物件），thread 通過 lock() 獲取 mutex：

int shared_data;
mutex m;          // 用于控制 shared_data 的 mutex

void f()
{
    unique_lock<mutex> lck{m};  // 獲取 mutex
    shared_data += 7;           // 操作共享資料
}   // 離開 f() 作用域，隱式自動釋放 mutex

unique_lock 的建構式通過呼叫 m.lock() 獲取 mutex，如果另一個執行緒已經獲取這個 mutex，當前執行緒等待（阻塞）直到另一個執行緒（通過 m.unlock()）釋放該 mutex，當 mutex 釋放，等待該 mutex 的執行緒恢復執行（喚醒），互斥、鎖在 <mutex> 頭檔案中，

共享資料和 mutex 之間的關聯需要自行約定：程式員需要知道哪個 mutex 對應哪個資料，這樣很容易出錯，但是我們可以通過一些方式使得他們之間的關聯更清晰明確：

class Record {
public:
    mutex rm;
};

不難猜到，對于一個 Record 物件 rec，在訪問 rec 其他資料之前，你應該先獲取 rec.rm，最好通過注釋或者良好的命名讓讀者清楚地知道 mutex 和資料的關聯，

有時執行某些操作需要同時訪問多個資源，有可能導致死鎖，例如，thread1 已經獲取了 mutex1，然后嘗試獲取 mutex2；與此同時，thread2 已經獲取 mutex2，嘗試獲取 mutex1，在這種情況下，兩個任務都無法進行下去，為解決這一問題，標準庫支持同時獲取多個鎖：

void f()
{
    unique_lock<mutex> lck1{m1,defer_lock};  // defer_lock：不立即獲取 mutex
    unique_lock<mutex> lck2{m2,defer_lock};
    unique_lock<mutex> lck3{m3,defer_lock};

    lock(lck1,lck2,lck3);                    // 嘗試獲取所有鎖
    // 操作共享資料
}   // 離開 f() 作用域，隱式自動釋放所有 mutexes

lock() 只有在獲取引數里所有的 mutex 之后才會繼續執行，并且在其持有 mutex 期間，不會阻塞（go to sleep），每個 unique_lock 的析構會確保離開作用域時，自動釋放所有的 mutex，

通過共享資料通信是相對底層的操作，編程人員要設計一套機制，弄清楚哪些任務完成了哪些作業，還有哪些未完成，從這個角度看，使用共享資料不如直接呼叫函式、回傳結果，另一方面，有些人認為共享資料比拷貝引數和回傳值效率更高，這個觀點可能在涉及大量資料的時候成立，但是 locking 和 unlocking 也是相對耗時的操作，不僅如此，現代計算機很擅長拷貝資料，尤其是像 vector 這種元素連續存盤的結構，所以，不要僅僅因為“效率”而選用共享資料進行通信，除非你真正實際測量過，

13.6 等待事件

有時執行緒需要等待外部事件，比如另一個執行緒完成了任務或者經過了一段時間，最簡單的事件是時間，借助 <chrono>，可以寫出：

using namespace std::chrono;

auto t0 = high_resolution_clock::now();
this_thread::sleep_for(milliseconds{20});
auto t1 = high_resolution_clock::now();

cout << duration_cast<nanoseconds>(t1-t0).count() << " nanoseconds passed\n";

注意，我甚至沒有啟動一個執行緒；默認情況下，this_thread 指當前唯一的執行緒，我用 duration_cast 把時間單位轉成了我想要的 nanoseconds，

condition_variable 提供了對通過外部事件通信的支持，允許一個執行緒等待另一個執行緒，比如等待另一個執行緒（完成某個作業，然后）觸發一個事件/條件，

condition_variable 支持很多優雅、高效的共享形式，但也可能會很棘手，考慮一個經典的生產者-消費者例子，兩個執行緒通過一個佇列傳遞訊息：

class Message { /**/ }; // 通信的物件

queue<Message> q;       // 訊息佇列
condition_variable cv;  // 傳遞事件的變數
mutex m;                // locking 機制

queue、condition_variable 以及 mutex 由標準庫提供，

消費者讀取并處理 Message

void consumer()
{
    while(true){
        unique_lock<mutex> lck{m}; // 獲取 mutex m
        cv.wait(lck);              // 先釋放 lck，等待事件/條件喚醒
                                   // 喚醒時再次重新獲得 lck
        auto m = q.front();        // 從佇列中取出 Message m
        q.pop();
        lck.unlock();              // 后續處理訊息不再操作佇列 q，提前釋放 lck
        // 處理 m
    }
}

這里我顯式地用 unique_lock<mutex> 保護 queue 和 condition_variable 上的操作，condition_variable 上的 cv.wait(lck) 會釋放引數中的鎖 lck，直到等待結束（佇列非空），然后再次獲取 lck，

相應的生產者代碼：

void producer()
{
    while(true) {
        Message m;
        // 填充 m
        unique_lock<mutex> lck{m}; // 保護操作
        q.push(m);
        cv.notify_one();           // 通知/喚醒等待中的 condition_variable
    } // 作用域結束自動釋放鎖
}

到目前為止，不論是 thread、mutex、lock 還是 condition_variable，都還是低層次的抽象，接下來我們馬上就能看到 C++ 對并發的高級抽象支持，

13.7 通信任務

標準庫還在頭檔案 <future> 中提供了一些機制，能夠讓程式員在更高的任務的概念層次上作業，而不是直接使用低層的執行緒、鎖：

future 和 promise：用于從另一個執行緒中回傳一個值
packaged_task：幫助啟動任務，封裝了 future 和 promise，并且建立兩者之間的關聯
async()：像呼叫一個函式那樣啟動一個任務，形式最簡單，但也最強大！

篇幅有點長，先到這里，13.7 節的具體內容及余下章節將在下一篇中展開：現代 C++ 對多執行緒/并發的支持（下）

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/308236.html

標籤：其他

上一篇：C語言檔案讀寫 ferror 函式 - C語言零基礎入門教程

下一篇：Go語言之Goroutine與信道、例外處理

現代 C++ 對多執行緒/并發的支持（上） -- 節選自 C++ 之父的 《A Tour of C++》