深入理解C++ 信号量/闩锁/屏障(C++20)

C++20 引入了信号量（std::counting_semaphore）、闩锁（std::latch）和屏障（std::barrier），作为比条件变量更轻量、语义更直接的同步原语。它们各自解决不同类型的协调问题，下面逐一展开原理与可运行示例。

总览与对比

关键区别在于复用性：闩锁是一次性的，计数归零后生命周期即结束；屏障可以反复使用，每一“轮”所有参与者到达后自动重置，进入下一轮。信号量则既不关心“哪些线程”到达，也不自动重置，只是一个可增减的资源计数器。

一、信号量：std::counting_semaphore / std::binary_semaphore

原理

信号量内部维护一个计数器，表示当前可用资源的数量。acquire() 尝试将计数减 1；若计数已为 0，调用线程阻塞，直到其他线程调用 release() 将计数加 1 并唤醒一个等待者。std::binary_semaphore 是 std::counting_semaphore<1> 的别名，退化为二值信号量。

与互斥锁的关键差异：信号量不要求“谁获取谁释放”。互斥锁的解锁必须由加锁的同一线程完成，而信号量的 release() 可以从任意线程调用，因此它可以跨线程传递“资源可用”的信号。

核心 API

cpp

#include <semaphore>

std::counting_semaphore<MaxValue> sem(initial_count);

sem.acquire(); // 计数减1，为0时阻塞
sem.try_acquire(); // 非阻塞尝试，失败返回false
sem.try_acquire_for(dur); // 带超时的尝试
sem.release(n = 1); // 计数加n，唤醒最多n个等待者

可运行示例：限制并发数为 3

以下程序创建 10 个线程，但任意时刻最多只有 3 个线程能同时执行“工作”段：

cpp

// semaphore_demo.cpp
// 编译：g++ -std=c++20 -pthread semaphore_demo.cpp -o sem
// 运行：./sem

#include <semaphore>
#include <thread>
#include <vector>
#include <iostream>
#include <syncstream>
#include <chrono>

using namespace std::chrono_literals;

std::counting_semaphore<3> sem(3); // 最多3个并发许可

void worker(int id) {
sem.acquire(); // 请求一个许可，无许可则等待
{
std::osyncstream(std::cout)
<< "[Worker " << id << "] 开始执行\\n";
std::this_thread::sleep_for(200ms); // 模拟工作
std::osyncstream(std::cout)
<< "[Worker " << id << "] 执行结束，释放许可\\n";
}
sem.release(); // 归还许可，唤醒等待者
}

int main() {
std::vector<std::jthread> threads;
for (int i = 0; i < 10; ++i)
threads.emplace_back(worker, i);
// jthread 析构时自动 join
}

运行时会看到输出严格呈现“同时最多 3 个 Worker 处于开始/结束之间”的节奏，后续 Worker 在 acquire() 处排队等待。

信号量的独特价值：跨线程信号

信号量也可以用作轻量的线程间事件通知，而不需要像条件变量那样搭配互斥锁和谓词。std::binary_semaphore 初始计数为 0 时相当于“门关闭”，另一线程调用 release() 即“开门”：

cpp

std::binary_semaphore ready(0); // 初始无许可，等待方会阻塞

// 线程 A（等待方）
ready.acquire(); // 阻塞直到有人 release

// 线程 B（通知方）
// … 完成准备工作 …
ready.release(); // 唤醒等待方

这本质上是对“事件是否发生”这一条件的最简表达，比条件变量方案少一个互斥锁和谓词检查。

二、闩锁：std::latch

原理

std::latch 是一个一次性倒计数器。构造时给定初始计数 N；任意线程可以调用 count_down() 将计数减 1；需要等待的线程调用 wait() 阻塞，直到计数归零。计数一旦归零，闩锁永久保持为零，不能重置或增加，因此它只能使用一次。

count_down() 不阻塞，调用后线程可以继续做别的事；wait() 才阻塞。如果需要“到达即等待”，使用 arrive_and_wait()，它等价于 count_down() + wait()。

核心 API

cpp

#include <latch>

std::latch lat(N);

lat.count_down(n = 1); // 计数减n，不阻塞
lat.wait(); // 阻塞直到计数为0
lat.arrive_and_wait(n = 1); // count_down(n) + wait()
lat.try_wait(); // 非阻塞检查是否已归零

// latch_demo.cpp
// 编译：g++ -std=c++20 -pthread latch_demo.cpp -o lat
// 运行：./lat

#include <latch>
#include <thread>
#include <vector>
#include <iostream>
#include <syncstream>
#include <chrono>

struct Service {
std::string name;
void initialize() {
// 模拟不同耗时的初始化
std::this_thread::sleep_for(
std::chrono::milliseconds(100 + name.size() * 30));
}
};

int main() {
constexpr int N = 4;
std::vector<Service> services = {
{"Database"}, {"Cache"}, {"Auth"}, {"Logger"}
};

std::latch init_latch(N); // 需要 N 次 count_down

std::vector<std::jthread> threads;
for (int i = 0; i < N; ++i) {
threads.emplace_back([&, i] {
services[i].initialize();
std::osyncstream(std::cout)
<< services[i].name << " 初始化完成\\n";
init_latch.count_down(); // 报告：我完成了，但不阻塞
});
}

init_latch.wait(); // 主线程阻塞，直到 N 个都完成
std::cout << "全部服务初始化完成，主流程继续\\n";
}