上一篇讲完std::thread的生命周期接下来必须面对并发编程真正的核心问题多个线程同时碰同一块内存时会发生什么。答案不是「结果可能慢一点」而是「结果可以是任意值」。下面从数据竞争的原理讲起把std::mutex家族的接口、RAII 锁的三种选择、以及一次锁多把时的死锁避免说清楚每个结论都配上能跑的代码。不加锁的累加结果随机下面这段代码看起来很合理8 个线程各加 10 万次期望 80 万。#includethread#includevectorlonglongg_value0;// 没有锁保护的共享状态voidunsynchronized_add(){// 反例不要这么写g_value 1 实际是「读 - 改 - 写」三步// 两个线程可能读到同一个旧值各自加 1 再写回 —— 一次自增被吞掉for(inti0;i100000;i)g_value1;}voidrace_demo(){std::vectorstd::threadpool;for(inti0;i8;i)pool.emplace_back(unsynchronized_add);for(autot:pool)t.join();// g_value 通常小于 800000而且每次运行结果都不一样 —— 这就是数据竞争}这段没有main()是纯展示片段。之所以不给它配真实输出是因为它压根没有确定输出同一份二进制反复跑会得到 79 万、80 万附近的随机值偶尔还能「碰巧」等于 80 万。这种「有时候对」比稳定错误更危险。本地测十次都对压力一大就出错而且几乎无法复现。根本原因在编译器允许的假设单线程视角下没有其它执行流会偷偷改这块内存。于是g_value 1被拆成「读入寄存器 → 加 1 → 写回内存」三步中间随时可以被另一个线程插进来。这属于 C 标准定义的数据竞争data race行为是未定义的。它不只是算错理论上编译器可以做任何事。官方文档C 内存模型与数据竞争 — cppreference「数据竞争」的正式定义就在这里std::mutex三个接口std::mutex提供的最基本契约只有三个动作接口行为失败时lock()阻塞直到拿到锁已持锁者再调即为 UB不返回失败失败一般表现为死锁try_lock()试一下拿不到立刻返回false不阻塞返回falseunlock()释放锁未持锁时调用是 UB—用起来是这样线程 A 线程 B mutex 状态 │ │ ┌──────────┐ │ lock() ──────────────────────────────────────────────► │ 被 A 持有 │ │ 临界区: 读 value_ / 改 / 写回 │ └──────────┘ │ │ lock() ─────────────────► 阻塞等待中 │ unlock() ────────────────────────────────────────────► ┌──────────┐ │ │ 临界区开始 │ 被 B 持有 │ │ │ unlock() ──────────────► └──────────┘ 空闲手写lock()/unlock()基本必错。看反例#includemutex#includestdexceptstd::mutex g_mtx;boolg_should_failfalse;voidmanual_lock_unlock(){g_mtx.lock();// 反例不要这么写违反 Core Guidelines CP.20 / CP.50// 中间任何一步抛异常unlock() 就被跳过 —— 这把锁永久泄漏// 其它线程全部永远阻塞在 lock() 上程序表现为「莫名卡死」if(g_should_fail)throwstd::runtime_error{boom};g_mtx.unlock();}问题不在「忘了写unlock」而在**unlock和lock之间隔着一个可能抛异常的调用**。你没法保证每条路径都走到unlock除非让析构函数来兜底。这就是 RAII把锁绑在对象生命周期上析构即解锁。官方文档std::mutex — cppreference、C Core Guidelines · CP.20用 RAII不要裸 lock/unlocklock_guard最省事的 RAII 锁std::lock_guard是最简单的一档构造即加锁析构即解锁中间不给你任何操作空间。// counter_guard.cpp — 编译: g -stdc17 -Wall -O2 -pthread counter_guard.cpp -o counter_guard#includecstdio#includemutex#includethread#includevectorclassCounter{longlongvalue_{0};mutablestd::mutex mtx_;// mutablevalue() 是 const 函数也要能加锁public:voidadd(longlongdelta){conststd::lock_guardstd::mutexlock{mtx_};// 构造即锁析构即解锁value_delta;}longlongvalue()const{conststd::lock_guardstd::mutexlock{mtx_};returnvalue_;}};intmain(){constexprintkThreads8;constexprlonglongkPerThread100000;Counter counter;std::vectorstd::threadpool;pool.reserve(kThreads);for(inti0;ikThreads;i){pool.emplace_back([counter]{for(longlongk0;kkPerThread;k)counter.add(1);});}for(autot:pool)t.join();constlonglongexpectedkThreads*kPerThread;constlonglongactualcounter.value();std::printf(累加结果 %lld\n,actual);std::printf(期望值 %lld\n,expected);std::printf(是否一致 %s\n,actualexpected?是:否);}累加结果 800000 期望值 800000 是否一致 是和开头那段被吞掉自增的代码相比唯一的区别就是那句lock_guard。结果永远确定这正是加锁要买的东西把随机的错误换成确定的正确。三个实践细节lock_guard对象要const。它本身不是用来操作的加const表明「我只负责生命周期」也防止误用。作用域要尽量小。{ }括住的区间就是临界区锁持有越久其它线程等得越久。上面value()里「锁 → 读 → 返回」是最小写法绝不要在持锁期间做printf、文件 IO 或分配大内存。mutable std::mutex是必需的value()是const成员函数而lock()会修改互斥量状态所以互斥量本身必须mutable。这是「逻辑常量性」的正当用法。官方文档std::lock_guard — cppreferencelock_guard的局限也很明确不能手动unlock、不能延迟加锁、不能转移所有权、不能配条件变量。需要这些就得换下一档。unique_lock更贵但什么都能做std::unique_lock是「全功能版」。它内部多存了一个「当前是否持有锁」的标志因此比lock_guard略大、略慢一点点。很多人把它理解成「能手动解锁的lock_guard」其实多出来的那点开销正是这个标志带来的换来的是四件lock_guard做不到的事能力写法延迟加锁只包装先不锁std::unique_lockstd::mutex l{m, std::defer_lock};尝试加锁失败不阻塞std::unique_lockstd::mutex l{m, std::try_to_lock};手动unlock()/lock()随时释放并重新获取转移所有权std::unique_lockstd::mutex b{std::move(a)};配条件变量这是unique_lock存在的主要理由std::condition_variable::wait要求它最有用的是std::defer_lockstd::lock这个组合它能一次原子地锁住多把互斥量从而避免经典的「A 锁 B、B 锁 A」死锁// unique_lock_demo.cpp — 编译: g -stdc17 -Wall -O2 -pthread unique_lock_demo.cpp -o unique_lock_demo#includecstdio#includemutex#includethread#includeutilityclassAccount{longlongbalance_;mutablestd::mutex mtx_;public:explicitAccount(longlonginit):balance_{init}{}longlongbalance()const{conststd::lock_guardstd::mutexlock{mtx_};returnbalance_;}friendvoidtransfer(Accountfrom,Accountto,longlongamount);};// 一次锁两把用 std::lock 的死锁避免算法而不是手写「先锁 from 再锁 to」voidtransfer(Accountfrom,Accountto,longlongamount){std::unique_lockstd::mutexlock_from{from.mtx_,std::defer_lock};// 只包装不加锁std::unique_lockstd::mutexlock_to{to.mtx_,std::defer_lock};std::lock(lock_from,lock_to);// 要么两把都拿到要么一把都不持有from.balance_-amount;to.balance_amount;}intmain(){Account a{1000};Account b{500};std::printf(转账前: a %lld, b %lld\n,a.balance(),b.balance());// 两个方向相反的并发转账若手写「先锁 A 再锁 B」这里就会死锁std::thread t1{[a,b]{transfer(a,b,100);}};std::thread t2{[a,b]{transfer(b,a,30);}};t1.join();t2.join();std::printf(转账后: a %lld, b %lld\n,a.balance(),b.balance());std::printf(总额守恒: %s\n,(a.balance()b.balance())1500?是:否);// unique_lock 可以不拥有锁也可以把所有权转移给别人std::mutex gate;std::unique_lockstd::mutexlock{gate,std::defer_lock};std::printf(defer_lock 之后 owns_lock() %s\n,lock.owns_lock()?true:false);lock.lock();std::printf(手动 lock 之后 owns_lock() %s\n,lock.owns_lock()?true:false);std::unique_lockstd::mutexhanded_over{std::move(lock)};std::printf(move 之后: 原对象 %s, 新对象 %s\n,lock.owns_lock()?true:false,handed_over.owns_lock()?true:false);}转账前: a 1000, b 500 转账后: a 930, b 570 总额守恒: 是 defer_lock 之后 owns_lock() false 手动 lock 之后 owns_lock() true move 之后: 原对象 false, 新对象 true两个关键点std::lock(l1, l2)内部用的是「全拿或全不拿」的算法标准要求它必须避免死锁所以两个方向相反的转账不会互相卡住。自己写「先锁 A 再锁 B」在单线程时永远正确一上并发就是定时炸弹。转移所有权后原对象的owns_lock()变成false解锁责任一并转移。这正是「把锁交给别人管理」的基础也是condition_variable能工作的前提wait需要在等待期间释放锁醒来后再重新获取。官方文档std::unique_lock — cppreference、std::lock多锁死锁避免、C Core Guidelines · CP.21 / CP.22 / CP.50try_lock 与 timed_mutex等不到就走人有些场景不能无限等锁比如「拿不到就跳过这一帧」的渲染循环或者「超时就直接报错」的服务端。这时候用std::timed_mutex// try_lock.cpp — 编译: g -stdc17 -Wall -O2 -pthread try_lock.cpp -o try_lock#includeatomic#includechrono#includecstdio#includemutex#includethreadintmain(){std::timed_mutex gate;// ① 没人竞争时try_lock 立刻成功它是「试一下」不是「等下去」constboolidle_okgate.try_lock();std::printf(没人竞争: try_lock() - %s\n,idle_ok?拿到:没拿到);if(idle_ok)gate.unlock();// ② 有人持锁时try_lock_for 到点就放弃绝不无限等std::atomicboolholder_ready{false};std::thread holder{[gate,holder_ready]{gate.lock();holder_ready.store(true);std::this_thread::sleep_for(std::chrono::milliseconds{500});// 故意占住 500msgate.unlock();}};while(!holder_ready.load())std::this_thread::yield();// 等持锁者就绪constboolbusy_okgate.try_lock_for(std::chrono::milliseconds{50});std::printf(有人持锁: try_lock_for(50ms) - %s\n,busy_ok?拿到:超时放弃);if(busy_ok)gate.unlock();holder.join();constboolafter_okgate.try_lock_for(std::chrono::milliseconds{50});std::printf(持锁者退出后: try_lock_for(50ms) - %s\n,after_ok?拿到:超时放弃);if(after_ok)gate.unlock();}没人竞争: try_lock() - 拿到 有人持锁: try_lock_for(50ms) - 超时放弃 持锁者退出后: try_lock_for(50ms) - 拿到注意第三行超时不是永久失败只是「这 50ms 内没拿到」。try_lock_for一定要检查返回值再决定是否unlock否则会对着没持有的锁调用unlock那是 UB。std::mutex家族该选哪个类型同线程可重复加锁支持超时什么时候用std::mutex✗✗默认选择绝大多数场景std::recursive_mutex✓✗见下文的「通常是设计问题」std::timed_mutex✗✓try_lock_for/try_lock_until需要「等不到就走人」std::recursive_timed_mutex✓✓两者并集实践中极少用std::shared_mutexC17✗✗读多写少的读写锁关于std::recursive_mutex说句务实的它允许同一个线程重复加锁内部记一个计数看起来能救活「递归函数里顺手加锁」的代码。但递归锁几乎总是设计问题的信号。它掩盖了「临界区边界没划清楚」这件事并且会让不变量invariant在递归中途处于半成品状态别的线程看不到你自己也容易写错。真正需要它的场合通常是「要兼容一个已经写好的、内部也会加锁的第三方接口」。其余情况请改成只在最外层加一次锁内部函数换成「假设调用方已持锁」的私有实现。官方文档std::timed_mutex、std::recursive_mutex、std::shared_mutexC17三种 RAII 锁怎么选维度lock_guardunique_lockscoped_lockC17加锁时机构造即锁可选defer_lock/try_to_lock/adopt_lock构造即锁手动unlock/lock✗✓✗能否转移所有权✗✓✗能否配条件变量✗✓ 必须用它✗一次锁多把需配std::lockadopt_lock✓ 配std::lock✓ 直接传多个参数内部用死锁避免算法额外开销最小可能被优化成一个标志位也没有多一个「是否持有」标志 少量分支小适用场景简单的局部临界区默认用它条件变量 / 延迟加锁 / 转移所有权一次锁 2 把以上选型只有一句话能用lock_guard就用lock_guard一次要锁多把就用scoped_lock只有需要defer_lock、转移、或者配条件变量时才上unique_lock。不要因为unique_lock功能多就无脑用它多出来的灵活性是有成本的。官方文档std::scoped_lock — cppreference、C Core Guidelines · CP.20 / CP.50线程安全的关键字统计器把前面的东西串起来做一个多线程上报事件的次数统计// stats.cpp — 编译: g -stdc17 -Wall -O2 -pthread stats.cpp -o stats#includecstdio#includemap#includemutex#includestring#includethread#includevectorclassStats{std::mapstd::string,longlongcounts_;mutablestd::mutex mtx_;public:voidrecord(conststd::stringkey){conststd::lock_guardstd::mutexlock{mtx_};counts_[key];// 临界区读改写一个可能触发扩容的容器}std::mapstd::string,longlongsnapshot()const{conststd::lock_guardstd::mutexlock{mtx_};returncounts_;// 拷一份出去锁内不做耗时操作}};intmain(){constexprintkThreads4;constexprintkPerThread10000;conststd::vectorstd::stringkeys{cpu,mem,disk};Stats stats;std::vectorstd::threadpool;pool.reserve(kThreads);for(inti0;ikThreads;i){pool.emplace_back([stats,keys]{for(intk0;kkPerThread;k){stats.record(keys[static_caststd::size_t(k)%keys.size()]);}});}for(autot:pool)t.join();longlongtotal0;for(constauto[key,n]:stats.snapshot()){// C17 结构化绑定std::printf(%s %lld\n,key.c_str(),n);totaln;}std::printf(总计 %lld期望 %d\n,total,kThreads*kPerThread);}cpu 13336 disk 13332 mem 13332 总计 40000期望 40000这段有两个值得抄走的写法snapshot()在锁内拷贝、锁外遍历。持锁时间只够一次map拷贝而不是「持锁 循环 四次printf」。临界区越短并发度越高这是加锁优化里最有效的一招。record()里的临界区就是一次counts_[key]。std::map的插入可能触发节点分配和树旋转这些都在锁保护下完成换成std::unordered_map还要额外注意 rehash 会让所有迭代器失效。所以绝不能在锁外拿着迭代器指向 map 内部。延伸阅读std::mutex — cppreference三大接口的正式语义注意lock对已持锁者调用是 UBstd::lock_guard / std::unique_lock / std::scoped_lock三种 RAII 锁的成员清单选型前对着看一遍std::lock多锁死锁避免算法的官方说明defer_lock组合的搭档C 内存模型 — cppreference数据竞争、happens-before、顺序一致性的权威定义理解「为什么锁能修好它」的必读项C Core Guidelines · 并发篇CPCP.20、CP.21、CP.22、CP.50 都是加锁相关的硬规则收个尾数据竞争的根源是「读-改-写」不是原子操作锁把它变成了原子。三条规矩值得背下来锁一律用 RAIIlock_guard默认、scoped_lock锁多把、unique_lock只在需要延迟、转移或条件变量时才用临界区尽可能短绝不在临界区里做 IO 或长耗时操作。真正的功夫在划边界不在挑锁型。锁型选错顶多慢一点边界划错就是死锁和数据竞争后者比前者难查得多。