写C并发代码的人迟早会遇到一个尴尬C11把mutex、条件变量、future全给了唯独信号量没给。标准委员会直到C20才补上std::counting_semaphore可现实里的项目大多是C11或C14没法升级。于是自己动手封装一个信号量就成了很多团队绕不开的活儿。我最初也是被逼无奈才写的真正用起来之后发现这个看似二十来行的小类其实相当讲究——设计、语义、坑一个不少。这篇文章就把我用C11实现信号量的完整思路、可复用代码和踩坑记录整理出来给同样需要的人一个参考。1. 为什么C11标准库偏偏没给信号量1.1 标准库到底给了什么并发工具C11首次在语言层面纳入了并发支持这是一次大跨越。std::thread负责创建线程std::mutex负责互斥std::condition_variable负责线程间的等待与唤醒std::future/std::promise负责异步结果传递std::atomic负责无锁的原子操作。这套组合拳看起来挺完整足以应对绝大多数并发场景。但仔细一琢磨会发现一个缺失没有信号量。标准库提供的互斥量和条件变量只能解决“锁”和“唤醒”这两个基础问题对于“允许多个线程同时进入临界资源但最多N个”这种更上层的需求并没有直接对应的原语。大家通常的做法是用mutex加一个计数变量再配合条件变量手动实现本质就是在写一个迷你信号量。标准委员会显然也意识到了这个问题于是在C20引入了std::counting_semaphore和std::binary_semaphore。对还在用C11的老项目来说自己封装还是唯一靠谱的路。1.2 信号量跟互斥锁到底差在哪很多人把信号量理解成“可以计数版本的锁”这个说法方向对但容易误导。互斥锁讲究的是“所有权”谁lock了谁就必须unlock别人不能替它解锁这是为了确保临界区的受控访问。信号量没有所有权的概念任何线程都可以执行signal操作去增加计数并不要求必须是之前执行wait的那个线程。这一点是信号量和互斥锁最本质的区别也是信号量能用来做“线程间通知”的原因。行为上互斥锁只有0和1两种状态信号量可以是一个非负整数。这就让信号量天然适合那些“资源池数量固定、多个线程争夺有限资源”的场景。比如数据库连接池允许同时最多10个连接线程池里最多同时跑8个任务或者生产者和消费者共享一个有上限的缓冲区。用互斥锁做这些事会很别扭因为互斥锁本身只有一个名额你只能设多个锁再加一堆逻辑去模拟复杂度会爆炸。信号量用一个计数器就把名额管理起来了。用一个粗糙的生活类比停车场入口的显示牌就是典型的信号量。显示牌上的剩余车位就是计数器每进去一辆车执行“wait”剩余车位减一每出去一辆车执行“signal”剩余车位加一。当剩余车位为0时后续的车就得等在门口。互斥锁则更像厕所里的插销一个人进去把门锁上别人进不来直到他出来解锁。信号量管理的是复数资源互斥锁管理的是单数资源两者面向的问题本质不同。1.3 没有信号量时的替代方案在没有信号量的年代大家常用的替代手段有两个一是用条件变量裸写一个“自制的计数等待”也就是用一个int计数加std::mutex再加std::condition_variable手动组合二是干脆把所有场景都设计成互斥锁加队列用队列的长度代替信号的计数。第二个方案在很多场景也能跑但代码可读性差而且队列本身还需要加锁保护层级一多就容易出死锁。替代方案能用的前提是“并发量不大、场景简单”一旦碰到经典的生产者消费者模型或者需要控制并发任务数量的限流场景要么代码绕得没法看要么就得有一个像样的信号量类。我在实际项目中两种方案都用过最后的结论是自己封装一个信号量成本极低收益很高。2. 信号量的语义设计动手之前先把状态机想清楚2.1 核心操作wait与signal的语义信号量本质是一个非负整数计数器核心操作就两个。wait也常叫P操作或acquire如果计数大于0把计数减1并立即返回如果计数等于0调用线程阻塞直到计数重新大于0然后再减1并返回。signal也常叫V操作或release把计数加1如果有线程正阻塞在wait上唤醒其中一个。这两个操作必须保证原子性也就是说同一个信号量上的并发wait/signal不能互相干扰计数绝不能出现负数。这个状态机不复杂但有一个隐蔽的点wait操作唤醒后必须把“计数大于0”这个检查放在循环里而不是只检查一次。原因很微妙后面我会专门讲这里先记住结论正确的实现一定是在循环里等并且把“计数大于0”作为条件传给条件变量的wait。2.2 底层组合mutex保证唯一修改权condition_variable负责阻塞与唤醒用C11实现信号量最标准的底层组合就是std::mutex加std::condition_variable。mutex管的是“谁都别同时改count”condition_variable管的是“等count大于0的线程怎么挂起、怎么唤醒”。为什么这么分工因为count是共享变量所有wait和signal都在改它不加锁的话并发修改会产生数据竞争这是未定义行为。而count从0变成大于0这个事件需要一个让线程“睡下并等待事件发生”的机制。std::condition_variable::wait会做一件很巧妙的事把当前线程挂起之前先原子地释放mutex这样别的线程才能拿到锁去修改count并调用notify被唤醒后wait返回之前又会重新获得mutex。这一整套“释放锁—挂起—唤醒—重新拿锁”的过程恰好能满足信号量对P/V操作原子性的要求。这套组合类似于你有一个仓库仓库门锁mutex保证只有一个人能进去盘货仓库门口有个广播喇叭condition_variable。货来了就广播一声等货的人听到广播再排队进仓库。广播本身不分发货物它只是叫人来取货货物数量是仓库里真实存的count。2.3 接口设计要考虑哪些操作信号量的接口不能只有wait和signal实际用到的时候至少还需要支持以下操作wait()阻塞式申请一个资源。try_wait()非阻塞尝试申请当时没有资源就返回false不等待。wait_for()带超时上限地申请超时就放弃。signal()释放一个资源并唤醒一个等待线程。signal_all()释放一个资源并唤醒所有等待线程某些特殊场景有用。其中try_wait和wait_for看似是锦上添花实则是刚需。很多业务场景不允许无限期等待比如网络请求的超时控制比如任务调度里的“抢不到就干别的”没有这两个接口使用方就得自己再造一层轮子去模拟那代码风格就很丑陋了。2.4 为什么裸写条件变量容易踩坑有人可能觉得信号量不就是“一个计数器加条件变量”么每次临时写一段不就行了为了一时方便确实可以但问题是一段没有边界和语义约束的临时代码非常容易犯错。最常见的错误是把wait写成if (count 0) cv.wait(lock);而不是while(count 0) cv.wait(lock);一旦发生所谓的“虚假唤醒”线程会在count仍然为0的情况下继续执行然后把count减成-1信号量就彻底坏了。另一个常见错误是signal里忘记唤醒或者只唤醒不把count加回去导致等待线程永远挂起。这些错误在并发环境下极难复现偶尔出现一次定位成本高得吓人。与其每次临时写不如花二十分钟封装一个正经的Semaphore类统一语义集中测试以后所有调用方都走同一个可靠的入口。这个决定是我在实际调试经验中得到的刚开始我也觉得“没必要”后来被一个偶现的卡死bug折磨了两天才醒悟。3. 手写一个可直接复用的Semaphore类3.1 完整实现源码先说结论下面是我目前项目里用的版本C11标准可以直接编译#include mutex #include condition_variable #include chrono class Semaphore { public: explicit Semaphore(int count 0) : count_(count) { if (count_ 0) count_ 0; } // 禁止拷贝 Semaphore(const Semaphore) delete; Semaphore operator(const Semaphore) delete; void wait() { std::unique_lockstd::mutex lock(mutex_); cond_.wait(lock, [this] { return count_ 0; }); --count_; } bool try_wait() { std::unique_lockstd::mutex lock(mutex_, std::try_to_lock); if (!lock.owns_lock()) return false; if (count_ 0) { --count_; return true; } return false; } template typename Rep, typename Period bool wait_for(const std::chrono::durationRep, Period timeout) { std::unique_lockstd::mutex lock(mutex_); bool acquired cond_.wait_for(lock, timeout, [this] { return count_ 0; }); if (acquired) --count_; return acquired; } void signal() { { std::lock_guardstd::mutex lock(mutex_); count_; } cond_.notify_one(); } void signal_all() { { std::lock_guardstd::mutex lock(mutex_); count_; } cond_.notify_all(); } private: std::mutex mutex_; std::condition_variable cond_; int count_; };这段代码不长但里面每一个细节都是有讲究的。我逐条拆一下。3.2 关键细节逐行解读先看wait()。cond_.wait(lock, predicate)这个带谓词的重载是C11就提供的它等价于下面这段手写循环while (!predicate()) cond_.wait(lock);也就是说它内部已经帮你处理了“虚假唤醒”的问题。线程被唤醒后会重新检查count_ 0如果此刻count仍然是0它会继续睡回去。这对信号量的正确性至关重要。然后--count_发生在锁内此时调用线程已经通过检查确认count至少为1减1不会让count变负。再来看signal()。我用一个块作用域把加锁和count_包起来然后出了作用域立即解锁再调用notify_one()。为什么在锁外通知因为条件变量notify不需要持有锁如果持锁通知唤醒的线程会立刻去抢这把还没释放的锁产生不必要的上下文切换。先解锁再通知能略微提升一点性能。这里有一个容易误解的地方有人担心先解锁再notify会导致“通知丢失”。实际上不会因为wait线程被唤醒后不是只靠通知信号来决定是否继续它还要重新检查谓词“count_ 0”。即使通知在wait真正挂起之前就发出去了线程也会在wait内部检查谓词时发现条件已满足直接不睡了。真正的防丢失机制是谓词本身而不是通知的时序。try_wait()这里我用了一个跟最初的版本不同的做法使用std::try_to_lock构造unique_lock。原因很简单try_wait()的语义是“有资源就抢没有或抢不到锁就立刻返回false”。如果用普通的lock()万一别的线程正持锁做操作我这个线程虽然只是检查一下count却要傻等锁释放这就把“非阻塞尝试”变成了“阻塞尝试”语义不对。搭配try_to_lock后lock()不再阻塞拿不到锁就返回false这才是真正的try。wait_for()函数只用了模板没有写死时间单位。调用方可以传std::chrono::milliseconds(100)也可以传std::chrono::seconds(2)甚至自定义时长都非常方便。cond_.wait_for(lock, timeout, predicate)同样带谓词重载返回值是bool而不是cv_status——返回true表示谓词在超时前满足了也就是成功获取信号量返回false表示超时且谓词仍未满足。在拿到true后--count_仍然是安全的因为谓词已经确认count大于0且整个过程都在锁保护下。3.3 try_wait的两种实现和取舍我见过不少实现把try_wait写得很随意比如直接lock()然后检查count再返回。这种写法的问题我上面说了它本质上会阻塞在锁获取上跟你期望的“非阻塞尝试”不符。另一种极端是用try_lock()裸调用但这样就得自己管锁的生命周期代码容易漏解锁。用std::unique_lockstd::mutex lock(mutex_, std::try_to_lock)是最稳妥的它把锁的生命周期交给RAII管理同时通过owns_lock()判断有没有真的拿到锁。有人会问既然用的是非阻塞尝试如果锁被占用导致try_wait返回false但实际上count是大于0的这算不算误判这确实是try_wait这种实现的一个内在特性。不过在信号量的场景下锁的持有时间极短就是几行改count的代码所以这种“因锁竞争而失败”的概率很低。如果真的在意可以设计成“转圈多试几次再放弃”但一般没必要徒增复杂度而已。3.4 拷贝与移动不能拷移动也要谨慎std::mutex和std::condition_variable都是不可拷贝、不可移动的所以我直接把Semaphore类的拷贝构造和拷贝赋值delete掉了。这样一来你没办法把一个Semaphore放到需要复制语义的容器里比如std::vectorSemaphore vec; vec.push_back(sem);会编译失败。这是特性而不是bug因为信号量本身就是系统资源复制一份会导致两个完全独立的信号量语义上是灾难。那如果就是想维护一组信号量怎么办常见的做法是用std::unique_ptrSemaphore比如std::vectorstd::unique_ptrSemaphore这样容器只复制指针真正的信号量对象保持唯一身份。还有另一个思路是给Semaphore实现移动语义构造新对象时重新初始化mutex和cond_并把count搬过去。但这有一个风险如果原对象上有线程正阻塞在wait移动之后那些线程等待的条件变量已经不是原来那个了这会导致未定义行为。所以我个人不建议给Semaphore提供移动语义除非你能保证移动时没有线程在等它而这个保证在实际项目中很难成立。4. 实战三个高频场景直接用起来4.1 经典生产者-消费者模型生产者消费者是信号量最典型的应用场景。一个有限大小的缓冲区生产者往里面放数据消费者从里面取数据缓冲区满时生产者得等缓冲区空时消费者得等。这里需要两个信号量empty表示空位数量初始为缓冲区大小full表示已有数据的数量初始为0。生产者放数据前先empty.wait()放完后再full.signal()消费者取数据前先full.wait()取完后再empty.signal()。两个信号量一阴一阳正好把双方挂在正确的等待条件上。写一个简化但完整的示例#include thread #include vector #include queue #include mutex #include condition_variable #include iostream Semaphore g_empty(5); Semaphore g_full(0); std::mutex g_queue_mutex; std::queueint g_queue; void producer(int id) { for (int i 0; i 10; i) { g_empty.wait(); { std::lock_guardstd::mutex lock(g_queue_mutex); g_queue.push(id * 100 i); } g_full.signal(); } } void consumer(int id) { for (int i 0; i 10; i) { g_full.wait(); int value; { std::lock_guardstd::mutex lock(g_queue_mutex); value g_queue.front(); g_queue.pop(); } g_empty.signal(); std::cout consumer id got value std::endl; } }注意这里一个很重要的细节我并没有试图让信号量本身保护队列的数据完整性。信号量负责的只是“空位”和“数据量”这两个计数条件队列本身另外用std::mutex保护。如果你在一个锁里面再去等信号量容易出问题。好的设计原则是要么先等信号量再拿队列锁要么先拿队列锁再等信号量在全项目里保持统一的顺序。我上面的代码是先等信号量再拿队列锁这个顺序可以避免死锁因为信号量wait不会持有队列锁去等另一个线程释放队列锁。4.2 并发限流同时只允许N个任务执行后台服务里经常要做限流。比如我有一个任务执行器希望任意时刻最多只有3个任务在真正运行其余任务排队等待。这种场景用互斥锁没法直接做因为互斥锁只允许一个线程进入但用信号量就是一句话的事Semaphore g_limit(3); void worker(int task_id) { g_limit.wait(); // 这里执行真正耗时的任务 std::this_thread::sleep_for(std::chrono::milliseconds(50)); g_limit.signal(); }这个例子里信号量g_limit的初始值3就是“许可证”的数量。前三个线程能立刻拿到许可证进入执行区第四个线程来了只能阻塞在wait上直到前三个里有某个执行完调用了signal。这套机制等价于一个轻量级的线程池限流器不需要额外维护任务队列和空闲线程池非常适合快速给现有代码加并发上限。实际项目里更常见的做法是把worker函数里的任务内容抽象成一个可调用对象配合std::async或者线程池来跑信号量只负责控制并发度。这种解耦方式的好处是限流策略与任务执行逻辑完全分离改上限只需要改一个数字。4.3 one-shot事件通知等初始化完成信号量的另一个隐蔽用途是做线程间的“一次性事件”通知。比如系统启动时主线程要先完成某个资源初始化工作线程必须等初始化完成后才能开始干活。用信号量实现非常优雅初始化为0主线程初始化完成后调用signal()工作线程调用wait()等待。这个模型跟条件变量加flag的写法相比代码量少一大截而且不需要多余的mutex和bool变量。Semaphore g_init_ready(0); void init_thread() { // 模拟耗时初始化 std::this_thread::sleep_for(std::chrono::milliseconds(200)); g_init_ready.signal(); } void worker_thread() { g_init_ready.wait(); // 初始化完成后的工作 }这里信号量不是被当作“资源计数”用而是被当作“事件标志”用。计数从0变1的那一下就是事件发生的瞬间。如果时多个工作线程同时等事件可以让它们都调用wait主线程用signal_all()唤醒所有等待者。5. 我踩过的坑和排查方法5.1 虚假唤醒为什么wait必须配predicate前面说了cond_.wait(lock, predicate)自带循环检查。但如果你想省事想自己写if (count_ 0) cond_.wait(lock);那就等着出bug。虚假唤醒是条件变量机制在操作系统层面的一个客观存在线程可能在没有收到notify的情况下被唤醒或者在收到notify之后、真正重新获得锁之前其他线程已经把条件再次改坏了。举例来说两个消费者同时等在信号量上此时count1。生产者调用signal把count变成2并唤醒两个线程。理论上两个线程都能拿到资源但如果某个实现细节有偏差或者遇到spurious wakeup一个线程醒来看count已经变成0却因为用的是if判断已经跳过了检查照样执行--count_直接把计数打成-1。这个bug就像幽灵写几十遍都不一定复现一次但只要线上出了就是大事故。5.2 阻塞线程与析构冲突Semaphore的析构函数我没有写自定义逻辑。如果某个线程还阻塞在这个信号量的wait上而Semaphore对象先被销毁了那这个等待的线程就会访问一个已经释放的mutex和condition_variable行为未定义大概率直接崩溃。这个坑特别隐蔽因为很多人的Semaphore是全局对象或者静态对象生命周期到程序结束才结束不会暴露。但一旦你把它作为局部对象用或者放在某个作用域里就得格外小心。我的经验是绝不销毁一个可能还有线程在等待的信号量。如果确实需要析构要确保所有可能调用wait的线程都已经join了或者通过某种机制比如先调用signal_all让它们醒来然后用一个退出标志让它们自行退出排空等待者。与其让析构函数去背这个锅不如在设计上保证使用边界清晰。5.3 notify_one还是notify_all什么时候会错过唤醒signal()里我用的notify_one()一次只唤醒一个等待线程。这符合信号量语义release一个资源只应该让一个线程获得许可。但有一个特殊情况如果程序逻辑依赖“所有等待线程必须都醒过来看看状态”notify_one()就会出问题。典型例子是多个线程等同一个one-shot事件此时应该用signal_all()配合count加1。怎么选我的经验法则是如果一次release只对应一个许可证用notify_one如果你的场景本质是“广播一个事件”用notify_all。选错了不会立刻崩溃但会出现“逻辑上该醒的人没醒”的偶发问题非常难查。5.4 用原子变量自旋的替代实现性能取舍条件变量版信号量最大的开销是等待线程会被内核挂起信号一到再被唤醒这个过程有上下文切换开销。如果在高吞吐、临界区极短的场景下比如保护一个计数器的加减用这么重量的机制有点杀鸡用牛刀。这时可以换成自旋版信号量用std::atomicint实现class SpinSemaphore { public: explicit SpinSemaphore(int count 0) : count_(count) {} void wait() { while (true) { int old count_.load(std::memory_order_acquire); if (old 0 count_.compare_exchange_weak(old, old - 1, std::memory_order_acq_rel, std::memory_order_acquire)) { break; } std::this_thread::yield(); } } void signal() { count_.fetch_add(1, std::memory_order_release); } private: std::atomicint count_; };自旋版本在低竞争下性能很好因为没有系统调用但高竞争下会让CPU空转反而更差。我的原则是如果临界区操作耗时超过几个微秒或者等待时间可能很长用条件变量版如果是极短的保护性操作用自旋版。默认无脑推荐条件变量版它更安全可预测性更好。5.5 一次线上问题的排查实录有一次我负责的模块出现线程卡死现象是某个业务请求迟迟不返回最后超时。查CPU占用不忙查线程栈发现一堆线程阻塞在同一个信号量的wait上。初步怀疑是某个signal没调用或者调用次数不够。后来仔细对了一遍业务逻辑发现有两个分支都该调signal但其中一个分支在异常提前返回时漏掉了一次signal导致许可证少发一个。修复后加上一个“兜底”用RAII包装信号量的获取与释放保证无论正常路径还是异常路径signal都会执行。这个经验告诉我使用信号量时释放动作最好封装在RAII对象里不要裸调signal。就像mutex要用lock_guard一样信号量的P/V操作也应该工具化。我还经历过一个更诡异的场景某个信号量初始化时传了负数构造函数做了保护变0但调用方可能是在数组越界写坏了内存之后才负数的。那个问题最后是通过ASAN工具定位出来的跟信号量类本身没关系。但这也提醒我排查信号量相关bug的时候不要只盯着信号量本身内存问题也可能以信号量异常的方式表现出来。6. C20来了之后手写的类还有什么用6.1 标准库counting_semaphore与手写版的差异C20提供了std::counting_semaphore用法上跟手写版很接近acquire()对应waitrelease()对应signaltry_acquire()对应try_waittry_acquire_for()对应wait_for。标准版的实现可以充分利用平台特性比如在支持的平台上直接用原子指令和futex性能上会比通用条件变量方案更优。标准版的release还可以传入指定数量的许可证这也是我手写版没有提供的能力。如果你的项目已经升级到C20确实没有理由不用标准库。手写版的适用场景非常明确还在维护C11/14/17的老代码库或者你不想为一个小功能引入第三方库再或者你对信号量的语义有特殊定制需求。6.2 手写版的价值与扩展思路即便将来所有项目都升级到C20理解手写信号量的过程仍然有价值。它让你看清条件变量和mutex是怎么组合成更高级同步原语的这种抽象能力在写其他并发组件时完全可以复用。在手写版的基础上还可以按需扩展几个能力。一个是“槽位上限”版经典信号量只限制下限不小于0但有些场景希望计数也不能超过某个上限在signal里判断count_ max_count_超了就丢弃或者阻塞这个用于限制通知积压非常有用。另一个是“许可证批量获取”版一次wait拿走N个许可证用于任务需要连续占用多个资源的场景。还有一个是增加一个count()查询接口返回当前许可证数用于监控和统计。这些扩展在你手写的时候想加就加拿到标准库反而得绕开接口设计。回到开头说的那个问题C11没有信号量怎么办。答案不是“忍一忍”也不是“引第三方依赖”而是“自己写一个顺便彻底搞懂它”。这个二十行的类几乎是我写过性价比最高的并发工具也希望它能在你的项目里少踩几个坑。