1. 并行基礎(chǔ)

std::thread 用于創(chuàng)建一個執(zhí)行的線程實(shí)例，所以它是一切并發(fā)編程的基礎(chǔ)，使用時需要包含 <thread> 頭文件， 它提供了很多基本的線程操作，例如 get_id() 來獲取所創(chuàng)建線程的線程 ID，使用 join() 來加入一個線程等等，例如：

#include <iostream>
#include <thread>
int main() {
    std::thread t([](){
        std::cout << "hello world." << std::endl;
    });
    t.join();
    return 0;
}

2. 互斥量與臨界區(qū)

我們在操作系統(tǒng)、亦或是數(shù)據(jù)庫的相關(guān)知識中已經(jīng)了解過了有關(guān)并發(fā)技術(shù)的基本知識，mutex 就是其中的核心之一。 C++11 引入了 mutex 相關(guān)的類，其所有相關(guān)的函數(shù)都放在 <mutex> 頭文件中。

std::mutex 是 C++11 中最基本的 mutex 類，通過實(shí)例化 std::mutex 可以創(chuàng)建互斥量， 而通過其成員函數(shù) lock() 可以進(jìn)行上鎖，unlock() 可以進(jìn)行解鎖。 但是在實(shí)際編寫代碼的過程中，最好不去直接調(diào)用成員函數(shù)， 因?yàn)檎{(diào)用成員函數(shù)就需要在每個臨界區(qū)的出口處調(diào)用 unlock()，當(dāng)然，還包括異常。 這時候 C++11 還為互斥量提供了一個 RAII 語法的模板類 std::lock_guard。 RAII 在不失代碼簡潔性的同時，很好的保證了代碼的異常安全性。

在 RAII 用法下，對于臨界區(qū)的互斥量的創(chuàng)建只需要在作用域的開始部分，例如：

#include <iostream>
#include <mutex>
#include <thread>
int v = 1;
void critical_section(int change_v) {
    static std::mutex mtx;
    std::lock_guard<std::mutex> lock(mtx);
    // 執(zhí)行競爭操作
    v = change_v;
    // 離開此作用域后 mtx 會被釋放
}
int main() {
    std::thread t1(critical_section, 2), t2(critical_section, 3);
    t1.join();
    t2.join();
    std::cout << v << std::endl;
    return 0;
}

由于 C++ 保證了所有棧對象在生命周期結(jié)束時會被銷毀，所以這樣的代碼也是異常安全的。 無論 critical_section() 正常返回、還是在中途拋出異常，都會引發(fā)堆棧回退，也就自動調(diào)用了 unlock()。

而 std::unique_lock 則是相對于 std::lock_guard 出現(xiàn)的，std::unique_lock 更加靈活， std::unique_lock 的對象會以獨(dú)占所有權(quán)（沒有其他的 unique_lock 對象同時擁有某個 mutex 對象的所有權(quán)） 的方式管理 mutex 對象上的上鎖和解鎖的操作。所以在并發(fā)編程中，推薦使用 std::unique_lock。

std::lock_guard 不能顯式的調(diào)用 lock 和 unlock， 而 std::unique_lock 可以在聲明后的任意位置調(diào)用， 可以縮小鎖的作用范圍，提供更高的并發(fā)度。

如果你用到了條件變量 std::condition_variable::wait 則必須使用 std::unique_lock 作為參數(shù)。

例如：

#include <iostream>
#include <mutex>
#include <thread>
int v = 1;
void critical_section(int change_v) {
    static std::mutex mtx;
    std::unique_lock<std::mutex> lock(mtx);
    // 執(zhí)行競爭操作
    v = change_v;
    std::cout << v << std::endl;
    // 將鎖進(jìn)行釋放
    lock.unlock();
    // 在此期間，任何人都可以搶奪 v 的持有權(quán)
    // 開始另一組競爭操作，再次加鎖
    lock.lock();
    v += 1;
    std::cout << v << std::endl;
}
int main() {
    std::thread t1(critical_section, 2), t2(critical_section, 3);
    t1.join();
    t2.join();
    return 0;
}

3. 期物

期物（Future）表現(xiàn)為 std::future，它提供了一個訪問異步操作結(jié)果的途徑，這句話很不好理解。 為了理解這個特性，我們需要先理解一下在 C++11 之前的多線程行為。

試想，如果我們的主線程 A 希望新開辟一個線程 B 去執(zhí)行某個我們預(yù)期的任務(wù)，并返回我一個結(jié)果。 而這時候，線程 A 可能正在忙其他的事情，無暇顧及 B 的結(jié)果， 所以我們會很自然的希望能夠在某個特定的時間獲得線程 B 的結(jié)果。

在 C++11 的 std::future 被引入之前，通常的做法是： 創(chuàng)建一個線程 A，在線程 A 里啟動任務(wù) B，當(dāng)準(zhǔn)備完畢后發(fā)送一個事件，并將結(jié)果保存在全局變量中。 而主函數(shù)線程 A 里正在做其他的事情，當(dāng)需要結(jié)果的時候，調(diào)用一個線程等待函數(shù)來獲得執(zhí)行的結(jié)果。

而 C++11 提供的 std::future 簡化了這個流程，可以用來獲取異步任務(wù)的結(jié)果。 自然地，我們很容易能夠想象到把它作為一種簡單的線程同步手段，即屏障（barrier）。

為了看一個例子，我們這里額外使用 std::packaged_task，它可以用來封裝任何可以調(diào)用的目標(biāo)，從而用于實(shí)現(xiàn)異步的調(diào)用。 舉例來說：

#include <iostream>
#include <future>
#include <thread>
int main() {
    // 將一個返回值為7的 lambda 表達(dá)式封裝到 task 中
    // std::packaged_task 的模板參數(shù)為要封裝函數(shù)的類型
    std::packaged_task<int()> task([](){return 7;});
    // 獲得 task 的期物
    std::future<int> result = task.get_future(); // 在一個線程中執(zhí)行 task
    std::thread(std::move(task)).detach();
    std::cout << "waiting...";
    result.wait(); // 在此設(shè)置屏障，阻塞到期物的完成
    // 輸出執(zhí)行結(jié)果
    std::cout << "done!" << std:: endl << "future result is "
              << result.get() << std::endl;
    return 0;
}

在封裝好要調(diào)用的目標(biāo)后，可以使用 get_future() 來獲得一個 std::future 對象，以便之后實(shí)施線程同步。

4. 條件變量

條件變量 std::condition_variable 是為了解決死鎖而生，當(dāng)互斥操作不夠用而引入的。 比如，線程可能需要等待某個條件為真才能繼續(xù)執(zhí)行， 而一個忙等待循環(huán)中可能會導(dǎo)致所有其他線程都無法進(jìn)入臨界區(qū)使得條件為真時，就會發(fā)生死鎖。 所以，condition_variable 實(shí)例被創(chuàng)建出現(xiàn)主要就是用于喚醒等待線程從而避免死鎖。 std::condition_variable的 notify_one() 用于喚醒一個線程； notify_all() 則是通知所有線程。下面是一個生產(chǎn)者和消費(fèi)者模型的例子：

#include <queue>
#include <chrono>
#include <mutex>
#include <thread>
#include <iostream>
#include <condition_variable>
int main() {
    std::queue<int> produced_nums;
    std::mutex mtx;
    std::condition_variable cv;
    bool notified = false;  // 通知信號
    // 生產(chǎn)者
    auto producer = [&]() {
        for (int i = 0; ; i++) {
            std::this_thread::sleep_for(std::chrono::milliseconds(900));
            std::unique_lock<std::mutex> lock(mtx);
            std::cout << "producing " << i << std::endl;
            produced_nums.push(i);
            notified = true;
            cv.notify_all(); // 此處也可以使用 notify_one
        }
    };
    // 消費(fèi)者
    auto consumer = [&]() {
        while (true) {
            std::unique_lock<std::mutex> lock(mtx);
            while (!notified) {  // 避免虛假喚醒
                cv.wait(lock);
            }
            // 短暫取消鎖，使得生產(chǎn)者有機(jī)會在消費(fèi)者消費(fèi)空前繼續(xù)生產(chǎn)
            lock.unlock();
            // 消費(fèi)者慢于生產(chǎn)者
            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            lock.lock();
            while (!produced_nums.empty()) {
                std::cout << "consuming " << produced_nums.front() << std::endl;
                produced_nums.pop();
            }
            notified = false;
        }
    };
    // 分別在不同的線程中運(yùn)行
    std::thread p(producer);
    std::thread cs[2];
    for (int i = 0; i < 2; ++i) {
        cs[i] = std::thread(consumer);
    }
    p.join();
    for (int i = 0; i < 2; ++i) {
        cs[i].join();
    }
    return 0;
}

值得一提的是，在生產(chǎn)者中我們雖然可以使用 notify_one()，但實(shí)際上并不建議在此處使用， 因?yàn)樵诙嘞M(fèi)者的情況下，我們的消費(fèi)者實(shí)現(xiàn)中簡單放棄了鎖的持有，這使得可能讓其他消費(fèi)者 爭奪此鎖，從而更好的利用多個消費(fèi)者之間的并發(fā)。話雖如此，但實(shí)際上因?yàn)?std::mutex 的排他性， 我們根本無法期待多個消費(fèi)者能真正意義上的并行消費(fèi)隊(duì)列的中生產(chǎn)的內(nèi)容，我們?nèi)孕枰６雀?xì)的手段。

