本文是 C 系列教程的第 24 篇。上一篇講解了線程與互斥鎖本篇深入同步機(jī)制進(jìn)階條件變量與生產(chǎn)者消費(fèi)者模型、原子類型與內(nèi)存序、無鎖編程入門與讀寫鎖覆蓋 9 個(gè)完整示例代碼。一、條件變量std::condition_variable1.1 為什么需要條件變量互斥鎖只能保證「同時(shí)只有一個(gè)人進(jìn)房間」但無法解決「等待某個(gè)條件成立」的問題。條件變量允許線程阻塞等待某個(gè)條件直到另一線程通知它醒來避免忙等待浪費(fèi) CPU#includeiostream#includethread#includemutex#includecondition_variableusingnamespacestd;mutex mtx;condition_variable cv;boolreadyfalse;voidwaiter(){unique_lockmutexlock(mtx);cv.wait(lock,[]{returnready;});// 阻塞直到 ready truecout等待者被喚醒開始工作endl;}voidnotifier(){this_thread::sleep_for(chrono::milliseconds(500));{lock_guardmutexlock(mtx);readytrue;// 修改條件必須在鎖內(nèi)}cv.notify_one();// 喚醒一個(gè)等待線程}intmain(){threadt1(waiter);threadt2(notifier);t1.join();t2.join();return0;}wait(lock, predicate)的謂詞重載等價(jià)于while (!pred()) wait(lock)能自動(dòng)處理虛假喚醒spurious wakeup。1.2 生產(chǎn)消費(fèi)者模型條件變量最經(jīng)典的應(yīng)用場景。生產(chǎn)者往隊(duì)列放數(shù)據(jù)并通知消費(fèi)者阻塞等待并從隊(duì)列取數(shù)據(jù)#includeiostream#includethread#includemutex#includecondition_variable#includequeueusingnamespacestd;mutex mtx;condition_variable cv;queueinttasks;booldonefalse;voidproducer(){for(inti1;i5;i){{lock_guardmutexlock(mtx);tasks.push(i);cout生產(chǎn): iendl;}cv.notify_one();// 通知消費(fèi)者this_thread::sleep_for(chrono::milliseconds(100));}{lock_guardmutexlock(mtx);donetrue;}cv.notify_all();// 喚醒所有消費(fèi)者處理結(jié)束}voidconsumer(intid){while(true){unique_lockmutexlock(mtx);cv.wait(lock,[]{return!tasks.empty()||done;});if(!tasks.empty()){inttasktasks.front();tasks.pop();cout 消費(fèi)者 id 消費(fèi): taskendl;}elseif(done){break;// 生產(chǎn)結(jié)束且隊(duì)列為空}}}intmain(){threadp(producer);threadc1(consumer,1);threadc2(consumer,2);p.join();c1.join();c2.join();cout生產(chǎn)消費(fèi)完成endl;return0;}關(guān)鍵點(diǎn)notify_one喚醒單個(gè)線程notify_all喚醒全部等待條件必須用while/謂詞重載以防虛假喚醒done標(biāo)志防止消費(fèi)者永久阻塞。二、原子操作std::atomic2.1 原子類型基礎(chǔ)std::atomicT提供無鎖或鎖內(nèi)部實(shí)現(xiàn)的原子操作fetch_add、exchange、compare_exchange等保證讀-改-寫完整性無需互斥鎖#includeiostream#includethread#includeatomic#includevectorusingnamespacestd;atomicintcounter{0};voidincrement(){for(inti0;i100000;i)counter.fetch_add(1);}intmain(){vectorthreadthreads;for(inti0;i4;i)threads.emplace_back(increment);for(autot:threads)t.join();coutcounter counter.load()endl;// 400000無需加鎖return0;}fetch_add原子完成「讀-加-寫」。load()原子讀取store()原子寫入。2.2 compare_exchange 與自旋鎖compare_exchange_strong是 CAS 指令的封裝當(dāng)前值等于期望值時(shí)寫入新值否則更新期望值為實(shí)際值?？捎盟鼘?shí)現(xiàn)自旋鎖#includeiostream#includethread#includeatomic#includevectorusingnamespacestd;classSpinLock{atomicboolflag{false};public:voidlock(){// 期望 false嘗試寫入 true失敗則自旋重試while(flag.exchange(true)){// 空轉(zhuǎn)等待可加 this_thread::yield() 讓出 CPU}}voidunlock(){flag.store(false);}};SpinLock spin;intcounter0;voidwork(){for(inti0;i50000;i){lock_guardSpinLocklock(spin);counter;}}intmain(){vectorthreadthreads;for(inti0;i4;i)threads.emplace_back(work);for(autot:threads)t.join();coutcounter counterendl;// 200000return0;}自旋鎖適合臨界區(qū)極短的場景臨界區(qū)長時(shí)應(yīng)使用會阻塞的std::mutex避免浪費(fèi) CPU。三、內(nèi)存序Memory Order3.1 為什么要關(guān)心內(nèi)存序編譯器與 CPU 可能重排指令單線程不可見多線程可導(dǎo)致意外行為。內(nèi)存序控制重排邊界#includeiostream#includethread#includeatomicusingnamespacestd;atomicboolready{false};intdata0;voidproducer(){data42;// 寫數(shù)據(jù)ready.store(tru e,memory_order_release);// 釋放語義之前的寫操作全部可見}voidconsumer(){while(!ready.load(memory_order_acquire)){}// 獲取語義確保讀到最新 datacoutdata dataendl;// 保證讀到 42}intmain(){threadt1(producer);threadt2(consumer);t1.join();t2.join();return0;}release寫側(cè)與acquire讀側(cè)配對使用形成同步關(guān)系happens-before保證生產(chǎn)者寫入的數(shù)據(jù)對消費(fèi)者可見。3.2 常見內(nèi)存序?qū)Ρ葍?nèi)存序語義用途memory_order_relaxed無同步僅保證原子性計(jì)數(shù)器、統(tǒng)計(jì)量memory_order_acquire其后讀寫不可越過本操作讀取標(biāo)志位memory_order_release其前讀寫不可越過本操作發(fā)布數(shù)據(jù)memory_order_acq_relacquire releaseRMW 操作memory_order_seq_cst全序一致默認(rèn)復(fù)雜同步易推理#includeiostream#includethread#includeatomicusingnamespacestd;atomiclonglonghits{0};voidreport(){for(inti0;i1000000;i){hits.fetch_add(1,memory_order_relaxed);// 只需原子性無需同步}}intmain(){threadt1(report);threadt2(report);t1.join();t2.join();couthits hits.load(memory_order_relaxed)endl;// 2000000return0;}經(jīng)驗(yàn)?zāi)?seq_cst 最容易正確性能足夠時(shí)優(yōu)先使用只有基準(zhǔn)測試證明是瓶頸才降級為 relaxed/acquire/release 并仔細(xì)論證正確性。四、實(shí)戰(zhàn)線程安全的任務(wù)隊(duì)列綜合運(yùn)用互斥鎖、條件變量與 RAII 封裝一個(gè)可直接復(fù)用的線程安全隊(duì)列#includeiostream#includethread#includemutex#includecondition_variable#includequeue#includeoptionalusingnamespacestd;templatetypenameTclassThreadSafeQueue{mutablemutex mtx;condition_variable cv;queueTq;public:voidpush(T value){{lock_guardmutexlock(mtx);q.push(move(value));}cv.notify_one();}// 阻塞彈出Tpop(){unique_lockmutexlock(mtx);cv.wait(lock,[this]{return!q.empty();});T valuemove(q.front());q.pop();returnvalue;}// 非阻塞嘗試彈出optionalTtryPop(){lock_guardmutexlock(mtx);if(q.empty())returnnullopt;T valuemove(q.front());q.pop();returnvalue;}size_tsize()const{lock_guardmutexlock(mtx);returnq.size();}};intmain(){ThreadSafeQueueinttq;threadproducer([]{for(inti1;i6;i){tq.push(i);this_thread::sleep_for(chrono::milliseconds(50));}});threadconsumer([]{for(inti0;i6;i){intvtq.pop();// 阻塞等待cout取出: v隊(duì)列剩余 tq.size()endl;}});producer.join();consumer.join();cout線程安全隊(duì)列測試完成endl;return0;}該隊(duì)列把鎖與條件變量的復(fù)雜性封裝在內(nèi)部對外提供push/pop/tryPop/size四個(gè)安全接口是生產(chǎn)環(huán)境常用的基礎(chǔ)組件?？偨Y(jié)本篇講解了并發(fā)同步進(jìn)階技術(shù)condition_variable實(shí)現(xiàn)阻塞等待與通知含謂詞重載防虛假喚醒、生產(chǎn)者消費(fèi)者模型的完整實(shí)現(xiàn)、atomic 原子類型fetch_add/CAS 自旋鎖避免數(shù)據(jù)競爭、內(nèi)存序relaxed/acquire/release/seq_cst控制可見性與重排最后封裝了一個(gè)線程安全的任務(wù)隊(duì)列組件。建議配合互斥鎖按場景選用臨界區(qū)短用原子等待條件用條件變量常規(guī)保護(hù)用 mutex。下一篇將講解C 并發(fā)編程異步任務(wù)與線程池實(shí)戰(zhàn)std::async、future、packaged_task 與線程池實(shí)現(xiàn)敬請期待