1. 项目概述为什么要在C17里自己造信号量信号量Semaphore这个概念对搞并发编程的朋友来说绝对不陌生。它就像十字路口的红绿灯控制着多个线程或进程对共享资源的访问流量。在C20标准之前标准库里是没有信号量这个“红绿灯”的我们得用互斥锁mutex和条件变量condition_variable自己搭一个或者依赖操作系统提供的API比如POSIX的sem_t。直到C20semaphore头文件才姗姗来迟提供了std::counting_semaphore和std::binary_semaphore。那问题来了既然C20都有了为什么还要在C17的环境下自己模拟实现一个呢原因很实在项目环境约束和深入理解原理。很多现有的大型项目或嵌入式平台编译器可能还停留在支持C17甚至更早的版本短期内无法升级到C20。这时候一个轻量级、可移植、行为符合预期的信号量实现就成了刚需。更重要的是自己动手实现一遍是理解信号量内部工作机制、线程同步精髓以及C并发编程陷阱的最佳途径。这比单纯调用semaphore.acquire()要深刻得多。今天要聊的就是如何在C17的环境下仅使用标准库的mutex和condition_variable打造一个行为与C20标准信号量高度一致的CountingSemaphore。我们会从最核心的“许可证”模型讲起一步步拆解acquire等绿灯和release亮绿灯的内部逻辑处理超时和中断并深入探讨那些教科书上不会写的、真正在实战中会踩到的坑比如虚假唤醒、资源泄漏和优先级反转的潜在风险。无论你是正在学习多线程的初学者还是被老旧编译器版本困扰的开发者这篇长文都能给你一份可以直接“抄作业”的、工业级的实现方案。2. 信号量核心原理与设计选型在动手写代码之前我们必须把信号量到底是个啥以及为什么选择特定的实现方式搞清楚。这决定了我们代码的骨架是否健壮。2.1 信号量本质一个计数器一个等待队列你可以把信号量想象成一个管理着若干张“许可证”的池子。这个池子的核心是一个非负整数计数器count。线程想要访问共享资源必须先从这个池子里“获取”acquire一张许可证。如果池子里有许可证count 0线程就直接拿走一张计数器减1然后继续执行畅通无阻。如果池子里没许可证了count 0这个线程就会被阻塞进入一个等待队列里排队直到有其他线程“释放”release许可证回来计数器加1并通知等待队列中的线程。这里的关键在于信号量允许多个线程同时持有许可证具体数量由初始计数器值决定它控制的是“并发访问的数量”而不是“互斥访问”。这与互斥锁Mutex有本质区别互斥锁任何时候只允许一个线程进入临界区是“独占”模式而信号量是“限量共享”模式。二进制信号量初始值为1在行为上虽然和互斥锁很像但它的语义是“有/无许可证”而非“锁的所有权”因此通常不用于保护临界区而多用于线程间的事件通知或任务同步。2.2 为什么用mutexcondition_variable来模拟C17标准库里能用于线程同步的核心工具就是std::mutex互斥锁和std::condition_variable条件变量。用它们来模拟信号量是经典且可靠的做法。std::mutex用来保护对内部计数器count的访问。任何读取或修改count的操作都必须在锁的保护下进行确保原子性防止数据竞争。std::condition_variable用来实现线程的等待和通知。当线程尝试获取许可证但发现count 0时它需要在这个条件变量上等待wait。当其他线程释放许可证后它通过条件变量通知notify_one或notify_all一个或所有等待的线程醒来检查条件。这个组合非常契合信号量“检查条件-等待-被通知”的工作模型。为什么不直接用原子操作std::atomic自旋等待呢因为自旋等待忙等待在获取不到资源时会持续占用CPU在等待时间可能较长的情况下性能损耗极大。而condition_variable会让等待的线程挂起不消耗CPU周期是更高效的选择。2.3 接口设计向C20标准看齐为了让我们的实现有更好的可移植性和未来兼容性接口设计上应该尽量向C20的std::counting_semaphore靠拢。核心接口就两个acquire(): 获取一个许可证。如果没有许可证则阻塞当前线程直到有许可证可用。release(ptrdiff_t update 1): 释放一个或多个许可证增加内部计数器并可能唤醒等待的线程。此外一些非阻塞或带超时的接口也非常实用 3.try_acquire(): 尝试获取许可证立即返回成功或失败不阻塞。 4.try_acquire_for(const std::chrono::duration rel_time): 在指定相对时间段内尝试获取超时则返回失败。 5.try_acquire_until(const std::chrono::time_point abs_time): 在指定绝对时间点前尝试获取。我们还会提供一个构造函数允许用户指定信号量的初始许可证数量。一个完整的、可复用的类框架就在我们脑中成型了。3. 核心实现拆解从骨架到血肉有了清晰的设计图我们现在开始搭建CountingSemaphore这个类。我会把代码分成几个部分并逐一解释每个细节的考量。3.1 类的基本骨架与成员变量首先我们定义类并声明其私有成员。这是整个信号量的状态核心。#include mutex #include condition_variable #include chrono #include cstdint class CountingSemaphore { public: // 构造函数初始化许可证数量 explicit CountingSemaphore(ptrdiff_t desired 0); // 禁止拷贝和赋值 CountingSemaphore(const CountingSemaphore) delete; CountingSemaphore operator(const CountingSemaphore) delete; // 核心接口 void acquire(); bool try_acquire(); bool try_acquire_for(const std::chrono::milliseconds rel_time); bool try_acquire_until(const std::chrono::system_clock::time_point abs_time); void release(ptrdiff_t update 1); private: mutable std::mutex mutex_; // 保护内部状态 std::condition_variable cv_; // 用于线程等待/通知 ptrdiff_t count_; // 当前可用的许可证数量 };要点解析mutable std::mutex mutex_: 互斥锁声明为mutable是因为它需要在const成员函数如try_acquire它不修改count_但需要锁保护中被修改加锁/解锁。这是一个常见的技巧。ptrdiff_t count_: 计数器类型选择ptrdiff_t有符号整型是为了与C20标准保持一致并且可以方便地检查溢出尽管我们会在release中做保护。初始值desired可以是0表示初始无许可证常用于线程执行的先后顺序控制。删除拷贝构造和赋值运算符这是并发编程中资源管理类的标准操作。信号量内部包含互斥锁和条件变量这些对象拷贝是没有意义的而且极易导致难以追踪的bug。必须显式禁止。3.2 构造与析构资源初始化的陷阱构造函数非常简单就是初始化计数器。但这里有一个重要的设计决策初始许可证数量desired是否允许为负数C20标准规定std::counting_semaphore的模板参数least_max_value必须大于0构造函数中的desired必须在[0, least_max_value]区间。为了安全和语义清晰我们的实现也应当强制要求desired 0。可以在构造函数中加一个断言assert。explicit CountingSemaphore(ptrdiff_t desired 0) : count_(desired) { // 确保初始值非负这是一个健壮性检查 // 在实际生产代码中可能会选择抛出异常但assert在调试时非常有用。 // assert(desired 0); }关于析构函数我们不需要显式定义。编译器生成的默认析构函数会依次析构cv_和mutex_。这里有一个至关重要的注意事项当信号量对象析构时如果还有线程在acquire上等待即阻塞在cv_.wait会发生什么行为是未定义的通常会导致程序崩溃。因此类的使用者必须确保在析构信号量时没有线程还在等待它。这是RAII资源获取即初始化原则在并发场景下的一个延伸生命周期管理必须同步。在实际项目中这通常通过更上层的逻辑如线程池关闭序列来保证。3.3acquire()的实现等待的艺术这是信号量最核心的阻塞获取操作。逻辑是锁住互斥量检查条件count_ 0如果条件不满足就释放锁并进入等待直到被其他线程唤醒后再次检查条件。void CountingSemaphore::acquire() { std::unique_lockstd::mutex lock(mutex_); // 使用条件变量的wait方法并传入一个lambda谓词 cv_.wait(lock, [this]() { return count_ 0; }); // 当wait返回时我们一定持有锁并且 count_ 0 --count_; }这里是第一个“坑”和精髓所在为什么wait需要用一个lambda表达式condition_variable::wait在C11之后有一个重载版本接受一个谓词Predicate。它的内部行为等价于while (!predicate()) { wait(lock); }这个while循环至关重要它解决了虚假唤醒Spurious Wakeup问题。操作系统可能因为某些原因如信号中断将等待的线程唤醒即使没有其他线程调用notify。如果只用cv_.wait(lock)线程被虚假唤醒后会直接往下执行此时count_可能依然为0导致逻辑错误。而带有谓词的wait会在每次唤醒后自动重新检查条件count_ 0只有条件真正满足时才会跳出循环从而保证了正确性。这是使用条件变量时必须牢记的铁律。3.4release()的实现通知的智慧释放许可证的操作相对直接增加计数器然后通知等待的线程。void CountingSemaphore::release(ptrdiff_t update) { if (update 0) { // 通常release一个非正数是没有意义的可以抛出异常或忽略。 // 为了健壮性我们至少应该处理一下。这里选择忽略或断言。 // throw std::invalid_argument(release update must be positive); return; } std::lock_guardstd::mutex lock(mutex_); // 注意这里存在溢出的风险ptrdiff_t是有范围的。 // 一个简单的保护是如果溢出就设置为最大值或饱和加法。 // 为了简单我们先不做复杂处理但心里要知道这个风险。 count_ update; // 通知等待的线程 if (update 1) { cv_.notify_one(); // 只唤醒一个等待线程 } else { cv_.notify_all(); // 唤醒所有等待线程 } }关键决策点用notify_one()还是notify_all()如果每次只释放一个许可证update 1那么唤醒一个等待线程就是最高效的因为被唤醒的线程会消耗掉这个许可证其他线程即使被唤醒也拿不到又会继续等待造成不必要的上下文切换开销。如果一次释放了多个许可证update 1那么唤醒所有等待线程是合理的因为可能有多个线程可以同时获取到许可证。但这里也有优化空间你可以选择notify_all()也可以选择调用update次notify_one()。前者更简单但可能唤醒过多线程后者更精细但调用次数多。在大多数场景下notify_all()在释放多个许可证时是标准做法。我们的实现采用了条件判断这是一种兼顾效率和正确性的策略。3.5 非阻塞与超时获取try_acquire系列阻塞获取是基础但非阻塞或限时等待在实际应用中同样重要它可以防止线程死锁提高系统响应性。3.5.1 立即尝试try_acquire()bool CountingSemaphore::try_acquire() { std::lock_guardstd::mutex lock(mutex_); if (count_ 0) { --count_; return true; } return false; }这个很简单就是检查-获取的原子操作。注意它也需要加锁因为检查和修改count_必须是原子的。3.5.2 超时获取try_acquire_for/until这是acquire()的增强版引入了超时机制。我们需要使用条件变量的wait_for或wait_until方法。bool CountingSemaphore::try_acquire_for(const std::chrono::milliseconds rel_time) { std::unique_lockstd::mutex lock(mutex_); // wait_for 返回一个状态表示是超时还是被唤醒 // 谓词条件依然是 count_ 0 if (cv_.wait_for(lock, rel_time, [this]() { return count_ 0; })) { // 谓词为真说明在超时前成功获取了锁且count_0 --count_; return true; } // 超时返回false return false; } bool CountingSemaphore::try_acquire_until(const std::chrono::system_clock::time_point abs_time) { std::unique_lockstd::mutex lock(mutex_); if (cv_.wait_until(lock, abs_time, [this]() { return count_ 0; })) { --count_; return true; } return false; }实操心得wait_for的返回值cv_.wait_for在返回时有两种情况1) 谓词条件变为真被notify唤醒且count_02) 超时。它的返回值直接就是谓词条件的最终状态。这个设计非常贴心我们直接判断返回值即可无需再额外检查count_或系统时间。一定要使用这个带谓词的版本避免自己手动处理复杂的超时和虚假唤醒逻辑。4. 完整代码实现与测试用例将上述所有部分组合起来我们就得到了一个完整的CountingSemaphore类。为了节省篇幅这里列出整合后的头文件和一个简单的测试用例。counting_semaphore.hpp#ifndef COUNTING_SEMAPHORE_HPP #define COUNTING_SEMAPHORE_HPP #include mutex #include condition_variable #include chrono #include cstdint class CountingSemaphore { public: explicit CountingSemaphore(ptrdiff_t desired 0) : count_(desired) {} CountingSemaphore(const CountingSemaphore) delete; CountingSemaphore operator(const CountingSemaphore) delete; void acquire() { std::unique_lockstd::mutex lock(mutex_); cv_.wait(lock, [this]() { return count_ 0; }); --count_; } bool try_acquire() { std::lock_guardstd::mutex lock(mutex_); if (count_ 0) { --count_; return true; } return false; } templateclass Rep, class Period bool try_acquire_for(const std::chrono::durationRep, Period rel_time) { std::unique_lockstd::mutex lock(mutex_); if (cv_.wait_for(lock, rel_time, [this]() { return count_ 0; })) { --count_; return true; } return false; } templateclass Clock, class Duration bool try_acquire_until(const std::chrono::time_pointClock, Duration abs_time) { std::unique_lockstd::mutex lock(mutex_); if (cv_.wait_until(lock, abs_time, [this]() { return count_ 0; })) { --count_; return true; } return false; } void release(ptrdiff_t update 1) { if (update 0) return; // 简单处理非法参数 std::lock_guardstd::mutex lock(mutex_); count_ update; if (update 1) { cv_.notify_one(); } else { cv_.notify_all(); } } private: mutable std::mutex mutex_; std::condition_variable cv_; ptrdiff_t count_ 0; }; #endif // COUNTING_SEMAPHORE_HPP简单的测试用例test_semaphore.cpp这个测试模拟了一个经典的生产者-消费者场景其中信号量用于控制对有限大小缓冲区的访问。#include counting_semaphore.hpp #include iostream #include vector #include thread #include chrono #include queue #include mutex int main() { constexpr size_t buffer_size 5; constexpr int total_items 20; std::queueint buffer; std::mutex buffer_mutex; // 保护buffer队列 CountingSemaphore empty_slots(buffer_size); // 初始空槽数量为缓冲区大小 CountingSemaphore full_slots(0); // 初始满槽数量为0 auto producer [](int id) { for (int i 0; i total_items / 2; i) { // 两个生产者各生产一半 int item id * 100 i; empty_slots.acquire(); // 等待空槽 { std::lock_guardstd::mutex lock(buffer_mutex); buffer.push(item); std::cout Producer id produced: item (Buffer size: buffer.size() )\n; } full_slots.release(); // 增加一个满槽通知消费者 std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 模拟生产耗时 } }; auto consumer [](int id) { for (int i 0; i total_items / 2; i) { // 两个消费者各消费一半 full_slots.acquire(); // 等待满槽 int item; { std::lock_guardstd::mutex lock(buffer_mutex); item buffer.front(); buffer.pop(); std::cout Consumer id consumed: item (Buffer size: buffer.size() )\n; } empty_slots.release(); // 增加一个空槽通知生产者 std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟消费耗时 } }; std::thread p1(producer, 1); std::thread p2(producer, 2); std::thread c1(consumer, 1); std::thread c2(consumer, 2); p1.join(); p2.join(); c1.join(); c2.join(); std::cout All producers and consumers finished. Buffer size: buffer.size() std::endl; return 0; }编译并运行这个测试例如g -stdc17 -pthread test_semaphore.cpp -o test ./test你会看到生产者和消费者交替工作缓冲区大小始终在0到5之间波动完美演示了信号量如何协调多个线程对有限资源的并发访问。5. 进阶探讨、性能考量与避坑指南实现一个能跑的信号量只是第一步。要让它在生产环境中稳定可靠我们还需要思考更多。5.1 与C20标准信号量的差异我们的实现与std::counting_semaphore在核心行为上是一致的但也存在一些差异了解这些差异有助于正确使用。最大计数值Least Max Value: C20的std::counting_semaphore是一个模板类templatestd::ptrdiff_t LeastMaxValue /* implementation-defined */ class counting_semaphore;。这个LeastMaxValue是一个编译期常量规定了信号量计数器可能的最大值或至少能达到的最大值。我们的实现没有这个上限除了ptrdiff_t类型的自然上限更灵活但也少了编译期的边界检查。溢出处理: C20标准规定release操作导致计数器超过LeastMaxValue是未定义行为。我们的简单实现在release一个很大的数时会导致count_溢出有符号整数溢出是未定义行为。一个工业级的实现应该进行饱和处理或抛出异常。例如void release(ptrdiff_t update 1) { if (update 0) return; std::lock_guardstd::mutex lock(mutex_); if (count_ (std::numeric_limitsptrdiff_t::max() - update)) { // 处理溢出可以饱和到最大值或者抛出std::overflow_error count_ std::numeric_limitsptrdiff_t::max(); } else { count_ update; } // ... 通知逻辑 }try_acquire的 noexcept 规范: C20的标准接口有更严格的异常规范。我们的实现因为使用了互斥锁和条件变量这些操作理论上可能抛出异常如系统资源不足所以不能标记为noexcept。5.2 性能瓶颈与优化思路mutexcondition_variable的实现虽然正确但在高并发、竞争激烈的场景下锁的争用可能成为性能瓶颈。每一次acquire和release都要获取同一把mutex_。优化方向1无锁Lock-free或原子操作。可以尝试使用std::atomic配合std::atomic::wait/notify_oneC20来实现一个无锁的信号量。这在竞争极端激烈时可能有优势但实现复杂度高且atomic::wait在C20才引入不符合我们C17的前提。在C17下用原子操作实现无阻塞的try_acquire很容易但要实现阻塞的acquire最终往往还是需要回退到类似条件变量的同步原语如futex或自旋等待而自旋等待在等待时间长时性能很差。优化方向2减少锁粒度。我们的实现中计数器count_的修改和条件变量的操作都在同一把锁下。这是标准的、正确的做法。对于大多数应用场景这个性能已经足够。如果确实遇到瓶颈首先要分析的是信号量本身的使用模式是否合理比如是否过度同步而不是盲目优化这个底层原语。结论对于绝大多数应用我们实现的这个基于锁的信号量在性能和正确性上已经取得了很好的平衡。“先确保正确再考虑优化”是并发编程的第一准则。5.3 常见陷阱与实战经验忘记配对使用Acquire/Release不匹配这是最常见的错误。每个acquire都必须对应一个release否则许可证会逐渐耗尽导致线程永久阻塞。在复杂的代码路径中尤其是存在异常或多个返回点务必使用RAII包装器。例如可以写一个SemaphoreGuard类在构造函数中acquire在析构函数中release确保异常安全。class SemaphoreGuard { public: explicit SemaphoreGuard(CountingSemaphore sem) : sem_(sem) { sem_.acquire(); } ~SemaphoreGuard() { sem_.release(); } // 禁止拷贝和移动 SemaphoreGuard(const SemaphoreGuard) delete; SemaphoreGuard operator(const SemaphoreGuard) delete; private: CountingSemaphore sem_; }; // 使用 { SemaphoreGuard guard(my_semaphore); // 自动acquire // ... 访问共享资源 } // 离开作用域guard析构自动release错误选择notify_one与notify_all如前所述在release(1)时使用notify_all是低效的它会导致“惊群效应”thundering herd所有等待线程都被唤醒去争夺一个许可证最终只有一个能成功其他线程又得回去睡觉产生大量无用的上下文切换。务必根据释放的许可证数量决定通知方式。信号量不是万能的信号量适合控制对一组完全相同资源的访问如连接池、线程池、缓冲区槽位。对于保护一段临界区代码即确保同一时间只有一个线程执行某段代码互斥锁std::mutex是更合适、语义更清晰的选择。虽然二进制信号量初始值为1可以达到类似效果但互斥锁具有“所有权”概念可以更好地与std::lock_guard、std::unique_lock等RAII机制配合并且一些调试工具能识别死锁与互斥锁的关系。死锁和互斥锁一样信号量也可能导致死锁。例如线程A持有信号量S1等待S2线程B持有S2等待S1。避免死锁的经典原则同样适用固定顺序获取资源、使用超时获取try_acquire_for、或者使用更高级的并发设施。生命周期管理再次强调确保信号量对象的生命周期长于所有使用它的线程。不要在还有线程等待时销毁信号量对象。在设计上通常将信号量作为全局对象、类的成员其生命周期与类实例绑定或通过std::shared_ptr进行共享管理。6. 扩展二进制信号量与更复杂的同步模式我们的CountingSemaphore是一个通用计数信号量。二进制信号量是它的一个特例初始值为1。你可以简单地用它来构建using BinarySemaphore CountingSemaphore; // 在构造时传入1 // 或者专门封装一下 class BinarySemaphore : public CountingSemaphore { public: BinarySemaphore() : CountingSemaphore(1) {} // 可以隐藏release的多参数版本因为二进制信号量通常只release(1) void release() { CountingSemaphore::release(1); } };此外信号量可以作为构建更复杂同步模式的基础模块例如读者-写者锁可以用两个信号量一个用于写者互斥一个用于控制读者计数来实现优先考虑读者或写者。屏障Barrier让一组线程在某个点同步等待所有线程到达后才能继续。有限资源池正如我们测试用例中的缓冲区信号量完美地建模了“空槽”和“满槽”的数量。实现这些模式是对信号量理解的很好练习。它们背后的核心思想依然是对共享状态的原子检查和条件等待。通过亲手实现这个C17信号量你获得的不只是一个可用的工具更重要的是对底层同步机制深刻、直观的理解这份理解在调试复杂的并发问题时将是无价之宝。当你下次再使用std::semaphore或者任何语言中的信号量时你都能清晰地看到它内部那个忙碌的计数器和那些在条件变量上安静等待的线程队列。
1. 项目概述为什么要在C17里自己造信号量信号量Semaphore这个概念对搞并发编程的朋友来说绝对不陌生。它就像十字路口的红绿灯控制着多个线程或进程对共享资源的访问流量。在C20标准之前标准库里是没有信号量这个“红绿灯”的我们得用互斥锁mutex和条件变量condition_variable自己搭一个或者依赖操作系统提供的API比如POSIX的sem_t。直到C20semaphore头文件才姗姗来迟提供了std::counting_semaphore和std::binary_semaphore。那问题来了既然C20都有了为什么还要在C17的环境下自己模拟实现一个呢原因很实在项目环境约束和深入理解原理。很多现有的大型项目或嵌入式平台编译器可能还停留在支持C17甚至更早的版本短期内无法升级到C20。这时候一个轻量级、可移植、行为符合预期的信号量实现就成了刚需。更重要的是自己动手实现一遍是理解信号量内部工作机制、线程同步精髓以及C并发编程陷阱的最佳途径。这比单纯调用semaphore.acquire()要深刻得多。今天要聊的就是如何在C17的环境下仅使用标准库的mutex和condition_variable打造一个行为与C20标准信号量高度一致的CountingSemaphore。我们会从最核心的“许可证”模型讲起一步步拆解acquire等绿灯和release亮绿灯的内部逻辑处理超时和中断并深入探讨那些教科书上不会写的、真正在实战中会踩到的坑比如虚假唤醒、资源泄漏和优先级反转的潜在风险。无论你是正在学习多线程的初学者还是被老旧编译器版本困扰的开发者这篇长文都能给你一份可以直接“抄作业”的、工业级的实现方案。2. 信号量核心原理与设计选型在动手写代码之前我们必须把信号量到底是个啥以及为什么选择特定的实现方式搞清楚。这决定了我们代码的骨架是否健壮。2.1 信号量本质一个计数器一个等待队列你可以把信号量想象成一个管理着若干张“许可证”的池子。这个池子的核心是一个非负整数计数器count。线程想要访问共享资源必须先从这个池子里“获取”acquire一张许可证。如果池子里有许可证count 0线程就直接拿走一张计数器减1然后继续执行畅通无阻。如果池子里没许可证了count 0这个线程就会被阻塞进入一个等待队列里排队直到有其他线程“释放”release许可证回来计数器加1并通知等待队列中的线程。这里的关键在于信号量允许多个线程同时持有许可证具体数量由初始计数器值决定它控制的是“并发访问的数量”而不是“互斥访问”。这与互斥锁Mutex有本质区别互斥锁任何时候只允许一个线程进入临界区是“独占”模式而信号量是“限量共享”模式。二进制信号量初始值为1在行为上虽然和互斥锁很像但它的语义是“有/无许可证”而非“锁的所有权”因此通常不用于保护临界区而多用于线程间的事件通知或任务同步。2.2 为什么用mutexcondition_variable来模拟C17标准库里能用于线程同步的核心工具就是std::mutex互斥锁和std::condition_variable条件变量。用它们来模拟信号量是经典且可靠的做法。std::mutex用来保护对内部计数器count的访问。任何读取或修改count的操作都必须在锁的保护下进行确保原子性防止数据竞争。std::condition_variable用来实现线程的等待和通知。当线程尝试获取许可证但发现count 0时它需要在这个条件变量上等待wait。当其他线程释放许可证后它通过条件变量通知notify_one或notify_all一个或所有等待的线程醒来检查条件。这个组合非常契合信号量“检查条件-等待-被通知”的工作模型。为什么不直接用原子操作std::atomic自旋等待呢因为自旋等待忙等待在获取不到资源时会持续占用CPU在等待时间可能较长的情况下性能损耗极大。而condition_variable会让等待的线程挂起不消耗CPU周期是更高效的选择。2.3 接口设计向C20标准看齐为了让我们的实现有更好的可移植性和未来兼容性接口设计上应该尽量向C20的std::counting_semaphore靠拢。核心接口就两个acquire(): 获取一个许可证。如果没有许可证则阻塞当前线程直到有许可证可用。release(ptrdiff_t update 1): 释放一个或多个许可证增加内部计数器并可能唤醒等待的线程。此外一些非阻塞或带超时的接口也非常实用 3.try_acquire(): 尝试获取许可证立即返回成功或失败不阻塞。 4.try_acquire_for(const std::chrono::duration rel_time): 在指定相对时间段内尝试获取超时则返回失败。 5.try_acquire_until(const std::chrono::time_point abs_time): 在指定绝对时间点前尝试获取。我们还会提供一个构造函数允许用户指定信号量的初始许可证数量。一个完整的、可复用的类框架就在我们脑中成型了。3. 核心实现拆解从骨架到血肉有了清晰的设计图我们现在开始搭建CountingSemaphore这个类。我会把代码分成几个部分并逐一解释每个细节的考量。3.1 类的基本骨架与成员变量首先我们定义类并声明其私有成员。这是整个信号量的状态核心。#include mutex #include condition_variable #include chrono #include cstdint class CountingSemaphore { public: // 构造函数初始化许可证数量 explicit CountingSemaphore(ptrdiff_t desired 0); // 禁止拷贝和赋值 CountingSemaphore(const CountingSemaphore) delete; CountingSemaphore operator(const CountingSemaphore) delete; // 核心接口 void acquire(); bool try_acquire(); bool try_acquire_for(const std::chrono::milliseconds rel_time); bool try_acquire_until(const std::chrono::system_clock::time_point abs_time); void release(ptrdiff_t update 1); private: mutable std::mutex mutex_; // 保护内部状态 std::condition_variable cv_; // 用于线程等待/通知 ptrdiff_t count_; // 当前可用的许可证数量 };要点解析mutable std::mutex mutex_: 互斥锁声明为mutable是因为它需要在const成员函数如try_acquire它不修改count_但需要锁保护中被修改加锁/解锁。这是一个常见的技巧。ptrdiff_t count_: 计数器类型选择ptrdiff_t有符号整型是为了与C20标准保持一致并且可以方便地检查溢出尽管我们会在release中做保护。初始值desired可以是0表示初始无许可证常用于线程执行的先后顺序控制。删除拷贝构造和赋值运算符这是并发编程中资源管理类的标准操作。信号量内部包含互斥锁和条件变量这些对象拷贝是没有意义的而且极易导致难以追踪的bug。必须显式禁止。3.2 构造与析构资源初始化的陷阱构造函数非常简单就是初始化计数器。但这里有一个重要的设计决策初始许可证数量desired是否允许为负数C20标准规定std::counting_semaphore的模板参数least_max_value必须大于0构造函数中的desired必须在[0, least_max_value]区间。为了安全和语义清晰我们的实现也应当强制要求desired 0。可以在构造函数中加一个断言assert。explicit CountingSemaphore(ptrdiff_t desired 0) : count_(desired) { // 确保初始值非负这是一个健壮性检查 // 在实际生产代码中可能会选择抛出异常但assert在调试时非常有用。 // assert(desired 0); }关于析构函数我们不需要显式定义。编译器生成的默认析构函数会依次析构cv_和mutex_。这里有一个至关重要的注意事项当信号量对象析构时如果还有线程在acquire上等待即阻塞在cv_.wait会发生什么行为是未定义的通常会导致程序崩溃。因此类的使用者必须确保在析构信号量时没有线程还在等待它。这是RAII资源获取即初始化原则在并发场景下的一个延伸生命周期管理必须同步。在实际项目中这通常通过更上层的逻辑如线程池关闭序列来保证。3.3acquire()的实现等待的艺术这是信号量最核心的阻塞获取操作。逻辑是锁住互斥量检查条件count_ 0如果条件不满足就释放锁并进入等待直到被其他线程唤醒后再次检查条件。void CountingSemaphore::acquire() { std::unique_lockstd::mutex lock(mutex_); // 使用条件变量的wait方法并传入一个lambda谓词 cv_.wait(lock, [this]() { return count_ 0; }); // 当wait返回时我们一定持有锁并且 count_ 0 --count_; }这里是第一个“坑”和精髓所在为什么wait需要用一个lambda表达式condition_variable::wait在C11之后有一个重载版本接受一个谓词Predicate。它的内部行为等价于while (!predicate()) { wait(lock); }这个while循环至关重要它解决了虚假唤醒Spurious Wakeup问题。操作系统可能因为某些原因如信号中断将等待的线程唤醒即使没有其他线程调用notify。如果只用cv_.wait(lock)线程被虚假唤醒后会直接往下执行此时count_可能依然为0导致逻辑错误。而带有谓词的wait会在每次唤醒后自动重新检查条件count_ 0只有条件真正满足时才会跳出循环从而保证了正确性。这是使用条件变量时必须牢记的铁律。3.4release()的实现通知的智慧释放许可证的操作相对直接增加计数器然后通知等待的线程。void CountingSemaphore::release(ptrdiff_t update) { if (update 0) { // 通常release一个非正数是没有意义的可以抛出异常或忽略。 // 为了健壮性我们至少应该处理一下。这里选择忽略或断言。 // throw std::invalid_argument(release update must be positive); return; } std::lock_guardstd::mutex lock(mutex_); // 注意这里存在溢出的风险ptrdiff_t是有范围的。 // 一个简单的保护是如果溢出就设置为最大值或饱和加法。 // 为了简单我们先不做复杂处理但心里要知道这个风险。 count_ update; // 通知等待的线程 if (update 1) { cv_.notify_one(); // 只唤醒一个等待线程 } else { cv_.notify_all(); // 唤醒所有等待线程 } }关键决策点用notify_one()还是notify_all()如果每次只释放一个许可证update 1那么唤醒一个等待线程就是最高效的因为被唤醒的线程会消耗掉这个许可证其他线程即使被唤醒也拿不到又会继续等待造成不必要的上下文切换开销。如果一次释放了多个许可证update 1那么唤醒所有等待线程是合理的因为可能有多个线程可以同时获取到许可证。但这里也有优化空间你可以选择notify_all()也可以选择调用update次notify_one()。前者更简单但可能唤醒过多线程后者更精细但调用次数多。在大多数场景下notify_all()在释放多个许可证时是标准做法。我们的实现采用了条件判断这是一种兼顾效率和正确性的策略。3.5 非阻塞与超时获取try_acquire系列阻塞获取是基础但非阻塞或限时等待在实际应用中同样重要它可以防止线程死锁提高系统响应性。3.5.1 立即尝试try_acquire()bool CountingSemaphore::try_acquire() { std::lock_guardstd::mutex lock(mutex_); if (count_ 0) { --count_; return true; } return false; }这个很简单就是检查-获取的原子操作。注意它也需要加锁因为检查和修改count_必须是原子的。3.5.2 超时获取try_acquire_for/until这是acquire()的增强版引入了超时机制。我们需要使用条件变量的wait_for或wait_until方法。bool CountingSemaphore::try_acquire_for(const std::chrono::milliseconds rel_time) { std::unique_lockstd::mutex lock(mutex_); // wait_for 返回一个状态表示是超时还是被唤醒 // 谓词条件依然是 count_ 0 if (cv_.wait_for(lock, rel_time, [this]() { return count_ 0; })) { // 谓词为真说明在超时前成功获取了锁且count_0 --count_; return true; } // 超时返回false return false; } bool CountingSemaphore::try_acquire_until(const std::chrono::system_clock::time_point abs_time) { std::unique_lockstd::mutex lock(mutex_); if (cv_.wait_until(lock, abs_time, [this]() { return count_ 0; })) { --count_; return true; } return false; }实操心得wait_for的返回值cv_.wait_for在返回时有两种情况1) 谓词条件变为真被notify唤醒且count_02) 超时。它的返回值直接就是谓词条件的最终状态。这个设计非常贴心我们直接判断返回值即可无需再额外检查count_或系统时间。一定要使用这个带谓词的版本避免自己手动处理复杂的超时和虚假唤醒逻辑。4. 完整代码实现与测试用例将上述所有部分组合起来我们就得到了一个完整的CountingSemaphore类。为了节省篇幅这里列出整合后的头文件和一个简单的测试用例。counting_semaphore.hpp#ifndef COUNTING_SEMAPHORE_HPP #define COUNTING_SEMAPHORE_HPP #include mutex #include condition_variable #include chrono #include cstdint class CountingSemaphore { public: explicit CountingSemaphore(ptrdiff_t desired 0) : count_(desired) {} CountingSemaphore(const CountingSemaphore) delete; CountingSemaphore operator(const CountingSemaphore) delete; void acquire() { std::unique_lockstd::mutex lock(mutex_); cv_.wait(lock, [this]() { return count_ 0; }); --count_; } bool try_acquire() { std::lock_guardstd::mutex lock(mutex_); if (count_ 0) { --count_; return true; } return false; } templateclass Rep, class Period bool try_acquire_for(const std::chrono::durationRep, Period rel_time) { std::unique_lockstd::mutex lock(mutex_); if (cv_.wait_for(lock, rel_time, [this]() { return count_ 0; })) { --count_; return true; } return false; } templateclass Clock, class Duration bool try_acquire_until(const std::chrono::time_pointClock, Duration abs_time) { std::unique_lockstd::mutex lock(mutex_); if (cv_.wait_until(lock, abs_time, [this]() { return count_ 0; })) { --count_; return true; } return false; } void release(ptrdiff_t update 1) { if (update 0) return; // 简单处理非法参数 std::lock_guardstd::mutex lock(mutex_); count_ update; if (update 1) { cv_.notify_one(); } else { cv_.notify_all(); } } private: mutable std::mutex mutex_; std::condition_variable cv_; ptrdiff_t count_ 0; }; #endif // COUNTING_SEMAPHORE_HPP简单的测试用例test_semaphore.cpp这个测试模拟了一个经典的生产者-消费者场景其中信号量用于控制对有限大小缓冲区的访问。#include counting_semaphore.hpp #include iostream #include vector #include thread #include chrono #include queue #include mutex int main() { constexpr size_t buffer_size 5; constexpr int total_items 20; std::queueint buffer; std::mutex buffer_mutex; // 保护buffer队列 CountingSemaphore empty_slots(buffer_size); // 初始空槽数量为缓冲区大小 CountingSemaphore full_slots(0); // 初始满槽数量为0 auto producer [](int id) { for (int i 0; i total_items / 2; i) { // 两个生产者各生产一半 int item id * 100 i; empty_slots.acquire(); // 等待空槽 { std::lock_guardstd::mutex lock(buffer_mutex); buffer.push(item); std::cout Producer id produced: item (Buffer size: buffer.size() )\n; } full_slots.release(); // 增加一个满槽通知消费者 std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 模拟生产耗时 } }; auto consumer [](int id) { for (int i 0; i total_items / 2; i) { // 两个消费者各消费一半 full_slots.acquire(); // 等待满槽 int item; { std::lock_guardstd::mutex lock(buffer_mutex); item buffer.front(); buffer.pop(); std::cout Consumer id consumed: item (Buffer size: buffer.size() )\n; } empty_slots.release(); // 增加一个空槽通知生产者 std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟消费耗时 } }; std::thread p1(producer, 1); std::thread p2(producer, 2); std::thread c1(consumer, 1); std::thread c2(consumer, 2); p1.join(); p2.join(); c1.join(); c2.join(); std::cout All producers and consumers finished. Buffer size: buffer.size() std::endl; return 0; }编译并运行这个测试例如g -stdc17 -pthread test_semaphore.cpp -o test ./test你会看到生产者和消费者交替工作缓冲区大小始终在0到5之间波动完美演示了信号量如何协调多个线程对有限资源的并发访问。5. 进阶探讨、性能考量与避坑指南实现一个能跑的信号量只是第一步。要让它在生产环境中稳定可靠我们还需要思考更多。5.1 与C20标准信号量的差异我们的实现与std::counting_semaphore在核心行为上是一致的但也存在一些差异了解这些差异有助于正确使用。最大计数值Least Max Value: C20的std::counting_semaphore是一个模板类templatestd::ptrdiff_t LeastMaxValue /* implementation-defined */ class counting_semaphore;。这个LeastMaxValue是一个编译期常量规定了信号量计数器可能的最大值或至少能达到的最大值。我们的实现没有这个上限除了ptrdiff_t类型的自然上限更灵活但也少了编译期的边界检查。溢出处理: C20标准规定release操作导致计数器超过LeastMaxValue是未定义行为。我们的简单实现在release一个很大的数时会导致count_溢出有符号整数溢出是未定义行为。一个工业级的实现应该进行饱和处理或抛出异常。例如void release(ptrdiff_t update 1) { if (update 0) return; std::lock_guardstd::mutex lock(mutex_); if (count_ (std::numeric_limitsptrdiff_t::max() - update)) { // 处理溢出可以饱和到最大值或者抛出std::overflow_error count_ std::numeric_limitsptrdiff_t::max(); } else { count_ update; } // ... 通知逻辑 }try_acquire的 noexcept 规范: C20的标准接口有更严格的异常规范。我们的实现因为使用了互斥锁和条件变量这些操作理论上可能抛出异常如系统资源不足所以不能标记为noexcept。5.2 性能瓶颈与优化思路mutexcondition_variable的实现虽然正确但在高并发、竞争激烈的场景下锁的争用可能成为性能瓶颈。每一次acquire和release都要获取同一把mutex_。优化方向1无锁Lock-free或原子操作。可以尝试使用std::atomic配合std::atomic::wait/notify_oneC20来实现一个无锁的信号量。这在竞争极端激烈时可能有优势但实现复杂度高且atomic::wait在C20才引入不符合我们C17的前提。在C17下用原子操作实现无阻塞的try_acquire很容易但要实现阻塞的acquire最终往往还是需要回退到类似条件变量的同步原语如futex或自旋等待而自旋等待在等待时间长时性能很差。优化方向2减少锁粒度。我们的实现中计数器count_的修改和条件变量的操作都在同一把锁下。这是标准的、正确的做法。对于大多数应用场景这个性能已经足够。如果确实遇到瓶颈首先要分析的是信号量本身的使用模式是否合理比如是否过度同步而不是盲目优化这个底层原语。结论对于绝大多数应用我们实现的这个基于锁的信号量在性能和正确性上已经取得了很好的平衡。“先确保正确再考虑优化”是并发编程的第一准则。5.3 常见陷阱与实战经验忘记配对使用Acquire/Release不匹配这是最常见的错误。每个acquire都必须对应一个release否则许可证会逐渐耗尽导致线程永久阻塞。在复杂的代码路径中尤其是存在异常或多个返回点务必使用RAII包装器。例如可以写一个SemaphoreGuard类在构造函数中acquire在析构函数中release确保异常安全。class SemaphoreGuard { public: explicit SemaphoreGuard(CountingSemaphore sem) : sem_(sem) { sem_.acquire(); } ~SemaphoreGuard() { sem_.release(); } // 禁止拷贝和移动 SemaphoreGuard(const SemaphoreGuard) delete; SemaphoreGuard operator(const SemaphoreGuard) delete; private: CountingSemaphore sem_; }; // 使用 { SemaphoreGuard guard(my_semaphore); // 自动acquire // ... 访问共享资源 } // 离开作用域guard析构自动release错误选择notify_one与notify_all如前所述在release(1)时使用notify_all是低效的它会导致“惊群效应”thundering herd所有等待线程都被唤醒去争夺一个许可证最终只有一个能成功其他线程又得回去睡觉产生大量无用的上下文切换。务必根据释放的许可证数量决定通知方式。信号量不是万能的信号量适合控制对一组完全相同资源的访问如连接池、线程池、缓冲区槽位。对于保护一段临界区代码即确保同一时间只有一个线程执行某段代码互斥锁std::mutex是更合适、语义更清晰的选择。虽然二进制信号量初始值为1可以达到类似效果但互斥锁具有“所有权”概念可以更好地与std::lock_guard、std::unique_lock等RAII机制配合并且一些调试工具能识别死锁与互斥锁的关系。死锁和互斥锁一样信号量也可能导致死锁。例如线程A持有信号量S1等待S2线程B持有S2等待S1。避免死锁的经典原则同样适用固定顺序获取资源、使用超时获取try_acquire_for、或者使用更高级的并发设施。生命周期管理再次强调确保信号量对象的生命周期长于所有使用它的线程。不要在还有线程等待时销毁信号量对象。在设计上通常将信号量作为全局对象、类的成员其生命周期与类实例绑定或通过std::shared_ptr进行共享管理。6. 扩展二进制信号量与更复杂的同步模式我们的CountingSemaphore是一个通用计数信号量。二进制信号量是它的一个特例初始值为1。你可以简单地用它来构建using BinarySemaphore CountingSemaphore; // 在构造时传入1 // 或者专门封装一下 class BinarySemaphore : public CountingSemaphore { public: BinarySemaphore() : CountingSemaphore(1) {} // 可以隐藏release的多参数版本因为二进制信号量通常只release(1) void release() { CountingSemaphore::release(1); } };此外信号量可以作为构建更复杂同步模式的基础模块例如读者-写者锁可以用两个信号量一个用于写者互斥一个用于控制读者计数来实现优先考虑读者或写者。屏障Barrier让一组线程在某个点同步等待所有线程到达后才能继续。有限资源池正如我们测试用例中的缓冲区信号量完美地建模了“空槽”和“满槽”的数量。实现这些模式是对信号量理解的很好练习。它们背后的核心思想依然是对共享状态的原子检查和条件等待。通过亲手实现这个C17信号量你获得的不只是一个可用的工具更重要的是对底层同步机制深刻、直观的理解这份理解在调试复杂的并发问题时将是无价之宝。当你下次再使用std::semaphore或者任何语言中的信号量时你都能清晰地看到它内部那个忙碌的计数器和那些在条件变量上安静等待的线程队列。