5. 原子操作與內(nèi)存模型

細(xì)心的讀者可能會對前一小節(jié)中生產(chǎn)者消費(fèi)者模型的例子可能存在編譯器優(yōu)化導(dǎo)致程序出錯的情況產(chǎn)生疑惑。 例如，布爾值 notified 沒有被 volatile 修飾，編譯器可能對此變量存在優(yōu)化，例如將其作為一個寄存器的值， 從而導(dǎo)致消費(fèi)者線程永遠(yuǎn)無法觀察到此值的變化。這是一個好問題，為了解釋清楚這個問題，我們需要進(jìn)一步討論 從 C++ 11 起引入的內(nèi)存模型這一概念。我們首先來看一個問題，下面這段代碼輸出結(jié)果是多少？

#include <thread>
#include <iostream>
int main() {
    int a = 0;
    int flag = 0;
    std::thread t1([&]() {
        while (flag != 1);
        int b = a;
        std::cout << "b = " << b << std::endl;
    });
    std::thread t2([&]() {
        a = 5;
        flag = 1;
    });
    t1.join();
    t2.join();
    return 0;
}

從直觀上看，t2 中 a = 5; 這一條語句似乎總在 flag = 1; 之前得到執(zhí)行，而 t1 中 while (flag != 1) 似乎保證了 std::cout << "b = " << b << std::endl; 不會再標(biāo)記被改變前執(zhí)行。從邏輯上看，似乎 b 的值應(yīng)該等于 5。 但實(shí)際情況遠(yuǎn)比此復(fù)雜得多，或者說這段代碼本身屬于未定義的行為，因?yàn)閷τ?a 和 flag 而言，他們在兩個并行的線程中被讀寫， 出現(xiàn)了競爭。除此之外，即便我們忽略競爭讀寫，仍然可能受 CPU 的亂序執(zhí)行，編譯器對指令的重排的影響， 導(dǎo)致 a = 5 發(fā)生在 flag = 1 之后。從而 b 可能輸出 0。

