C++11并发编程：call_once一次性执行+atomic原子类型+CAS无锁编程+自旋锁

承接上一篇互斥锁与RAII锁管理本篇聚焦于更加轻量级的同步方案std::call_once解决多线程环境下的一次性初始化问题std::atomic硬件级别的原子操作实现无锁同步性能远超互斥锁自旋锁基于std::atomic_flag实现的忙等待锁适合极短临界区场景1std::call_once线程安全的一次性执行1为什么需要call_once在多线程编程中经常遇到某个函数/代码块只需要执行一次的场景单例模式的初始化全局资源的加载配置文件的读取如果用普通的互斥锁实现会有两个问题每次调用都要加锁性能开销大传统的 双重检查锁定 (DCL) 在 C11 之前存在指令重排序问题行为未定义std::call_once是 C11 标准库提供的线程安全的一次性执行方案底层做了优化性能远高于手动加锁。2基本用法std::call_once需要配合std::once_flag使用template class Fn, class... Args void call_once(once_flag flag, Fn fn, Args... args);flag一次性标志每个需要执行一次的操作对应一个独立的once_flagfn要执行的可调用对象args传递给fn的参数代码示例#include iostream #include thread #include mutex using namespace std; once_flag g_once_flag; void init_resource() { cout 资源初始化完成只执行一次 endl; // 模拟资源初始化 this_thread::sleep_for(chrono::milliseconds(100)); } void worker(int id) { cout 线程 id 开始运行 endl; // 只有第一个到达这里的线程会执行init_resource call_once(g_once_flag, init_resource); cout 线程 id 继续执行 endl; } int main() { thread threads[5]; for (int i 0; i 5; i) { threads[i] thread(worker, i); } for (auto th : threads) { th.join(); } return 0; }3细节异常处理如果fn执行时抛出异常那么call_once会认为这次执行失败下一个到达的线程会再次尝试执行fn。只有当fn正常返回时才会标记为 已执行。once_flag 的生命周期once_flag的生命周期必须长于所有调用call_once的线程否则行为未定义。通常将once_flag声明为全局变量或静态变量。不能拷贝 / 移动once_flag不支持拷贝构造、移动构造、拷贝赋值和移动赋值每个once_flag只能对应一个一次性操作。4实际应用线程安全的单例模式这是call_once最经典的应用场景完美解决了传统单例的线程安全问题class Singleton { private: Singleton() { cout 单例对象创建 endl; } ~Singleton() { cout 单例对象销毁 endl; } Singleton(const Singleton) delete; Singleton operator(const Singleton) delete; static once_flag s_once_flag; static Singleton* s_instance; public: static Singleton* getInstance() { call_once(s_once_flag, [](){ s_instance new Singleton(); }); return s_instance; } }; // 静态成员初始化 once_flag Singleton::s_once_flag; Singleton* Singleton::s_instance nullptr;2std::atomic原子类型1什么是原子操作原子操作是指不可分割的操作要么全部执行完成要么完全不执行中间不会被其他线程打断。和互斥锁的对比互斥锁操作系统级别的同步会导致线程阻塞和上下文切换开销大但适合复杂临界区原子操作硬件 CPU 指令级别的同步无阻塞、无上下文切换开销极小但只能用于简单的操作如计数、标志位、指针交换2atomic的模版参数要求std::atomic是一个模板类可以实例化任何满足以下条件的类型T可平凡复制 (TriviallyCopyable)没有自定义的拷贝构造函数、析构函数等可复制构造和可复制赋值没有 cv 限定符const/volatile可以用以下类型特征检查std::is_trivially_copyableT::value std::is_copy_constructibleT::value std::is_copy_assignableT::value支持的常用类型所有整数类型int、long、char、bool等指针类型浮点数类型C20 起完全支持不支持的类型std::string、std::vector等复杂类型自定义类除非是平凡可复制的。3基本原子操作1load和store原子读写T load(memory_order order memory_order_seq_cst) const noexcept; void store(T desired, memory_order order memory_order_seq_cst) noexcept;load()原子地读取原子变量的值store()原子地将desired写入原子变量示例atomicbool flag(false); // 线程1设置标志 flag.store(true); // 线程2读取标志 if (flag.load()) { // 执行操作 }2运算符重载对于整数和指针类型std::atomic重载了常用的运算符自增自减、--前缀和后缀复合赋值、-、、|、^这些运算符都是原子操作等价于对应的fetch_*操作。示例原子计数器atomicint cnt(0); void add() { for (int i 0; i 1000000; i) { cnt; // 原子自增等价于cnt.fetch_add(1) } } int main() { thread t1(add); thread t2(add); t1.join(); t2.join(); cout cnt endl; // 输出2000000正确 return 0; }4读-修改-写原子操作fetch_*系列fetch_*系列函数是原子的读 - 修改 - 写操作执行以下步骤原子地读取原子变量的当前值对当前值执行指定的修改操作原子地将修改后的值写回原子变量返回修改前的原始值函数作用适用类型fetch_add(desired)原子加整数、指针fetch_sub(desired)原子减整数、指针fetch_and(desired)原子按位与整数fetch_or(desired)原子按位或整数fetch_xor(desired)原子按位异或整数atomicint x(10); int old x.fetch_add(5); // old10x变为15 cout old x endl; // 输出10 155CAS操作无锁变成的核心CASCompare And Swap比较并交换是无锁编程的基础几乎所有无锁数据结构都基于 CAS 实现。C11 提供了两个 CAS 成员函数bool compare_exchange_weak(T expected, T desired, memory_order success memory_order_seq_cst, memory_order failure memory_order_seq_cst) noexcept; bool compare_exchange_strong(T expected, T desired, memory_order success memory_order_seq_cst, memory_order failure memory_order_seq_cst) noexcept;CAS 操作的逻辑比较原子变量的当前值和expected的值如果相等将原子变量的值设置为desired返回true如果不相等将expected的值更新为原子变量的当前值返回false1compare_exchange_weak和compare_exchange_strong特性compare_exchange_weakcompare_exchange_strong虚假失败可能发生不会发生性能更高略低用法必须放在循环中可以不放在循环中什么是虚假失败在某些 CPU 架构如 ARM上即使原子变量的值等于expectedcompare_exchange_weak也可能返回false。这是由于硬件实现的限制不是逻辑错误。因此compare_exchange_weak必须放在循环中使用而compare_exchange_strong可以保证在值相等时一定成功。2CAS的经典用法无锁计数器#include atomic #include thread #include iostream using namespace std; atomicint cnt(0); void add() { for (int i 0; i 1000000; i) { int old cnt.load(); // 循环尝试CAS直到成功 while (!cnt.compare_exchange_weak(old, old 1)) { // 什么都不用做old已经被更新为当前值 } } } int main() { thread t1(add); thread t2(add); t1.join(); t2.join(); cout cnt endl; // 输出2000000 return 0; }注意这个例子只是为了演示 CAS 的用法实际中直接用cnt更简单高效。3CAS的ABA问题CAS 存在一个经典的ABA 问题线程 1 读取原子变量的值为 A线程 2 将原子变量的值改为 B然后又改回 A线程 1 执行 CAS发现值还是 A认为没有被修改操作成功但实际上原子变量的值已经被修改过了这在某些场景下会导致严重问题如无锁队列。解决方法使用带版本号的原子变量每次修改时同时递增版本号这样即使值相同版本号不同CAS 也会失败。6内存序内存序Memory Order用于控制编译器和 CPU 的指令重排序以及多线程之间的内存可见性。C11 提供了六种内存序选项从宽松到严格依次为memory_order_relaxed最宽松仅保证原子性不提供任何同步或顺序约束memory_order_consume保证依赖于当前加载操作的数据的可见性实际很少使用memory_order_acquire保证当前操作之前的所有读写操作不会被重排序到当前操作之后用于加载操作memory_order_release保证当前操作之后的所有读写操作不会被重排序到当前操作之前用于存储操作memory_order_acq_rel同时具有 acquire 和 release 语义用于读 - 修改 - 写操作memory_order_seq_cst最严格全局顺序一致性所有线程看到的操作顺序一致默认内存序常用的三种内存序memory_order_relaxed仅保证操作的原子性不保证顺序和可见性适用于不需要同步的场景如独立的计数器示例x.store(42, memory_order_relaxed);memory_order_acquire / memory_order_release成对使用实现线程间的同步当线程 A 用release存储一个变量线程 B 用acquire加载同一个变量时线程 A 中所有在release之前的写操作在线程 B 中都是可见的典型应用生产者 - 消费者模型atomicbool ready(false); int data 0; void producer() { data 42; // 写数据 ready.store(true, memory_order_release); // 发布数据 } void consumer() { while (!ready.load(memory_order_acquire)) { // 等待数据准备好 } cout data endl; // 保证看到data42 }memory_order_seq_cst默认内存序最安全性能最差保证所有线程看到的所有原子操作的顺序是一致的适用于需要强一致性的场景7std::atomic_flagstd::atomic_flag是 C11 中唯一保证无锁的原子类型比std::atomicbool更轻量性能更高。它只有两个核心操作// 原子地将flag设置为true并返回之前的值 bool test_and_set(memory_order order memory_order_seq_cst) noexcept; // 原子地将flag设置为false void clear(memory_order order memory_order_seq_cst) noexcept;atomic_flag flag ATOMIC_FLAG_INIT; // 初始化为false3自旋锁1什么是自旋锁自旋锁是一种忙等待的锁机制当一个线程尝试获取锁但失败时它不会进入阻塞状态而是会在一个循环中不断尝试获取锁直到成功为止。和互斥锁的对比特性自旋锁互斥锁等待方式忙等待循环尝试阻塞等待操作系统调度上下文切换无有开销锁持有时间短低锁持有时间长高锁持有时间短高锁持有时间长低适用场景极短的临界区几十条指令以内较长的临界区2自旋锁优缺点优点没有上下文切换开销在锁竞争不激烈且持有时间短的情况下性能远超互斥锁实现简单基于原子操作即可实现缺点忙等待会占用 CPU 资源如果锁持有时间长会导致 CPU 利用率飙升不支持递归加锁不支持条件变量单核 CPU 上使用自旋锁可能导致死锁因为持有锁的线程无法被抢占等待的线程会一直自旋3用std::atomic_flag实现自旋锁基于std::atomic_flag的test_and_set和clear方法可以非常简单地实现一个自旋锁#include atomic #include thread #include iostream #include vector using namespace std; class SpinLock { private: // 初始化为false表示未加锁 atomic_flag flag ATOMIC_FLAG_INIT; public: // 加锁循环尝试test_and_set直到返回false之前未加锁 void lock() { // 使用memory_order_acquire保证加锁之前的操作不会被重排序到加锁之后 while (flag.test_and_set(memory_order_acquire)) { // 空循环自旋等待 // 可以加入CPU pause指令减少CPU功耗 // #ifdef __x86_64__ // __asm__ __volatile__(pause); // #endif } } // 解锁将flag设置为false void unlock() { // 使用memory_order_release保证解锁之前的操作不会被重排序到解锁之后 flag.clear(memory_order_release); } // 禁止拷贝和移动 SpinLock(const SpinLock) delete; SpinLock operator(const SpinLock) delete; };代码解释lock()方法循环调用test_and_set如果返回false说明之前锁是未加锁状态当前线程成功获取锁如果返回true说明锁已经被其他线程持有继续循环等待。unlock()方法调用clear将 flag 设置为false释放锁。内存序使用memory_order_acquire和memory_order_release保证临界区的内存可见性比默认的memory_order_seq_cst性能更好。4自旋锁的测试和使用SpinLock spin_lock; int cnt 0; void add(int n) { for (int i 0; i n; i) { spin_lock.lock(); cnt; spin_lock.unlock(); } } int main() { const int thread_num 4; const int per_thread 1000000; vectorthread threads; for (int i 0; i thread_num; i) { threads.emplace_back(add, per_thread); } for (auto th : threads) { th.join(); } cout 最终计数 cnt endl; cout 预期计数 thread_num * per_thread endl; return 0; }5自旋锁的注意事项仅适用于极短临界区如果临界区执行时间超过 100 条指令自旋锁的性能会急剧下降此时应该使用互斥锁。不要在单核 CPU 上使用单核 CPU 上持有锁的线程无法被抢占等待的线程会一直自旋浪费 CPU 时间甚至导致死锁。持有自旋锁时不要调用阻塞函数如果持有自旋锁的线程调用sleep()、malloc()等阻塞函数会导致其他等待的线程长时间自旋浪费大量 CPU 资源。不要递归加锁自旋锁不支持递归加锁同一线程重复加锁会导致死锁。加入 CPU pause 指令在自旋循环中加入pause指令x86 架构可以减少 CPU 的功耗和流水线停顿提高性能。4总结std::call_once配合std::once_flag可以实现线程安全的一次性执行完美解决单例模式等初始化问题性能优于手动加锁。std::atomic提供了硬件级别的原子操作无阻塞、无上下文切换适合简单的同步场景是无锁编程的基础。CAS比较并交换是无锁编程的核心compare_exchange_weak性能更高但可能虚假失败必须放在循环中使用compare_exchange_strong保证成功但性能略低。内存序用于控制指令重排序和内存可见性最常用的是memory_order_relaxed和memory_order_acquire/release默认的memory_order_seq_cst最安全但性能最差。std::atomic_flag是最轻量的原子类型基于它可以实现简单高效的自旋锁自旋锁适合极短临界区场景但使用时需要注意避免 CPU 浪费。