前言：

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

1、 并發(fā)介紹

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

C++ 提供了適當?shù)膬?nèi)存模型（memory model）和一組原子操作（atomic operation），以支持在同一地址空間內(nèi)并發(fā)執(zhí)行多個線程。原子操作使得無鎖編程成為可能。內(nèi)存模型保證了在避免數(shù)據(jù)競爭（data races，不受控地同時訪問可變數(shù)據(jù)）的前提下，一切按照預期工作。

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

那為什么要用標準庫而非操作系統(tǒng)的并發(fā)？可移植性。

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

2、 任務和線程

如果一個計算有可能（potentially）和另一個計算并發(fā)執(zhí)行，我們稱之為任務（task）。線程是任務的系統(tǒng)級表示。任務可以通過構(gòu)造一個 std::thread 來啟動，任務作為參數(shù)。

• 任務是一個函數(shù)或者函數(shù)對象。
• 任務是一個函數(shù)或者函數(shù)對象。
• 任務是一個函數(shù)或者函數(shù)對象。

void f();              // 函數(shù)

struct F {             // 函數(shù)對象
    void operator()()  // F 的調(diào)用操作符
};

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

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

join() 確保線程完成后才退出 user() ，“join 線程”的意思是“等待線程結(jié)束”。

一個程序的線程共享同一地址空間。線程不同于進程，進程通常不直接共享數(shù)據(jù)。線程間可以通過共享對象（shared object）通信，這類通信一般用鎖或其他機制控制，以避免數(shù)據(jù)競爭。

編寫并發(fā)任務可能會非常棘手，假如上述例子中的 f 和 F 實現(xiàn)如下：

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

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

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

PaHerallllel o World!

定義一個并發(fā)程序中的任務時，我們的目標是保持任務之間完全獨立。最簡單的方法就是把并發(fā)任務看作是一個恰巧可以和調(diào)用者同時運行的函數(shù)：我們只要傳遞參數(shù)、取回結(jié)果，保證該過程中沒有使用共享數(shù)據(jù)（沒有數(shù)據(jù)競爭）即可。

3、傳遞參數(shù)

一般來說，任務需要處理一些數(shù)據(jù)。我們可以通過參數(shù)傳遞數(shù)據(jù)（或者數(shù)據(jù)的指針或引用）。

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

struct F {                 // 處理 v 的函數(shù)對象
    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) 在另一個線程中執(zhí)行
    thread t2{F{vec2}};         // F{vec2}() 在另一個線程中執(zhí)行

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

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

t1 通過 {f,ref(some_vec)} 初始化，用到了 thread 的可變參數(shù)模板構(gòu)造，可以接受任意序列的參數(shù)。ref() 是來自 <functional> 的類型函數(shù)。為了讓可變參數(shù)模板把 some_vec 當作一個引用而非對象，ref() 不能省略。編譯器檢查第一個參數(shù)可以通過其后面的參數(shù)調(diào)用，并構(gòu)建必要的函數(shù)對象，傳遞給線程。如果 F::operator()() 和 f() 執(zhí)行了相同的算法，兩個任務的處理幾乎是等同的：兩種情況下，都各自構(gòu)建了一個函數(shù)對象，讓 thread 去執(zhí)行。

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

4、返回結(jié)果

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

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

class F {
public:
    F(const vector<double>& vv, double *p) : v(vv), res(p) {}
    void operator()();  // 結(jié)果保存到 *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) 在另一個線程中執(zhí)行
    thread t2{F{vec2,&res2}};          // F{vec2,&res2}() 在另一個線程中執(zhí)行

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

這么做沒問題，也很常見。但我不覺得通過參數(shù)傳遞返回結(jié)果有多優(yōu)雅，我會在 13.7.1 節(jié)再次討論這個話題。

通過參數(shù)（出參）傳遞結(jié)果并不優(yōu)雅

5、共享數(shù)據(jù)

有時任務需要共享數(shù)據(jù)，這種情況下，對共享數(shù)據(jù)的訪問需要進行同步，同一時刻只能有一個任務訪問數(shù)據(jù)（但是多任務同時讀取不變量是沒有問題的）。我們要考慮如何保證在同一時刻最多只有一個任務能夠訪問一組對象。

解決這個問題需要通過 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;           // 操作共享數(shù)據(jù)
}   // 離開 f() 作用域，隱式自動釋放 mutex