5.1原子操作

std::mutex 可以解決上面出現(xiàn)的并發(fā)讀寫的問題，但互斥鎖是操作系統(tǒng)級的功能， 這是因?yàn)橐粋€互斥鎖的實(shí)現(xiàn)通常包含兩條基本原理：

提供線程間自動的狀態(tài)轉(zhuǎn)換，即『鎖住』這個狀態(tài)

保障在互斥鎖操作期間，所操作變量的內(nèi)存與臨界區(qū)外進(jìn)行隔離

這是一組非常強(qiáng)的同步條件，換句話說當(dāng)最終編譯為 CPU 指令時會表現(xiàn)為非常多的指令（我們之后再來看如何實(shí)現(xiàn)一個簡單的互斥鎖）。 這對于一個僅需原子級操作（沒有中間態(tài)）的變量，似乎太苛刻了。

關(guān)于同步條件的研究有著非常久遠(yuǎn)的歷史，我們在這里不進(jìn)行贅述。讀者應(yīng)該明白，現(xiàn)代 CPU 體系結(jié)構(gòu)提供了 CPU 指令級的原子操作， 因此在 C++11 中多線程下共享變量的讀寫這一問題上，還引入了 std::atomic 模板，使得我們實(shí)例化一個原子類型，將一個 原子類型讀寫操作從一組指令，最小化到單個 CPU 指令。例如：

