上一篇講完std::thread的生命周期接下來(lái)必須面對(duì)并發(fā)編程真正的核心問(wèn)題多個(gè)線程同時(shí)碰同一塊內(nèi)存時(shí)會(huì)發(fā)生什么。答案不是「結(jié)果可能慢一點(diǎn)」而是「結(jié)果可以是任意值」。下面從數(shù)據(jù)競(jìng)爭(zhēng)的原理講起把std::mutex家族的接口、RAII 鎖的三種選擇、以及一次鎖多把時(shí)的死鎖避免說(shuō)清楚每個(gè)結(jié)論都配上能跑的代碼。不加鎖的累加結(jié)果隨機(jī)下面這段代碼看起來(lái)很合理8 個(gè)線程各加 10 萬(wàn)次期望 80 萬(wàn)。#includethread#includevectorlonglongg_value0;// 沒(méi)有鎖保護(hù)的共享狀態(tài)voidunsynchronized_add(){// 反例不要這么寫(xiě)g_value 1 實(shí)際是「讀 - 改 - 寫(xiě)」三步// 兩個(gè)線程可能讀到同一個(gè)舊值各自加 1 再寫(xiě)回 —— 一次自增被吞掉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而且每次運(yùn)行結(jié)果都不一樣 —— 這就是數(shù)據(jù)競(jìng)爭(zhēng)}這段沒(méi)有main()是純展示片段。之所以不給它配真實(shí)輸出是因?yàn)樗鼔焊鶝](méi)有確定輸出同一份二進(jìn)制反復(fù)跑會(huì)得到 79 萬(wàn)、80 萬(wàn)附近的隨機(jī)值偶爾還能「碰巧」等于 80 萬(wàn)。這種「有時(shí)候?qū)Α贡确€(wěn)定錯(cuò)誤更危險(xiǎn)。本地測(cè)十次都對(duì)壓力一大就出錯(cuò)而且?guī)缀鯚o(wú)法復(fù)現(xiàn)。根本原因在編譯器允許的假設(shè)單線程視角下沒(méi)有其它執(zhí)行流會(huì)偷偷改這塊內(nèi)存。于是g_value 1被拆成「讀入寄存器 → 加 1 → 寫(xiě)回內(nèi)存」三步中間隨時(shí)可以被另一個(gè)線程插進(jìn)來(lái)。這屬于 C 標(biāo)準(zhǔn)定義的數(shù)據(jù)競(jìng)爭(zhēng)data race行為是未定義的。它不只是算錯(cuò)理論上編譯器可以做任何事。官方文檔C 內(nèi)存模型與數(shù)據(jù)競(jìng)爭(zhēng) — cppreference「數(shù)據(jù)競(jìng)爭(zhēng)」的正式定義就在這里std::mutex三個(gè)接口std::mutex提供的最基本契約只有三個(gè)動(dòng)作接口行為失敗時(shí)lock()阻塞直到拿到鎖已持鎖者再調(diào)即為 UB不返回失敗失敗一般表現(xiàn)為死鎖try_lock()試一下拿不到立刻返回false不阻塞返回falseunlock()釋放鎖未持鎖時(shí)調(diào)用是 UB—用起來(lái)是這樣線程 A 線程 B mutex 狀態(tài) │ │ ┌──────────┐ │ lock() ──────────────────────────────────────────────? │ 被 A 持有 │ │ 臨界區(qū): 讀 value_ / 改 / 寫(xiě)回 │ └──────────┘ │ │ lock() ─────────────────? 阻塞等待中 │ unlock() ────────────────────────────────────────────? ┌──────────┐ │ │ 臨界區(qū)開(kāi)始 │ 被 B 持有 │ │ │ unlock() ──────────────? └──────────┘ 空閑手寫(xiě)lock()/unlock()基本必錯(cuò)?？捶蠢?includemutex#includestdexceptstd::mutex g_mtx;boolg_should_failfalse;voidmanual_lock_unlock(){g_mtx.lock();// 反例不要這么寫(xiě)違反 Core Guidelines CP.20 / CP.50// 中間任何一步拋異常unlock() 就被跳過(guò) —— 這把鎖永久泄漏// 其它線程全部永遠(yuǎn)阻塞在 lock() 上程序表現(xiàn)為「莫名卡死」if(g_should_fail)throwstd::runtime_error{boom};g_mtx.unlock();}問(wèn)題不在「忘了寫(xiě)unlock」而在**unlock和lock之間隔著一個(gè)可能拋異常的調(diào)用**。你沒(méi)法保證每條路徑都走到unlock除非讓析構(gòu)函數(shù)來(lái)兜底。這就是 RAII把鎖綁在對(duì)象生命周期上析構(gòu)即解鎖。官方文檔std::mutex — cppreference、C Core Guidelines · CP.20用 RAII不要裸 lock/unlocklock_guard最省事的 RAII 鎖std::lock_guard是最簡(jiǎn)單的一檔構(gòu)造即加鎖析構(gòu)即解鎖中間不給你任何操作空間。// 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 函數(shù)也要能加鎖public:voidadd(longlongdelta){conststd::lock_guardstd::mutexlock{mtx_};// 構(gòu)造即鎖析構(gòu)即解鎖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(累加結(jié)果 %lld\n,actual);std::printf(期望值 %lld\n,expected);std::printf(是否一致 %s\n,actualexpected?是:否);}累加結(jié)果 800000 期望值 800000 是否一致 是和開(kāi)頭那段被吞掉自增的代碼相比唯一的區(qū)別就是那句lock_guard。結(jié)果永遠(yuǎn)確定這正是加鎖要買(mǎi)的東西把隨機(jī)的錯(cuò)誤換成確定的正確。三個(gè)實(shí)踐細(xì)節(jié)lock_guard對(duì)象要const。它本身不是用來(lái)操作的加const表明「我只負(fù)責(zé)生命周期」也防止誤用。作用域要盡量小。{ }括住的區(qū)間就是臨界區(qū)鎖持有越久其它線程等得越久。上面value()里「鎖 → 讀 → 返回」是最小寫(xiě)法絕不要在持鎖期間做printf、文件 IO 或分配大內(nèi)存。mutable std::mutex是必需的value()是const成員函數(shù)而lock()會(huì)修改互斥量狀態(tài)所以互斥量本身必須mutable。這是「邏輯常量性」的正當(dāng)用法。官方文檔std::lock_guard — cppreferencelock_guard的局限也很明確不能手動(dòng)unlock、不能延遲加鎖、不能轉(zhuǎn)移所有權(quán)、不能配條件變量。需要這些就得換下一檔。unique_lock更貴但什么都能做std::unique_lock是「全功能版」。它內(nèi)部多存了一個(gè)「當(dāng)前是否持有鎖」的標(biāo)志因此比lock_guard略大、略慢一點(diǎn)點(diǎn)。很多人把它理解成「能手動(dòng)解鎖的lock_guard」其實(shí)多出來(lái)的那點(diǎn)開(kāi)銷(xiāo)正是這個(gè)標(biāo)志帶來(lái)的換來(lái)的是四件lock_guard做不到的事能力寫(xiě)法延遲加鎖只包裝先不鎖std::unique_lockstd::mutex l{m, std::defer_lock};嘗試加鎖失敗不阻塞std::unique_lockstd::mutex l{m, std::try_to_lock};手動(dòng)unlock()/lock()隨時(shí)釋放并重新獲取轉(zhuǎn)移所有權(quán)std::unique_lockstd::mutex b{std::move(a)};配條件變量這是unique_lock存在的主要理由std::condition_variable::wait要求它最有用的是std::defer_lockstd::lock這個(gè)組合它能一次原子地鎖住多把互斥量從而避免經(jīng)典的「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 的死鎖避免算法而不是手寫(xiě)「先鎖 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(轉(zhuǎn)賬前: a %lld, b %lld\n,a.balance(),b.balance());// 兩個(gè)方向相反的并發(fā)轉(zhuǎn)賬若手寫(xiě)「先鎖 A 再鎖 B」這里就會(huì)死鎖std::thread t1{[a,b]{transfer(a,b,100);}};std::thread t2{[a,b]{transfer(b,a,30);}};t1.join();t2.join();std::printf(轉(zhuǎn)賬后: a %lld, b %lld\n,a.balance(),b.balance());std::printf(總額守恒: %s\n,(a.balance()b.balance())1500?是:否);// unique_lock 可以不擁有鎖也可以把所有權(quán)轉(zhuǎn)移給別人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(手動(dòng) lock 之后 owns_lock() %s\n,lock.owns_lock()?true:false);std::unique_lockstd::mutexhanded_over{std::move(lock)};std::printf(move 之后: 原對(duì)象 %s, 新對(duì)象 %s\n,lock.owns_lock()?true:false,handed_over.owns_lock()?true:false);}轉(zhuǎn)賬前: a 1000, b 500 轉(zhuǎn)賬后: a 930, b 570 總額守恒: 是 defer_lock 之后 owns_lock() false 手動(dòng) lock 之后 owns_lock() true move 之后: 原對(duì)象 false, 新對(duì)象 true兩個(gè)關(guān)鍵點(diǎn)std::lock(l1, l2)內(nèi)部用的是「全拿或全不拿」的算法標(biāo)準(zhǔn)要求它必須避免死鎖所以?xún)蓚€(gè)方向相反的轉(zhuǎn)賬不會(huì)互相卡住。自己寫(xiě)「先鎖 A 再鎖 B」在單線程時(shí)永遠(yuǎn)正確一上并發(fā)就是定時(shí)炸彈。轉(zhuǎn)移所有權(quán)后原對(duì)象的owns_lock()變成false解鎖責(zé)任一并轉(zhuǎn)移。這正是「把鎖交給別人管理」的基礎(chǔ)也是condition_variable能工作的前提wait需要在等待期間釋放鎖醒來(lái)后再重新獲取。官方文檔std::unique_lock — cppreference、std::lock多鎖死鎖避免、C Core Guidelines · CP.21 / CP.22 / CP.50try_lock 與 timed_mutex等不到就走人有些場(chǎng)景不能無(wú)限等鎖比如「拿不到就跳過(guò)這一幀」的渲染循環(huán)或者「超時(shí)就直接報(bào)錯(cuò)」的服務(wù)端。這時(shí)候用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;// ① 沒(méi)人競(jìng)爭(zhēng)時(shí)try_lock 立刻成功它是「試一下」不是「等下去」constboolidle_okgate.try_lock();std::printf(沒(méi)人競(jìng)爭(zhēng): try_lock() - %s\n,idle_ok?拿到:沒(méi)拿到);if(idle_ok)gate.unlock();// ② 有人持鎖時(shí)try_lock_for 到點(diǎn)就放棄絕不無(wú)限等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?拿到:超時(shí)放棄);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?拿到:超時(shí)放棄);if(after_ok)gate.unlock();}沒(méi)人競(jìng)爭(zhēng): try_lock() - 拿到 有人持鎖: try_lock_for(50ms) - 超時(shí)放棄 持鎖者退出后: try_lock_for(50ms) - 拿到注意第三行超時(shí)不是永久失敗只是「這 50ms 內(nèi)沒(méi)拿到」。try_lock_for一定要檢查返回值再?zèng)Q定是否unlock否則會(huì)對(duì)著沒(méi)持有的鎖調(diào)用unlock那是 UB。std::mutex家族該選哪個(gè)類(lèi)型同線程可重復(fù)加鎖支持超時(shí)什么時(shí)候用std::mutex??默認(rèn)選擇絕大多數(shù)場(chǎng)景std::recursive_mutex??見(jiàn)下文的「通常是設(shè)計(jì)問(wèn)題」std::timed_mutex??try_lock_for/try_lock_until需要「等不到就走人」std::recursive_timed_mutex??兩者并集實(shí)踐中極少用std::shared_mutexC17??讀多寫(xiě)少的讀寫(xiě)鎖關(guān)于std::recursive_mutex說(shuō)句務(wù)實(shí)的它允許同一個(gè)線程重復(fù)加鎖內(nèi)部記一個(gè)計(jì)數(shù)看起來(lái)能救活「遞歸函數(shù)里順手加鎖」的代碼。但遞歸鎖幾乎總是設(shè)計(jì)問(wèn)題的信號(hào)。它掩蓋了「臨界區(qū)邊界沒(méi)劃清楚」這件事并且會(huì)讓不變量invariant在遞歸中途處于半成品狀態(tài)別的線程看不到你自己也容易寫(xiě)錯(cuò)。真正需要它的場(chǎng)合通常是「要兼容一個(gè)已經(jīng)寫(xiě)好的、內(nèi)部也會(huì)加鎖的第三方接口」。其余情況請(qǐng)改成只在最外層加一次鎖內(nèi)部函數(shù)換成「假設(shè)調(diào)用方已持鎖」的私有實(shí)現(xiàn)。官方文檔std::timed_mutex、std::recursive_mutex、std::shared_mutexC17三種 RAII 鎖怎么選維度lock_guardunique_lockscoped_lockC17加鎖時(shí)機(jī)構(gòu)造即鎖可選defer_lock/try_to_lock/adopt_lock構(gòu)造即鎖手動(dòng)unlock/lock???能否轉(zhuǎn)移所有權(quán)???能否配條件變量?? 必須用它?一次鎖多把需配std::lockadopt_lock? 配std::lock? 直接傳多個(gè)參數(shù)內(nèi)部用死鎖避免算法額外開(kāi)銷(xiāo)最小可能被優(yōu)化成一個(gè)標(biāo)志位也沒(méi)有多一個(gè)「是否持有」標(biāo)志 少量分支小適用場(chǎng)景簡(jiǎn)單的局部臨界區(qū)默認(rèn)用它條件變量 / 延遲加鎖 / 轉(zhuǎn)移所有權(quán)一次鎖 2 把以上選型只有一句話(huà)能用lock_guard就用lock_guard一次要鎖多把就用scoped_lock只有需要defer_lock、轉(zhuǎn)移、或者配條件變量時(shí)才上unique_lock。不要因?yàn)閡nique_lock功能多就無(wú)腦用它多出來(lái)的靈活性是有成本的。官方文檔std::scoped_lock — cppreference、C Core Guidelines · CP.20 / CP.50線程安全的關(guān)鍵字統(tǒng)計(jì)器把前面的東西串起來(lái)做一個(gè)多線程上報(bào)事件的次數(shù)統(tǒng)計(jì)// 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];// 臨界區(qū)讀改寫(xiě)一個(gè)可能觸發(fā)擴(kuò)容的容器}std::mapstd::string,longlongsnapshot()const{conststd::lock_guardstd::mutexlock{mtx_};returncounts_;// 拷一份出去鎖內(nèi)不做耗時(shí)操作}};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 結(jié)構(gòu)化綁定std::printf(%s %lld\n,key.c_str(),n);totaln;}std::printf(總計(jì) %lld期望 %d\n,total,kThreads*kPerThread);}cpu 13336 disk 13332 mem 13332 總計(jì) 40000期望 40000這段有兩個(gè)值得抄走的寫(xiě)法snapshot()在鎖內(nèi)拷貝、鎖外遍歷。持鎖時(shí)間只夠一次map拷貝而不是「持鎖 循環(huán) 四次printf」。臨界區(qū)越短并發(fā)度越高這是加鎖優(yōu)化里最有效的一招。record()里的臨界區(qū)就是一次counts_[key]。std::map的插入可能觸發(fā)節(jié)點(diǎn)分配和樹(shù)旋轉(zhuǎn)這些都在鎖保護(hù)下完成換成std::unordered_map還要額外注意 rehash 會(huì)讓所有迭代器失效。所以絕不能在鎖外拿著迭代器指向 map 內(nèi)部。延伸閱讀std::mutex — cppreference三大接口的正式語(yǔ)義注意lock對(duì)已持鎖者調(diào)用是 UBstd::lock_guard / std::unique_lock / std::scoped_lock三種 RAII 鎖的成員清單選型前對(duì)著看一遍std::lock多鎖死鎖避免算法的官方說(shuō)明defer_lock組合的搭檔C 內(nèi)存模型 — cppreference數(shù)據(jù)競(jìng)爭(zhēng)、happens-before、順序一致性的權(quán)威定義理解「為什么鎖能修好它」的必讀項(xiàng)C Core Guidelines · 并發(fā)篇CPCP.20、CP.21、CP.22、CP.50 都是加鎖相關(guān)的硬規(guī)則收個(gè)尾數(shù)據(jù)競(jìng)爭(zhēng)的根源是「讀-改-寫(xiě)」不是原子操作鎖把它變成了原子。三條規(guī)矩值得背下來(lái)鎖一律用 RAIIlock_guard默認(rèn)、scoped_lock鎖多把、unique_lock只在需要延遲、轉(zhuǎn)移或條件變量時(shí)才用臨界區(qū)盡可能短絕不在臨界區(qū)里做 IO 或長(zhǎng)耗時(shí)操作。真正的功夫在劃邊界不在挑鎖型。鎖型選錯(cuò)頂多慢一點(diǎn)邊界劃錯(cuò)就是死鎖和數(shù)據(jù)競(jìng)爭(zhēng)后者比前者難查得多。