unique_lock 的構(gòu)造函數(shù)通過調(diào)用 m.lock() 獲取 mutex。如果另一個線程已經(jīng)獲取這個 mutex，當前線程等待（阻塞）直到另一個線程（通過 m.unlock( ) ）釋放該 mutex。當 mutex 釋放，等待該 mutex 的線程恢復執(zhí)行（喚醒）?；コ?、鎖在 <mutex> 頭文件中。

共享數(shù)據(jù)和 mutex 之間的關(guān)聯(lián)需要自行約定：程序員需要知道哪個 mutex 對應哪個數(shù)據(jù)。這樣很容易出錯，但是我們可以通過一些方式使得他們之間的關(guān)聯(lián)更清晰明確：

class Record {
public:
    mutex rm;
};

不難猜到，對于一個 Record 對象 rec，在訪問 rec 其他數(shù)據(jù)之前，你應該先獲取 rec.rm。最好通過注釋或者良好的命名讓讀者清楚地知道 mutex 和數(shù)據(jù)的關(guān)聯(lián)。

有時執(zhí)行某些操作需要同時訪問多個資源，有可能導致死鎖。例如，thread1 已經(jīng)獲取了 mutex1，然后嘗試獲取 mutex2；與此同時，thread2 已經(jīng)獲取 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);                    // 嘗試獲取所有鎖
    // 操作共享數(shù)據(jù)
}   // 離開 f() 作用域，隱式自動釋放所有 mutexes

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

通過共享數(shù)據(jù)通信是相對底層的操作。編程人員要設(shè)計一套機制，弄清楚哪些任務完成了哪些工作，還有哪些未完成。從這個角度看， 使用共享數(shù)據(jù)不如直接調(diào)用函數(shù)、返回結(jié)果。另一方面，有些人認為共享數(shù)據(jù)比拷貝參數(shù)和返回值效率更高。這個觀點可能在涉及大量數(shù)據(jù)的時候成立，但是 locking 和 unlocking 也是相對耗時的操作。不僅如此，現(xiàn)代計算機很擅長拷貝數(shù)據(jù)，尤其是像 vector 這種元素連續(xù)存儲的結(jié)構(gòu)。所以，不要僅僅因為“效率”而選用共享數(shù)據(jù)進行通信，除非你真正實際測量過。

6、等待事件

有時線程需要等待外部事件，比如另一個線程完成了任務或者經(jīng)過了一段時間。最簡單的事件是時間。借助 <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 把時間單位轉(zhuǎn)成了我想要的 nanoseconds。

condition_variable 提供了對通過外部事件通信的支持，允許一個線程等待另一個線程，比如等待另一個線程（完成某個工作，然后）觸發(fā)一個事件/條件。

condition_variable 支持很多優(yōu)雅、高效的共享形式，但也可能會很棘手?？紤]一個經(jīng)典的生產(chǎn)者-消費者例子，兩個線程通過一個隊列傳遞消息：

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();              // 后續(xù)處理消息不再操作隊列 q，提前釋放 lck
        // 處理 m
    }
}

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

相應的生產(chǎn)者代碼：

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

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

7、通信任務

標準庫還在頭文件 <future> 中提供了一些機制，能夠讓程序員在更高的任務的概念層次上工作，而不是直接使用低層的線程、鎖：

• future 和 promise：用于從另一個線程中返回一個值
• packaged_task：幫助啟動任務，封裝了 future 和 promise，并且建立兩者之間的關(guān)聯(lián)
• async()：像調(diào)用一個函數(shù)那樣啟動一個任務。形式最簡單，但也最強大！

到此這篇關(guān)于C++ 對多線程/并發(fā)的支持(上)的文章就介紹到這了,更多相關(guān)C++ 對多線程并發(fā)的支持內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論