std::atomic<int> counter;

并為整數(shù)或浮點(diǎn)數(shù)的原子類型提供了基本的數(shù)值成員函數(shù)，舉例來說， 包括 fetch_add, fetch_sub 等，同時通過重載方便的提供了對應(yīng)的 +，- 版本。 比如下面的例子：

#include <atomic>
#include <thread>
#include <iostream>
std::atomic<int> count = {0};
int main() {
    std::thread t1([](){
        count.fetch_add(1);
    });
    std::thread t2([](){
        count++;        // 等價于 fetch_add
        count += 1;     // 等價于 fetch_add
    });
    t1.join();
    t2.join();
    std::cout << count << std::endl;
    return 0;
}

當(dāng)然，并非所有的類型都能提供原子操作，這是因?yàn)樵硬僮鞯目尚行匀Q于具體的 CPU 架構(gòu)，以及所實(shí)例化的類型結(jié)構(gòu)是否能夠滿足該 CPU 架構(gòu)對內(nèi)存對齊 條件的要求，因而我們總是可以通過 std::atomic<T>::is_lock_free 來檢查該原子類型是否需支持原子操作，例如：

#include <atomic>
#include <iostream>
struct A {
    float x;
    int y;
    long long z;
};
int main() {
    std::atomic<A> a;
    std::cout << std::boolalpha << a.is_lock_free() << std::endl;
    return 0;
}

5.2一致性模型

并行執(zhí)行的多個線程，從某種宏觀層面上討論，可以粗略的視為一種分布式系統(tǒng)。 在分布式系統(tǒng)中，任何通信乃至本地操作都需要消耗一定時間，甚至出現(xiàn)不可靠的通信。

如果我們強(qiáng)行將一個變量 v 在多個線程之間的操作設(shè)為原子操作，即任何一個線程在操作完 v 后， 其他線程均能同步感知到 v 的變化，則對于變量 v 而言，表現(xiàn)為順序執(zhí)行的程序，它并沒有由于引入多線程 而得到任何效率上的收益。對此有什么辦法能夠適當(dāng)?shù)募铀倌?？答案便是削弱原子操作的在進(jìn)程間的同步條件。

從原理上看，每個線程可以對應(yīng)為一個集群節(jié)點(diǎn)，而線程間的通信也幾乎等價于集群節(jié)點(diǎn)間的通信。 削弱進(jìn)程間的同步條件，通常我們會考慮四種不同的一致性模型：

線性一致性：又稱強(qiáng)一致性或原子一致性。它要求任何一次讀操作都能讀到某個數(shù)據(jù)的最近一次寫的數(shù)據(jù)，并且所有線程的操作順序與全局時鐘下的順序是一致的。

        x.store(1)      x.load()
T1 ---------+----------------+------>
T2 -------------------+------------->
                x.store(2)

在這種情況下線程 T1, T2 對 x 的兩次寫操作是原子的，且 x.store(1) 是嚴(yán)格的發(fā)生在 x.store(2) 之前，x.store(2) 嚴(yán)格的發(fā)生在 x.load() 之前。 值得一提的是，線性一致性對全局時鐘的要求是難以實(shí)現(xiàn)的，這也是人們不斷研究比這個一致性更弱條件下其他一致性的算法的原因。

順序一致性：同樣要求任何一次讀操作都能讀到數(shù)據(jù)最近一次寫入的數(shù)據(jù)，但未要求與全局時鐘的順序一致。

        x.store(1)  x.store(3)   x.load()
