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