[[428229]]

本文轉(zhuǎn)載自微信公眾號(hào)「程序喵大人」，作者程序喵大人。轉(zhuǎn)載本文請(qǐng)聯(lián)系程序喵大人公眾號(hào)。

之前我整理過一篇C++20新特性的文章全網(wǎng)首發(fā)!!C++20新特性全在這一張圖里了，里面提到過latch、barrier和semaphore，但是沒有詳細(xì)介紹過三者的作用和區(qū)別，這里詳細(xì)介紹下。

latch

這個(gè)可能大多數(shù)人都有所了解，這就是我們經(jīng)常會(huì)用到的CountDownLatch。用于使一個(gè)線程先阻塞，等待其他線程完成各自的工作后再繼續(xù)執(zhí)行。

CountDownLatch是通過計(jì)數(shù)器實(shí)現(xiàn)，計(jì)數(shù)器的初始值為線程的數(shù)量。每當(dāng)一個(gè)線程完成了自己的任務(wù)后，計(jì)數(shù)器的值就會(huì)減1。當(dāng)計(jì)數(shù)器值到達(dá)0時(shí)，它表示所有的線程已經(jīng)完成了任務(wù)，然后等待的線程就可以打斷阻塞去繼續(xù)執(zhí)行任務(wù)。

自己之前實(shí)現(xiàn)過一個(gè)CountDownLatch，源碼大概這樣：

CountDownLatch::CountDownLatch(int32_t count) : count_(count) {} 
 
void CountDownLatch::CountDown() { 
    std::unique_lock<std::mutex> lock(mutex_); 
    --count_; 
    if (count_ == 0) { 
        cv_.notify_all(); 
    } 
} 
 
void CountDownLatch::Await(int32_t time_ms) { 
    std::unique_lock<std::mutex> lock(mutex_); 
    while (count_ > 0) { 
        if (time_ms > 0) { 
            cv_.wait_for(lock, std::chrono::milliseconds(time_ms)); 
        } else { 
            cv_.wait(lock); 
        } 
    } 
} 
 
int32_t CountDownLatch::GetCount() const { 
    std::unique_lock<std::mutex> lock(mutex_); 
    return count_; 
} 

barrier

許多線程在阻塞點(diǎn)阻塞，當(dāng)?shù)竭_(dá)阻塞點(diǎn)的線程達(dá)到一定數(shù)量時(shí)，會(huì)執(zhí)行完成的回調(diào)，然后解除所有相關(guān)線程的阻塞，然后重置線程計(jì)數(shù)器，繼續(xù)開始下一階段的阻塞。

假設(shè)有很多線程并發(fā)執(zhí)行，并在一個(gè)循環(huán)中執(zhí)行一些計(jì)算。進(jìn)一步假設(shè)一旦這些計(jì)算完成，需要在線程開始其循環(huán)的新迭代之前對(duì)結(jié)果進(jìn)行一些處理。

看以下示例代碼(摘自cppreference)：

#include <barrier> 
#include <iostream> 
#include <string> 
#include <thread> 
#include <vector> 
  
int main() { 
  const auto workers = { "anil", "busara", "carl" }; 
  
  auto on_completion = []() noexcept {  
    // locking not needed here 
    static auto phase = "... done\n" "Cleaning up...\n"; 
    std::cout << phase; 
    phase = "... done\n"; 
  }; 
  std::barrier sync_point(std::ssize(workers), on_completion); 
  
  auto work = [&](std::string name) { 
    std::string product = "  " + name + " worked\n"; 
    std::cout << product;  // ok, op<< call is atomic 
    sync_point.arrive_and_wait(); 
  
    product = "  " + name + " cleaned\n"; 
    std::cout << product; 
    sync_point.arrive_and_wait(); 
  }; 
  
  std::cout << "Starting...\n"; 
  std::vector<std::thread> threads; 
  for (auto const& worker : workers) { 
    threads.emplace_back(work, worker); 
  } 
  for (auto& thread : threads) { 
    thread.join(); 
  } 
} 

可能的輸出如下：

Starting... 
  anil worked 
  carl worked 
  busara worked 
... done 
Cleaning up... 
  busara cleaned 
  carl cleaned 
  anil cleaned 
... done 

semaphore

信號(hào)量，這個(gè)估計(jì)大家都很熟悉，本質(zhì)也是個(gè)計(jì)數(shù)器，主要有兩個(gè)方法：

acquire()：遞減計(jì)數(shù)器，當(dāng)計(jì)數(shù)器為零時(shí)阻塞，直到計(jì)數(shù)器再次遞增。

release()：遞增計(jì)數(shù)器(可傳遞具體數(shù)字)，并解除在acquire調(diào)用中的線程的阻塞。

示例代碼如下：

#include <iostream> 
#include <thread> 
#include <chrono> 
#include <semaphore> 
 
std::binary_semaphore 
  smphSignalMainToThread(0), 
  smphSignalThreadToMain(0); 
  
void ThreadProc() {   
  smphSignalMainToThread.acquire(); 
  std::cout << "[thread] Got the signal\n"; // response message 
  using namespace std::literals; 
  std::this_thread::sleep_for(3s); 
  
  std::cout << "[thread] Send the signal\n"; // message 
  smphSignalThreadToMain.release(); 
} 
  
int main() { 
  std::thread thrWorker(ThreadProc); 
  std::cout << "[main] Send the signal\n"; // message 
  
  smphSignalMainToThread.release(); 
  
  smphSignalThreadToMain.acquire(); 
  
  std::cout << "[main] Got the signal\n"; // response message 
  thrWorker.join(); 
} 
輸出如下： 
[main] Send the signal 
[thread] Got the signal 
[thread] Send the signal 
[main] Got the signal 

信號(hào)量也可以當(dāng)作條件變量使用，這個(gè)我估計(jì)大家應(yīng)該知道怎么做。

打完收工。