T1 ---------+-----------+----------+----->
T2 ---------------+---------------------->
              x.store(2)

或者

        x.store(1)  x.store(3)   x.load()
T1 ---------+-----------+----------+----->
T2 ------+------------------------------->
      x.store(2)

在順序一致性的要求下，x.load() 必須讀到最近一次寫入的數(shù)據(jù)，因此 x.store(2) 與 x.store(1) 并無任何先后保障，即 只要 T2 的 x.store(2) 發(fā)生在 x.store(3) 之前即可。

因果一致性：它的要求進(jìn)一步降低，只需要有因果關(guān)系的操作順序得到保障，而非因果關(guān)系的操作順序則不做要求。

      a = 1      b = 2
T1 ----+-----------+---------------------------->
T2 ------+--------------------+--------+-------->
      x.store(3)         c = a + b    y.load()

或者

      a = 1      b = 2
T1 ----+-----------+---------------------------->
T2 ------+--------------------+--------+-------->
      x.store(3)          y.load()   c = a + b

亦或者

     b = 2       a = 1
T1 ----+-----------+---------------------------->
T2 ------+--------------------+--------+-------->
      y.load()            c = a + b  x.store(3)

上面給出的三種例子都是屬于因果一致的，因?yàn)檎麄€過程中，只有 c 對 a 和 b 產(chǎn)生依賴，而 x 和 y 在此例子中表現(xiàn)為沒有關(guān)系（但實(shí)際情況中我們需要更詳細(xì)的信息才能確定 x 與 y 確實(shí)無關(guān)）

最終一致性：是最弱的一致性要求，它只保障某個操作在未來的某個時間節(jié)點(diǎn)上會被觀察到，但并未要求被觀察到的時間。因此我們甚至可以對此條件稍作加強(qiáng)，例如規(guī)定某個操作被觀察到的時間總是有界的。當(dāng)然這已經(jīng)不在我們的討論范圍之內(nèi)了。

    x.store(3)  x.store(4)
T1 ----+-----------+-------------------------------------------->
T2 ---------+------------+--------------------+--------+-------->
         x.read      x.read()           x.read()   x.read()

在上面的情況中，如果我們假設(shè) x 的初始值為 0，則 T2 中四次 x.read() 結(jié)果可能但不限于以下情況：

3 4 4 4 // x 的寫操作被很快觀察到

0 3 3 4 // x 的寫操作被觀察到的時間存在一定延遲

0 0 0 4 // 最后一次讀操作讀到了 x 的最終值，但此前的變化并未觀察到

0 0 0 0 // 在當(dāng)前時間段內(nèi) x 的寫操作均未被觀察到，

// 但未來某個時間點(diǎn)上一定能觀察到 x 為 4 的情況

5.3內(nèi)存順序

為了追求極致的性能，實(shí)現(xiàn)各種強(qiáng)度要求的一致性，C++11 為原子操作定義了六種不同的內(nèi)存順序 std::memory_order 的選項(xiàng)，表達(dá)了四種多線程間的同步模型：

寬松模型：在此模型下，單個線程內(nèi)的原子操作都是順序執(zhí)行的，不允許指令重排，但不同線程間原子操作的順序是任意的。類型通過 std::memory_order_relaxed 指定。我們來看一個例子：

std::atomic<int> counter = {0};
std::vector<std::thread> vt;
for (int i = 0; i < 100; ++i) {
    vt.emplace_back([&](){
        counter.fetch_add(1, std::memory_order_relaxed);
    });
}
for (auto& t : vt) {
    t.join();
}
std::cout << "current counter:" << counter << std::endl;

釋放/消費(fèi)模型：在此模型中，我們開始限制進(jìn)程間的操作順序，如果某個線程需要修改某個值，但另一個線程會對該值的某次操作產(chǎn)生依賴，即后者依賴前者。具體而言，線程 A 完成了三次對 x 的寫操作，線程 B 僅依賴其中第三次 x 的寫操作，與 x 的前兩次寫行為無關(guān)，則當(dāng) A 主動 x.release() 時候（即使用 std::memory_order_release），選項(xiàng) std::memory_order_consume 能夠確保 B 在調(diào)用 x.load() 時候觀察到 A 中第三次對 x 的寫操作。我們來看一個例子：

