1. 項(xiàng)目概述為什么我們需要線(xiàn)程安全的數(shù)據(jù)結(jié)構(gòu)在C的多線(xiàn)程編程世界里數(shù)據(jù)競(jìng)爭(zhēng)Data Race是程序員最常遇到的“鬼影”之一。想象一下你精心設(shè)計(jì)了一個(gè)高性能的服務(wù)端程序用上了std::queue來(lái)作為任務(wù)隊(duì)列多個(gè)工作線(xiàn)程Worker Thread從中拉取任務(wù)處理。某一天在高并發(fā)壓力下程序毫無(wú)征兆地崩潰了或者更糟悄無(wú)聲息地產(chǎn)生了錯(cuò)誤的結(jié)果。排查下來(lái)很可能就是多個(gè)線(xiàn)程同時(shí)對(duì)一個(gè)隊(duì)列進(jìn)行push和pop操作導(dǎo)致其內(nèi)部狀態(tài)被破壞。標(biāo)準(zhǔn)庫(kù)提供的stack、queue、list、map等容器本身并不是線(xiàn)程安全的。這意味著如果多個(gè)線(xiàn)程在沒(méi)有同步機(jī)制保護(hù)的情況下訪(fǎng)問(wèn)同一個(gè)容器對(duì)象行為是未定義的Undefined Behavior。這就是我們這次要?jiǎng)邮謱?shí)現(xiàn)的東西基于鎖Lock的線(xiàn)程安全數(shù)據(jù)結(jié)構(gòu)。鎖特別是互斥鎖Mutex是多線(xiàn)程同步中最基礎(chǔ)、最直觀的“守門(mén)員”。我們的目標(biāo)不是發(fā)明新輪子而是給這些常用的容器“穿上盔甲”讓它們能在多線(xiàn)程環(huán)境中安全、正確地被使用。這個(gè)項(xiàng)目看似基礎(chǔ)卻是深入理解C并發(fā)編程、RAII資源獲取即初始化思想以及設(shè)計(jì)線(xiàn)程安全接口的絕佳練習(xí)。無(wú)論你是正在準(zhǔn)備面試還是在實(shí)際項(xiàng)目中遇到了并發(fā)數(shù)據(jù)訪(fǎng)問(wèn)的難題親手實(shí)現(xiàn)一遍這些封裝都會(huì)讓你對(duì)“線(xiàn)程安全”有更肌肉記憶般的理解。2. 核心設(shè)計(jì)思路與鎖的選擇在動(dòng)手寫(xiě)代碼之前我們必須先厘清幾個(gè)關(guān)鍵的設(shè)計(jì)決策。線(xiàn)程安全不是簡(jiǎn)單地在每個(gè)成員函數(shù)里加個(gè)鎖那么簡(jiǎn)單它關(guān)乎接口設(shè)計(jì)、性能權(quán)衡和死鎖預(yù)防。2.1 鎖的粒度與范圍鎖的粒度指的是鎖保護(hù)的數(shù)據(jù)范圍大小。一個(gè)最直接的想法是為整個(gè)數(shù)據(jù)結(jié)構(gòu)實(shí)例配備一個(gè)互斥鎖std::mutex任何訪(fǎng)問(wèn)該實(shí)例的公有成員函數(shù)都先上鎖執(zhí)行完再解鎖。這種“粗粒度鎖”設(shè)計(jì)簡(jiǎn)單能保證強(qiáng)線(xiàn)程安全但可能成為性能瓶頸。例如對(duì)于std::map如果find只讀操作和insert寫(xiě)入操作互斥在高讀低寫(xiě)的場(chǎng)景下會(huì)無(wú)謂地阻塞大量讀線(xiàn)程。另一種思路是“細(xì)粒度鎖”例如在鏈表list的每個(gè)節(jié)點(diǎn)上加鎖或者對(duì)哈希表map的不同桶bucket加不同的鎖。這能極大提升并發(fā)度但實(shí)現(xiàn)復(fù)雜度呈指數(shù)級(jí)上升需要考慮鎖的獲取順序來(lái)避免死鎖并且對(duì)數(shù)據(jù)結(jié)構(gòu)的內(nèi)部實(shí)現(xiàn)侵入性很強(qiáng)。對(duì)于我們的練習(xí)項(xiàng)目目標(biāo)是清晰、正確地展示基于鎖的線(xiàn)程安全封裝因此采用**每個(gè)容器實(shí)例一個(gè)互斥鎖粗粒度**的策略是合理且實(shí)用的起點(diǎn)。它確保了操作的原子性和狀態(tài)的完整性是工程中常見(jiàn)且有效的模式。2.2 鎖的類(lèi)型選擇C11標(biāo)準(zhǔn)庫(kù)在mutex頭文件中提供了幾種互斥鎖std::mutex: 最基本的互斥鎖不可遞歸同一線(xiàn)程重復(fù)加鎖會(huì)導(dǎo)致死鎖。std::recursive_mutex: 遞歸互斥鎖允許同一線(xiàn)程多次加鎖。std::timed_mutex: 帶超時(shí)功能的互斥鎖。std::shared_mutex(C17): 共享互斥鎖支持“讀-寫(xiě)鎖”語(yǔ)義允許多個(gè)讀線(xiàn)程同時(shí)訪(fǎng)問(wèn)?？紤]到我們的封裝需要支持const成員函數(shù)如top(),front(),find()而這些函數(shù)理論上只讀使用std::mutex會(huì)阻止并發(fā)讀。為了更優(yōu)的性能我們可以采用std::shared_mutex。這樣只讀操作可以共享地獲取鎖lock_shared而寫(xiě)入操作如push,pop,insert則獨(dú)占地獲取鎖lock。這能顯著提升以讀為主場(chǎng)景下的并發(fā)性能。注意使用std::shared_mutex需要C17或更高標(biāo)準(zhǔn)。如果你的項(xiàng)目環(huán)境限定在C11/14那么使用std::mutex是穩(wěn)妥的選擇。本文后續(xù)示例將基于std::shared_mutex進(jìn)行以展示更優(yōu)的設(shè)計(jì)同時(shí)也會(huì)給出std::mutex的替代方案。2.3 接口設(shè)計(jì)哲學(xué)是提供完整STL接口還是最小化接口STL容器的接口非常豐富。我們是否需要為線(xiàn)程安全版本實(shí)現(xiàn)所有接口比如std::stack的emplace、swapstd::map的迭代器相關(guān)操作。這里有一個(gè)重要的權(quán)衡接口越復(fù)雜線(xiàn)程安全的設(shè)計(jì)就越困難出錯(cuò)的概率也越高。例如提供迭代器意味著將內(nèi)部數(shù)據(jù)的“引用”或“指針”暴露給用戶(hù)用戶(hù)可以在鎖的范圍外持有并使用它這完全破壞了線(xiàn)程安全。因此一個(gè)常見(jiàn)的、也是更安全的設(shè)計(jì)是提供精簡(jiǎn)的、復(fù)合操作的接口。我們將遵循以下原則不暴露迭代器避免用戶(hù)繞過(guò)鎖進(jìn)行非原子操作。提供原子性復(fù)合操作例如pop操作通常需要返回被移除的元素。但std::stack::pop()只移除不返回需要結(jié)合top()使用這在多線(xiàn)程下不是原子的。我們將設(shè)計(jì)一個(gè)bool try_pop(T value)這樣的函數(shù)在鎖的保護(hù)下完成檢查和移除。謹(jǐn)慎設(shè)計(jì)const成員函數(shù)確保它們?cè)趕hared_lock的保護(hù)下執(zhí)行?；谶@些思路我們將逐一實(shí)現(xiàn)四個(gè)容器ThreadSafeStackThreadSafeQueueThreadSafeListThreadSafeMap。3. 基礎(chǔ)構(gòu)建RAII鎖守衛(wèi)與異常安全在實(shí)現(xiàn)具體容器前必須先理解并運(yùn)用好“鎖守衛(wèi)”Lock Guard。手動(dòng)調(diào)用lock()和unlock()是極易出錯(cuò)的尤其是在有異常拋出的情況下可能導(dǎo)致鎖無(wú)法釋放。C標(biāo)準(zhǔn)庫(kù)提供了std::lock_guardstd::unique_lock和std::shared_lock來(lái)實(shí)現(xiàn)RAII式的鎖管理。#include mutex #include shared_mutex std::shared_mutex rw_mutex; // 寫(xiě)入操作使用獨(dú)占鎖 { std::unique_lockstd::shared_mutex writer_lock(rw_mutex); // 構(gòu)造時(shí)加鎖獨(dú)占 // ... 執(zhí)行寫(xiě)入操作 } // 析構(gòu)時(shí)自動(dòng)解鎖 // 讀取操作使用共享鎖 { std::shared_lockstd::shared_mutex reader_lock(rw_mutex); // 構(gòu)造時(shí)加鎖共享 // ... 執(zhí)行只讀操作 } // 析構(gòu)時(shí)自動(dòng)解鎖std::unique_lock比std::lock_guard更靈活支持延遲加鎖、轉(zhuǎn)移所有權(quán)等但開(kāi)銷(xiāo)稍大。對(duì)于簡(jiǎn)單的加鎖-解鎖場(chǎng)景std::lock_guard就足夠了。但在我們需要配合條件變量std::condition_variable時(shí)必須使用std::unique_lock。實(shí)操心得始終優(yōu)先使用RAII鎖管理對(duì)象而不是裸的lock()/unlock()調(diào)用。這不僅是代碼簡(jiǎn)潔的問(wèn)題更是保證異常安全Exception Safety的生命線(xiàn)。即使你認(rèn)為當(dāng)前代碼塊不可能拋出異常未來(lái)的修改也可能引入異常使用鎖守衛(wèi)是防御性編程的好習(xí)慣。4. 核心實(shí)現(xiàn)線(xiàn)程安全棧ThreadSafeStack棧Stack是LIFO后進(jìn)先出結(jié)構(gòu)接口相對(duì)簡(jiǎn)單。我們面臨的主要挑戰(zhàn)是如何安全地實(shí)現(xiàn)“檢查并彈出”這個(gè)復(fù)合操作。4.1 類(lèi)定義與數(shù)據(jù)成員#include stack #include mutex #include shared_mutex #include memory // 用于std::shared_ptr #include exception templatetypename T class ThreadSafeStack { private: std::stackT data_; // 底層容器 mutable std::shared_mutex mutex_; // mutable使得const成員函數(shù)也能修改它加鎖 public: ThreadSafeStack() default; ThreadSafeStack(const ThreadSafeStack other) { // 拷貝構(gòu)造需要同時(shí)鎖住兩個(gè)對(duì)象的鎖避免死鎖 std::unique_lockstd::shared_mutex lock_other(other.mutex_); std::unique_lockstd::shared_mutex lock_this(mutex_, std::defer_lock); std::lock(lock_this, lock_other); // 同時(shí)鎖住兩個(gè)避免死鎖 data_ other.data_; } ThreadSafeStack operator(const ThreadSafeStack) delete; // 簡(jiǎn)單起見(jiàn)禁止賦值 void push(T new_value) { std::unique_lockstd::shared_mutex lock(mutex_); data_.push(std::move(new_value)); // 使用移動(dòng)語(yǔ)義提升性能 } // 關(guān)鍵安全的pop操作通過(guò)輸出參數(shù)返回 bool try_pop(T value) { std::unique_lockstd::shared_mutex lock(mutex_); if(data_.empty()) { return false; } value std::move(data_.top()); // 移動(dòng)賦值 data_.pop(); return true; } // 返回std::shared_ptr的版本避免拷貝/移動(dòng)可能拋異常 std::shared_ptrT try_pop() { std::unique_lockstd::shared_mutex lock(mutex_); if(data_.empty()) { return std::shared_ptrT(); } std::shared_ptrT const res(std::make_sharedT(std::move(data_.top()))); data_.pop(); return res; } // 只讀操作top 和 empty使用共享鎖 bool empty() const { std::shared_lockstd::shared_mutex lock(mutex_); return data_.empty(); } // 注意此top返回副本而非引用避免外部修改破壞線(xiàn)程安全 T top() const { std::shared_lockstd::shared_mutex lock(mutex_); if(data_.empty()) { throw std::runtime_error(empty stack); } return data_.top(); // 返回值的拷貝 } };4.2 關(guān)鍵點(diǎn)解析與避坑指南mutable關(guān)鍵字mutex_成員變量需要在const成員函數(shù)如empty(),top()中被加鎖。加鎖操作會(huì)改變互斥量的內(nèi)部狀態(tài)因此必須用mutable修飾告訴編譯器“這個(gè)變量即使在const成員函數(shù)中也是可變的”?？截悩?gòu)造與死鎖拷貝構(gòu)造函數(shù)需要同時(shí)訪(fǎng)問(wèn)this和other的數(shù)據(jù)。如果先鎖自己再鎖別人或反之在兩個(gè)線(xiàn)程同時(shí)以相反順序拷貝對(duì)方時(shí)就會(huì)形成經(jīng)典的死鎖。解決方案是使用std::lock函數(shù)它能一次性鎖住多個(gè)鎖對(duì)象且保證不會(huì)死鎖。我們使用std::defer_lock先創(chuàng)建鎖但不加鎖然后將兩個(gè)鎖對(duì)象傳給std::lock。try_pop的兩種形式bool try_pop(T value)通過(guò)引用輸出參數(shù)返回元素。優(yōu)點(diǎn)是效率高直接移動(dòng)數(shù)據(jù)。缺點(diǎn)是調(diào)用者需要先構(gòu)造一個(gè)T對(duì)象可能開(kāi)銷(xiāo)大且T的移動(dòng)賦值操作符必須是異常安全的。std::shared_ptrT try_pop()返回智能指針。這是更推薦的做法。首先std::make_shared在堆上構(gòu)造對(duì)象即使T的拷貝/移動(dòng)構(gòu)造函數(shù)拋異常也不會(huì)影響棧的原始數(shù)據(jù)。其次智能指針管理生命周期方便安全。最后如果pop失敗返回空指針語(yǔ)義清晰。top()返回拷貝為了線(xiàn)程安全我們不能返回棧頂元素的引用或指針因?yàn)殒i在函數(shù)返回后就釋放了外部持有引用進(jìn)行修改是非法的。因此top()必須返回一個(gè)副本。這帶來(lái)了拷貝開(kāi)銷(xiāo)但換來(lái)了安全。如果T對(duì)象很大可以考慮返回std::shared_ptrT就像try_pop那樣。異常安全push操作中data_.push(std::move(new_value))可能會(huì)因?yàn)閮?nèi)存分配失敗而拋出std::bad_alloc。但此時(shí)鎖已被獲取并且在lock對(duì)象析構(gòu)時(shí)會(huì)自動(dòng)釋放因此不會(huì)導(dǎo)致死鎖。這是RAII帶來(lái)的基本異常安全保證。注意事項(xiàng)謹(jǐn)慎處理“檢查-執(zhí)行”模式。像if(!stack.empty()) { value stack.top(); stack.pop(); }這樣的代碼在多線(xiàn)程下是絕對(duì)不安全的因?yàn)樵趀mpty()和top()之間其他線(xiàn)程可能已經(jīng)修改了棧。我們的try_pop將檢查和執(zhí)行合并為一個(gè)原子操作是唯一安全的方式。5. 核心實(shí)現(xiàn)線(xiàn)程安全隊(duì)列ThreadSafeQueue隊(duì)列Queue是FIFO先進(jìn)先出結(jié)構(gòu)是生產(chǎn)者-消費(fèi)者模型的典型媒介。除了基本的線(xiàn)程安全我們經(jīng)常需要讓消費(fèi)者線(xiàn)程在隊(duì)列為空時(shí)等待而不是忙等待busy-waiting。這就需要引入條件變量std::condition_variable。5.1 支持等待的線(xiàn)程安全隊(duì)列我們將實(shí)現(xiàn)一個(gè)更實(shí)用的、支持阻塞等待的ThreadSafeQueue。#include queue #include mutex #include condition_variable templatetypename T class ThreadSafeQueue { private: mutable std::mutex mutex_; // 條件變量需要std::mutex且通常讀寫(xiě)都需要互斥直接用mutex std::queueT data_; std::condition_variable data_cond_; public: ThreadSafeQueue() default; void push(T new_value) { std::lock_guardstd::mutex lock(mutex_); data_.push(std::move(new_value)); data_cond_.notify_one(); // 通知一個(gè)等待的消費(fèi)者 } // 等待并彈出 void wait_and_pop(T value) { std::unique_lockstd::mutex lock(mutex_); // 等待條件隊(duì)列非空。lambda表達(dá)式是謂詞防止虛假喚醒 data_cond_.wait(lock, [this]{ return !data_.empty(); }); value std::move(data_.front()); data_.pop(); } std::shared_ptrT wait_and_pop() { std::unique_lockstd::mutex lock(mutex_); data_cond_.wait(lock, [this]{ return !data_.empty(); }); std::shared_ptrT res(std::make_sharedT(std::move(data_.front()))); data_.pop(); return res; } // 非阻塞嘗試 bool try_pop(T value) { std::lock_guardstd::mutex lock(mutex_); if(data_.empty()) { return false; } value std::move(data_.front()); data_.pop(); return true; } std::shared_ptrT try_pop() { std::lock_guardstd::mutex lock(mutex_); if(data_.empty()) { return std::shared_ptrT(); } std::shared_ptrT res(std::make_sharedT(std::move(data_.front()))); data_.pop(); return res; } bool empty() const { std::lock_guardstd::mutex lock(mutex_); return data_.empty(); } };5.2 條件變量的使用與虛假喚醒std::condition_variable它允許線(xiàn)程等待某個(gè)條件成立。必須與std::unique_lockstd::mutex配合使用。wait方法data_cond_.wait(lock, predicate)。這里predicate是一個(gè)可調(diào)用對(duì)象我們用了lambda返回bool。wait的內(nèi)部邏輯是檢查predicate如果為true則繼續(xù)如果為false則原子地釋放鎖并使線(xiàn)程進(jìn)入等待狀態(tài)。當(dāng)被notify_one()或notify_all()喚醒時(shí)線(xiàn)程會(huì)重新獲取鎖并再次檢查predicate。這個(gè)循環(huán)檢查是必須的用來(lái)防止“虛假喚醒”Spurious Wakeup——即線(xiàn)程可能在沒(méi)有收到任何通知的情況下被操作系統(tǒng)喚醒。notify_one()vsnotify_all()notify_one()喚醒一個(gè)正在等待的線(xiàn)程如果有notify_all()喚醒所有等待的線(xiàn)程。在單生產(chǎn)者-多消費(fèi)者場(chǎng)景下使用notify_one()可能更高效因?yàn)橹挥幸粋€(gè)元素被加入只需要喚醒一個(gè)消費(fèi)者。但如果消費(fèi)者線(xiàn)程有不同的任務(wù)或者你想讓所有消費(fèi)者檢查新的狀態(tài)則用notify_all()。實(shí)操心得永遠(yuǎn)使用帶謂詞predicate的wait。直接使用data_cond_.wait(lock)然后在后面用if判斷條件是錯(cuò)誤的模式無(wú)法抵御虛假喚醒。將條件檢查放入wait的謂詞中是C并發(fā)編程的標(biāo)準(zhǔn)做法。6. 核心實(shí)現(xiàn)線(xiàn)程安全單向鏈表ThreadSafeList鏈表List的插入和刪除操作可以在內(nèi)部節(jié)點(diǎn)完成理論上可以實(shí)現(xiàn)比全局鎖更細(xì)粒度的并發(fā)。但為了保持實(shí)現(xiàn)的清晰和作為教學(xué)示例我們?nèi)匀皇褂靡粋€(gè)全局互斥鎖。這里我們實(shí)現(xiàn)一個(gè)簡(jiǎn)單的單向鏈表。6.1 鏈表節(jié)點(diǎn)與類(lèi)定義#include memory #include mutex templatetypename T class ThreadSafeList { private: struct Node { std::shared_ptrT data; // 存儲(chǔ)數(shù)據(jù) std::unique_ptrNode next; // 下一個(gè)節(jié)點(diǎn)的所有權(quán) Node() : next(nullptr) {} explicit Node(T value) : data(std::make_sharedT(std::move(value))), next(nullptr) {} }; Node head_; // 啞節(jié)點(diǎn)dummy node簡(jiǎn)化邊界處理 mutable std::mutex mutex_; public: ThreadSafeList() default; ~ThreadSafeList() { remove_if([](const Node){ return true; }); } // 析構(gòu)時(shí)刪除所有節(jié)點(diǎn) ThreadSafeList(const ThreadSafeList) delete; ThreadSafeList operator(const ThreadSafeList) delete; void push_front(T value) { std::unique_ptrNode new_node(new Node(std::move(value))); // 在鎖外構(gòu)造新節(jié)點(diǎn) std::lock_guardstd::mutex lock(mutex_); new_node-next std::move(head_.next); // 接管原頭節(jié)點(diǎn)之后的鏈表 head_.next std::move(new_node); // 新節(jié)點(diǎn)成為頭節(jié)點(diǎn)之后第一個(gè) } // 遍歷鏈表并對(duì)每個(gè)元素執(zhí)行函數(shù)Func templatetypename Func void for_each(Func func) { std::lock_guardstd::mutex lock(mutex_); Node* current head_; while(Node* const next current-next.get()) { // 遍歷真實(shí)節(jié)點(diǎn) func(*next-data); // 對(duì)數(shù)據(jù)執(zhí)行操作 current next; } } // 查找第一個(gè)使謂詞p返回true的元素返回其數(shù)據(jù)的shared_ptr templatetypename Predicate std::shared_ptrT find_first_if(Predicate p) { std::lock_guardstd::mutex lock(mutex_); Node* current head_; while(Node* const next current-next.get()) { if(p(*next-data)) { return next-data; } current next; } return std::shared_ptrT(); } // 移除所有使謂詞p返回true的節(jié)點(diǎn) templatetypename Predicate void remove_if(Predicate p) { std::lock_guardstd::mutex lock(mutex_); Node* current head_; while(Node* const next current-next.get()) { if(p(*next-data)) { std::unique_ptrNode old_next std::move(current-next); current-next std::move(next-next); // old_next 在作用域結(jié)束時(shí)自動(dòng)刪除 } else { current next; } } } };6.2 設(shè)計(jì)亮點(diǎn)與線(xiàn)程安全考量啞節(jié)點(diǎn)Dummy Nodehead_是一個(gè)不存儲(chǔ)實(shí)際數(shù)據(jù)的節(jié)點(diǎn)。這極大地簡(jiǎn)化了插入和刪除的邏輯因?yàn)槲覀冇肋h(yuǎn)不需要修改head_指針本身只需要修改head_.next。所有實(shí)際數(shù)據(jù)都從head_.next開(kāi)始。節(jié)點(diǎn)所有權(quán)與unique_ptr每個(gè)節(jié)點(diǎn)擁有其next節(jié)點(diǎn)的唯一所有權(quán)。這保證了鏈表結(jié)構(gòu)的清晰和內(nèi)存管理的自動(dòng)化。當(dāng)從鏈表中移除一個(gè)節(jié)點(diǎn)時(shí)只需調(diào)整指針std::unique_ptr會(huì)自動(dòng)釋放被移除節(jié)點(diǎn)的內(nèi)存。數(shù)據(jù)存儲(chǔ)與shared_ptr節(jié)點(diǎn)內(nèi)部數(shù)據(jù)用std::shared_ptrT存儲(chǔ)。這樣做的好處是即使一個(gè)節(jié)點(diǎn)正在被遍歷for_each或查找find_first_if另一個(gè)線(xiàn)程刪除了這個(gè)節(jié)點(diǎn)只要還有shared_ptr持有數(shù)據(jù)數(shù)據(jù)對(duì)象本身就不會(huì)被銷(xiāo)毀避免了懸垂指針。find_first_if返回的也是shared_ptr延長(zhǎng)了數(shù)據(jù)的生命周期。操作粒度push_front、for_each、find_first_if、remove_if每個(gè)操作都持有鎖。for_each和find_first_if遍歷整個(gè)鏈表持有鎖的時(shí)間可能較長(zhǎng)這在長(zhǎng)鏈表和高并發(fā)下可能成為瓶頸。這是粗粒度鎖的典型缺點(diǎn)。更高級(jí)的實(shí)現(xiàn)可以為每個(gè)節(jié)點(diǎn)配備一個(gè)鎖但鎖的管理會(huì)非常復(fù)雜。函數(shù)模板的使用for_each、find_first_if、remove_if都接受一個(gè)可調(diào)用對(duì)象函數(shù)、lambda表達(dá)式等。這提供了極大的靈活性用戶(hù)可以在鎖的保護(hù)下執(zhí)行自定義操作而無(wú)需將數(shù)據(jù)拷貝出去。注意事項(xiàng)警惕在鎖范圍內(nèi)執(zhí)行用戶(hù)代碼。for_each和remove_if中的func和p是用戶(hù)提供的。如果這些函數(shù)執(zhí)行了非常耗時(shí)的操作或者嘗試去獲取其他鎖可能會(huì)導(dǎo)致本鎖被長(zhǎng)期持有甚至引發(fā)死鎖。在設(shè)計(jì)這類(lèi)接口時(shí)需要清楚地告知用戶(hù)傳入的函數(shù)應(yīng)盡量輕量且避免執(zhí)行可能產(chǎn)生死鎖的操作。7. 核心實(shí)現(xiàn)線(xiàn)程安全映射表ThreadSafeMap映射表Map通常指基于紅黑樹(shù)的std::map或基于哈希表的std::unordered_map。其線(xiàn)程安全封裝需要考慮的關(guān)鍵點(diǎn)是查找find和插入/更新insert/operator[]的并發(fā)。7.1 基于std::map的線(xiàn)程安全封裝我們選擇std::map作為底層容器并繼續(xù)使用std::shared_mutex來(lái)區(qū)分讀寫(xiě)鎖。#include map #include shared_mutex #include memory #include optional // C17 用于安全返回可能不存在的值 templatetypename Key, typename Value, typename Compare std::lessKey class ThreadSafeMap { private: std::mapKey, Value, Compare data_; mutable std::shared_mutex mutex_; public: ThreadSafeMap() default; // 插入或賦值。返回bool表示是否為新插入。 bool insert_or_assign(const Key key, Value value) { std::unique_lockstd::shared_mutex lock(mutex_); auto [it, inserted] data_.try_emplace(key, std::move(value)); if (!inserted) { it-second std::move(value); // 已存在則賦值 } return inserted; } // 僅當(dāng)鍵不存在時(shí)插入 bool insert_if_not_exist(const Key key, Value value) { std::unique_lockstd::shared_mutex lock(mutex_); return data_.try_emplace(key, std::move(value)).second; } // 安全的查找返回std::optional (C17) std::optionalValue find(const Key key) const { std::shared_lockstd::shared_mutex lock(mutex_); auto it data_.find(key); if (it ! data_.end()) { return it-second; // 隱式構(gòu)造std::optionalValue } return std::nullopt; // 未找到 } // 查找返回shared_ptr兼容C11/14 std::shared_ptrValue find_ptr(const Key key) const { std::shared_lockstd::shared_mutex lock(mutex_); auto it data_.find(key); if (it ! data_.end()) { return std::make_sharedValue(it-second); // 返回拷貝的shared_ptr // 注意這里返回的是拷貝如果Value很大開(kāi)銷(xiāo)需要考慮。 // 另一種設(shè)計(jì)是返回std::shared_ptrconst Value并直接指向map內(nèi)的元素。 // 但這要求Value在map存活期間不被移動(dòng)且需要更復(fù)雜的生命周期管理。 } return nullptr; } // 刪除指定鍵 bool erase(const Key key) { std::unique_lockstd::shared_mutex lock(mutex_); return data_.erase(key) 0; } // 遍歷所有鍵值對(duì)只讀 templatetypename Func void for_each(Func func) const { std::shared_lockstd::shared_mutex lock(mutex_); for (const auto kv_pair : data_) { func(kv_pair.first, kv_pair.second); } } // 清空 void clear() { std::unique_lockstd::shared_mutex lock(mutex_); data_.clear(); } // 獲取大小 size_t size() const { std::shared_lockstd::shared_mutex lock(mutex_); return data_.size(); } };7.2 關(guān)鍵設(shè)計(jì)決策與性能權(quán)衡insert_or_assignvsoperator[]我們沒(méi)有重載operator[]因?yàn)樗恼Z(yǔ)義“如果不存在則插入一個(gè)默認(rèn)構(gòu)造的Value”在多線(xiàn)程下可能不是用戶(hù)想要的而且它返回引用線(xiàn)程不安全。我們提供了insert_or_assign來(lái)明確語(yǔ)義。try_emplace的優(yōu)勢(shì)C17的try_emplace在鍵不存在時(shí)直接在容器內(nèi)構(gòu)造對(duì)象避免了不必要的拷貝或移動(dòng)。這比先f(wàn)ind再insert或emplace更高效。返回類(lèi)型的選擇std::optionalValue(C17)這是最現(xiàn)代、最清晰的方式明確表達(dá)了“可能有值可能無(wú)值”。std::shared_ptrValue兼容性更好C11并且通過(guò)智能指針管理生命周期更安全。但注意我們的實(shí)現(xiàn)返回的是數(shù)據(jù)的拷貝。如果Value類(lèi)型很大這有性能開(kāi)銷(xiāo)。一個(gè)更激進(jìn)但復(fù)雜的設(shè)計(jì)是讓ThreadSafeMap內(nèi)部存儲(chǔ)std::shared_ptrValue這樣find_ptr可以直接返回內(nèi)部指針的引用計(jì)數(shù)拷貝無(wú)需拷貝數(shù)據(jù)本身。但這改變了容器的語(yǔ)義和內(nèi)存布局。for_each遍歷和ThreadSafeList一樣我們?cè)阪i的保護(hù)下執(zhí)行用戶(hù)函數(shù)。對(duì)于大的map持有讀鎖的時(shí)間可能較長(zhǎng)。如果遍歷操作非常耗時(shí)需要考慮是否真的需要線(xiàn)程安全或者能否將數(shù)據(jù)快照拷貝出來(lái)再處理。沒(méi)有提供“更新現(xiàn)有值”的原子操作有時(shí)我們需要“查找-計(jì)算-更新”這樣一個(gè)復(fù)合操作例如map[key] 1。我們目前的接口無(wú)法原子地完成這個(gè)操作。用戶(hù)需要先f(wàn)ind計(jì)算新值再insert_or_assign這中間map可能已被其他線(xiàn)程修改。如果需要此類(lèi)操作可以增加一個(gè)成員函數(shù)templatetypename Updater bool update(const Key key, Updater updater) { std::unique_lockstd::shared_mutex lock(mutex_); auto it data_.find(key); if (it ! data_.end()) { updater(it-second); // 用戶(hù)提供的更新函數(shù) return true; } return false; }這樣整個(gè)查找和更新過(guò)程在獨(dú)占鎖的保護(hù)下完成是原子的。常見(jiàn)問(wèn)題“我該用std::map還是std::unordered_map作為底層容器”這取決于你的使用場(chǎng)景。std::map基于紅黑樹(shù)鍵是有序的插入、刪除、查找的平均時(shí)間復(fù)雜度是O(log n)。std::unordered_map基于哈希表平均時(shí)間復(fù)雜度是O(1)但鍵是無(wú)序的且哈希函數(shù)和負(fù)載因子會(huì)影響性能。在并發(fā)環(huán)境下如果讀遠(yuǎn)大于寫(xiě)std::unordered_map的O(1)查找可能更有優(yōu)勢(shì)。但線(xiàn)程安全封裝本身的鎖開(kāi)銷(xiāo)可能遠(yuǎn)大于容器操作的開(kāi)銷(xiāo)所以底層容器的選擇需要根據(jù)實(shí)際數(shù)據(jù)規(guī)模和訪(fǎng)問(wèn)模式進(jìn)行性能測(cè)試。8. 性能考量、死鎖預(yù)防與進(jìn)階話(huà)題實(shí)現(xiàn)完基本版本后我們必須審視其性能和潛在風(fēng)險(xiǎn)。8.1 性能瓶頸分析我們實(shí)現(xiàn)的四個(gè)容器都使用了“全局一把鎖”的策略。這在并發(fā)度不高、操作簡(jiǎn)單的場(chǎng)景下是可行的。但在高并發(fā)場(chǎng)景下它可能成為嚴(yán)重的瓶頸鎖競(jìng)爭(zhēng)所有線(xiàn)程都在爭(zhēng)奪同一個(gè)鎖即使它們?cè)L問(wèn)的是數(shù)據(jù)結(jié)構(gòu)的不同部分例如一個(gè)在鏈表頭插入一個(gè)在鏈表尾查找。鎖持有時(shí)間像for_each這樣的遍歷操作會(huì)長(zhǎng)時(shí)間持有鎖阻塞所有其他操作。優(yōu)化方向更細(xì)粒度的鎖如為鏈表的每個(gè)節(jié)點(diǎn)、哈希表的每個(gè)桶配備獨(dú)立的鎖。這能極大提升并發(fā)度但實(shí)現(xiàn)復(fù)雜度高且可能增加內(nèi)存開(kāi)銷(xiāo)和鎖管理開(kāi)銷(xiāo)。無(wú)鎖Lock-Free數(shù)據(jù)結(jié)構(gòu)使用原子操作std::atomic和內(nèi)存序Memory Order來(lái)實(shí)現(xiàn)并發(fā)安全完全避免互斥鎖。性能可能極高但實(shí)現(xiàn)極其復(fù)雜正確性難以保證通常只適用于特定場(chǎng)景如簡(jiǎn)單的棧、隊(duì)列。讀寫(xiě)鎖Read-Write Lock我們已經(jīng)使用了std::shared_mutex這對(duì)讀多寫(xiě)少的場(chǎng)景是有效的優(yōu)化?？s小臨界區(qū)在鎖范圍內(nèi)只做最必要的操作。例如在push操作中先在鎖外構(gòu)造好新節(jié)點(diǎn)或數(shù)據(jù)鎖內(nèi)只進(jìn)行指針鏈接或容器插入。8.2 死鎖預(yù)防死鎖通常發(fā)生在需要獲取多個(gè)鎖時(shí)。我們的拷貝構(gòu)造函數(shù)已經(jīng)展示了如何使用std::lock來(lái)一次性鎖住多個(gè)互斥量避免因加鎖順序不一致導(dǎo)致的死鎖。死鎖產(chǎn)生的四個(gè)必要條件必須同時(shí)滿(mǎn)足互斥條件請(qǐng)求與保持條件不剝奪條件循環(huán)等待條件預(yù)防死鎖的實(shí)踐準(zhǔn)則固定鎖的順序如果一段代碼必須獲取鎖A和鎖B那么在所有地方都約定先獲取A再獲取B。使用std::lock當(dāng)需要獲取多個(gè)鎖時(shí)使用std::lock一次性獲取它使用死鎖避免算法。避免在鎖范圍內(nèi)調(diào)用用戶(hù)代碼我們之前提到過(guò)在for_each中執(zhí)行用戶(hù)函數(shù)是危險(xiǎn)的。如果用戶(hù)函數(shù)內(nèi)部又試圖獲取另一個(gè)鎖而另一個(gè)線(xiàn)程以相反的順序獲取這兩個(gè)鎖就會(huì)死鎖。使用層次鎖Hierarchical Mutex給鎖分配層級(jí)編號(hào)規(guī)定只能獲取層級(jí)更低的鎖。這可以在編譯期或運(yùn)行期檢查鎖的順序。8.3 內(nèi)存模型與std::atomic對(duì)于簡(jiǎn)單的計(jì)數(shù)器或標(biāo)志位使用std::atomic類(lèi)型通常比“互斥鎖普通變量”性能更好。例如如果你想在ThreadSafeQueue中添加一個(gè)“已處理任務(wù)計(jì)數(shù)”可以這樣class ThreadSafeQueue { // ... 其他成員 ... std::atomicsize_t pop_counter_{0}; public: std::shared_ptrT wait_and_pop() { // ... 原有邏輯 ... pop_counter_; // 原子操作無(wú)需額外鎖 return res; } size_t get_pop_count() const { return pop_counter_.load(std::memory_order_relaxed); } };std::atomic保證了該變量的讀寫(xiě)是原子的并且通過(guò)指定內(nèi)存序如std::memory_order_relaxed,std::memory_order_acquire,std::memory_order_release可以控制線(xiàn)程間的內(nèi)存可見(jiàn)性實(shí)現(xiàn)更精細(xì)的同步。但對(duì)于復(fù)雜的數(shù)據(jù)結(jié)構(gòu)僅靠atomic是不夠的。9. 測(cè)試與驗(yàn)證策略編寫(xiě)線(xiàn)程安全代碼測(cè)試至關(guān)重要但也非常困難。因?yàn)閿?shù)據(jù)競(jìng)爭(zhēng)和死鎖問(wèn)題往往是偶發(fā)的。測(cè)試建議單元測(cè)試單線(xiàn)程首先確保你的封裝在單線(xiàn)程下的行為與底層STL容器一致。壓力測(cè)試多線(xiàn)程使用std::thread創(chuàng)建大量生產(chǎn)者線(xiàn)程和消費(fèi)者線(xiàn)程對(duì)隊(duì)列進(jìn)行密集的push和pop。運(yùn)行足夠長(zhǎng)的時(shí)間比如幾分鐘并檢查最終狀態(tài)是否正確例如所有push進(jìn)去的元素都被pop出來(lái)了沒(méi)有丟失或重復(fù)?？梢允褂胹td::atomic計(jì)數(shù)器來(lái)跟蹤生產(chǎn)和消費(fèi)的數(shù)量。使用線(xiàn)程消毒劑Thread Sanitizer在GCC/Clang中編譯時(shí)添加-fsanitizethread選項(xiàng)。在運(yùn)行時(shí)它能檢測(cè)出數(shù)據(jù)競(jìng)爭(zhēng)、死鎖等問(wèn)題。這是發(fā)現(xiàn)并發(fā)bug的利器。靜態(tài)分析工具一些現(xiàn)代靜態(tài)分析工具也能對(duì)潛在的并發(fā)問(wèn)題進(jìn)行提示。模糊測(cè)試Fuzz Testing隨機(jī)生成不同的線(xiàn)程操作序列長(zhǎng)時(shí)間運(yùn)行試圖觸發(fā)隱藏的競(jìng)態(tài)條件。一個(gè)簡(jiǎn)單的隊(duì)列壓力測(cè)試示例#include iostream #include vector #include thread #include atomic #include cassert #include “ThreadSafeQueue.hpp” // 你的頭文件 void test_queue() { ThreadSafeQueueint queue; std::atomicint producer_count{0}; std::atomicint consumer_count{0}; const int num_items 100000; const int num_producers 4; const int num_consumers 4; std::vectorstd::thread producers, consumers; // 啟動(dòng)生產(chǎn)者 for(int i0; inum_producers; i) { producers.emplace_back([queue, producer_count, num_items](){ for(int j0; jnum_items/num_producers; j) { queue.push(j); producer_count.fetch_add(1, std::memory_order_relaxed); } }); } // 啟動(dòng)消費(fèi)者 for(int i0; inum_consumers; i) { consumers.emplace_back([queue, consumer_count, num_items](){ int value; while(consumer_count.load(std::memory_order_relaxed) num_items) { if(queue.try_pop(value)) { consumer_count.fetch_add(1, std::memory_order_relaxed); } } }); } // 等待所有線(xiàn)程結(jié)束 for(auto t : producers) t.join(); for(auto t : consumers) t.join(); std::cout “Produced: “ producer_count.load() “\n”; std::cout “Consumed: “ consumer_count.load() “\n”; assert(producer_count num_items); assert(consumer_count num_items); assert(queue.empty()); std::cout “Test passed!\n”; } int main() { test_queue(); return 0; }這個(gè)測(cè)試創(chuàng)建了多個(gè)生產(chǎn)者和消費(fèi)者并驗(yàn)證了所有生產(chǎn)的數(shù)據(jù)都被消費(fèi)了且隊(duì)列最終為空。這是一個(gè)基本的正確性測(cè)試。10. 總結(jié)與個(gè)人體會(huì)從頭實(shí)現(xiàn)一遍這些基于鎖的線(xiàn)程安全容器是一個(gè)“知其所以然”的過(guò)程。它強(qiáng)迫你去思考鎖應(yīng)該加在哪里鎖的粒度多大合適接口如何設(shè)計(jì)才能既安全又易用拷貝和移動(dòng)語(yǔ)義在并發(fā)下如何工作異常安全如何保證我個(gè)人在實(shí)際項(xiàng)目中的體會(huì)是不要輕易自己造輪子。對(duì)于大多數(shù)應(yīng)用場(chǎng)景標(biāo)準(zhǔn)庫(kù)的std::sync相關(guān)容器如std::sync::MutexT在Rust中C標(biāo)準(zhǔn)庫(kù)沒(méi)有直接提供或成熟的第三方并發(fā)庫(kù)如Intel TBB、Facebook Folly中的并發(fā)容器是更優(yōu)的選擇。它們經(jīng)過(guò)了更嚴(yán)格的測(cè)試和性能優(yōu)化。那么這個(gè)練習(xí)的意義何在在于理解原理和邊界。當(dāng)你使用一個(gè)現(xiàn)成的線(xiàn)程安全隊(duì)列時(shí)你知道它的pop操作在隊(duì)列為空時(shí)會(huì)阻塞是因?yàn)閮?nèi)部用了條件變量。當(dāng)你看到性能分析中鎖競(jìng)爭(zhēng)激烈時(shí)你能想到可能是鎖粒度過(guò)粗。當(dāng)你在設(shè)計(jì)一個(gè)需要高度并發(fā)的系統(tǒng)時(shí)你能判斷在什么情況下需要引入無(wú)鎖數(shù)據(jù)結(jié)構(gòu)而不是盲目使用。最后再分享一個(gè)小技巧在C17及以上可以考慮使用std::scoped_lock來(lái)代替std::lock_guard因?yàn)樗芙邮芏鄠€(gè)互斥量并且使用std::lock的算法來(lái)避免死鎖語(yǔ)法更簡(jiǎn)潔安全。例如拷貝構(gòu)造函數(shù)可以寫(xiě)成ThreadSafeStack(const ThreadSafeStack other) { std::scoped_lock lock(mutex_, other.mutex_); // C17 data_ other.data_; }并發(fā)編程是C中最有挑戰(zhàn)性也最有趣的部分之一。從一把粗鎖開(kāi)始理解其利弊再逐步探索更精細(xì)的同步機(jī)制這條學(xué)習(xí)路徑是扎實(shí)而有效的。希望這篇長(zhǎng)文和這些代碼示例能成為你征服C并發(fā)世界的一塊堅(jiān)實(shí)墊腳石。