// 初始化為 nullptr 防止 consumer 線程從野指針進(jìn)行讀取
std::atomic<int*> ptr(nullptr);
int v;
std::thread producer([&]() {
    int* p = new int(42);
    v = 1024;
    ptr.store(p, std::memory_order_release);
});
std::thread consumer([&]() {
    int* p;
    while(!(p = ptr.load(std::memory_order_consume)));
    std::cout << "p: " << *p << std::endl;
    std::cout << "v: " << v << std::endl;
});
producer.join();
consumer.join();

釋放/獲取模型：在此模型下，我們可以進(jìn)一步加緊對不同線程間原子操作的順序的限制，在釋放 std::memory_order_release 和獲取 std::memory_order_acquire 之間規(guī)定時序，即發(fā)生在釋放（release）操作之前的所有寫操作，對其他線程的任何獲?。╝cquire）操作都是可見的，亦即發(fā)生順序（happens-before）。

可以看到，std::memory_order_release 確保了它之前的寫操作不會發(fā)生在釋放操作之后，是一個向后的屏障（backward），而 std::memory_order_acquire 確保了它之前的寫行為不會發(fā)生在該獲取操作之后，是一個向前的屏障（forward）。對于選項(xiàng) std::memory_order_acq_rel 而言，則結(jié)合了這兩者的特點(diǎn)，唯一確定了一個內(nèi)存屏障，使得當(dāng)前線程對內(nèi)存的讀寫不會被重排并越過此操作的前后：

我們來看一個例子：

std::vector<int> v;
std::atomic<int> flag = {0};
std::thread release([&]() {
    v.push_back(42);
    flag.store(1, std::memory_order_release);
});
std::thread acqrel([&]() {
    int expected = 1; // must before compare_exchange_strong
    while(!flag.compare_exchange_strong(expected, 2, std::memory_order_acq_rel))
        expected = 1; // must after compare_exchange_strong
    // flag has changed to 2
});
std::thread acquire([&]() {
    while(flag.load(std::memory_order_acquire) < 2);
    std::cout << v.at(0) << std::endl; // must be 42
});
release.join();
acqrel.join();
acquire.join();

在此例中我們使用了 compare_exchange_strong 比較交換原語（Compare-and-swap primitive），它有一個更弱的版本，即 compare_exchange_weak，它允許即便交換成功，也仍然返回 false 失敗。其原因是因?yàn)樵谀承┢脚_上虛假故障導(dǎo)致的，具體而言，當(dāng) CPU 進(jìn)行上下文切換時，另一線程加載同一地址產(chǎn)生的不一致。除此之外，compare_exchange_strong 的性能可能稍差于 compare_exchange_weak，但大部分情況下，鑒于其使用的復(fù)雜度而言，compare_exchange_weak 應(yīng)該被有限考慮。

順序一致模型：在此模型下，原子操作滿足順序一致性，進(jìn)而可能對性能產(chǎn)生損耗?？娠@式的通過 std::memory_order_seq_cst 進(jìn)行指定。最后來看一個例子：
std::atomic<int> counter = {0};
std::vector<std::thread> vt;
for (int i = 0; i < 100; ++i) {
    vt.emplace_back([&](){
        counter.fetch_add(1, std::memory_order_seq_cst);
    });
}
for (auto& t : vt) {
    t.join();
}
std::cout << "current counter:" << counter << std::endl;

這個例子與第一個寬松模型的例子本質(zhì)上沒有區(qū)別，僅僅只是將原子操作的內(nèi)存順序修改為了 memory_order_seq_cst，有興趣的讀者可以自行編寫程序測量這兩種不同內(nèi)存順序?qū)е碌男阅懿町悺?/p>

到此這篇關(guān)于C++中的并行與并發(fā)基礎(chǔ)與使用詳解的文章就介紹到這了,更多相關(guān)C++并行與并發(fā)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！