當析構函數遇到多線程
── C++ 中線程安全的對象回調
陳碩 (giantchen_AT_gmail)
本文 PDF 下載： http://www.shnenglu.com/Files/Solstice/dtor_meets_mt.pdf
摘要
編寫線程安全的類不是難事，用同步原語保護內部狀態即可。但是對象的生與死不能由對象自身擁有的互斥器來保護。如何保證即將析構對象 x 的時候，不會有另一個線程正在調用 x 的成員函數？或者說，如何保證在執行 x 的成員函數期間，對象 x 不會在另一個線程被析構？如何避免這種競態條件是 C++ 多線程編程面臨的基本問題，可以借助 boost 的 shared_ptr 和 weak_ptr 完美解決。這也是實現線程安全的 Observer 模式的必備技術。
本文源自我在 2009 年 12 月上海 C++ 技術大會的一場演講《當析構函數遇到多線程》，內容略有增刪。原始 PPT 可從 http://download.csdn.net/source/1982430 下載，或者在 http://www.docin.com/p-41918023.html 直接觀看。
本文讀者應具有 C++ 多線程編程經驗，熟悉互斥器、競態條件等概念，了解智能指針，知道 Observer 設計模式。
目錄
1 多線程下的對象生命期管理 2
線程安全的定義 3
Mutex 與 MutexLock 3
一個線程安全的 Counter 示例 3
2 對象的創建很簡單 4
3 銷毀太難 5
Mutex 不是辦法 5
作為數據成員的 Mutex 6
4 線程安全的 Observer 有多難？ 6
5 一些啟發 8
原始指針有何不妥？ 8
一個“解決辦法” 8
一個更好的解決辦法 9
一個萬能的解決方案 9
6 神器 shared_ptr/weak_ptr 9
7 插曲：系統地避免各種指針錯誤 10
8 應用到 Observer 上 11
解決了嗎？ 11
9 再論 shared_ptr 的線程安全 12
10 shared_ptr 技術與陷阱 13
對象池 15
enable_shared_from_this 16
弱回調 17
11 替代方案？ 19
其他語言怎么辦 19
12 心得與總結 20
總結 20
13 附錄：Observer 之謬 21
14 后記 21
1 多線程下的對象生命期管理
與其他面向對象語言不同，C++ 要求程序員自己管理對象的生命期，這在多線程環境下顯得尤為困難。當一個對象能被多個線程同時看到，那么對象的銷毀時機就會變得模糊不清，可能出現多種競態條件：
l 在即將析構一個對象時，從何而知是否有另外的線程正在執行該對象的成員函數？
l 如何保證在執行成員函數期間，對象不會在另一個線程被析構？
l 在調用某個對象的成員函數之前，如何得知這個對象還活著？它的析構函數會不會剛執行到一半？
解決這些 race condition 是 C++ 多線程編程面臨的基本問題。本文試圖以 shared_ptr 一勞永逸地解決這些問題，減輕 C++ 多線程編程的精神負擔。
線程安全的定義
依據《Java 并發編程實踐》/《Java Concurrency in Practice》一書，一個線程安全的 class 應當滿足三個條件：
l 從多個線程訪問時，其表現出正確的行為
l 無論操作系統如何調度這些線程，無論這些線程的執行順序如何交織
l 調用端代碼無需額外的同步或其他協調動作
依據這個定義，C++ 標準庫里的大多數類都不是線程安全的，無論 std::string 還是 std::vector 或 std::map，因為這些類通常需要在外部加鎖。
Mutex 與 MutexLock
為了便于后文討論，先約定兩個工具類。我相信每個寫C++ 多線程程序的人都實現過或使用過類似功能的類，代碼從略。
Mutex 封裝臨界區（Critical secion），這是一個簡單的資源類，用 RAII 手法 [CCS:13]封裝互斥器的創建與銷毀。臨界區在 Windows 上是 CRITICAL_SECTION，是可重入的；在 Linux 下是 pthread_mutex_t，默認是不可重入的。Mutex 一般是別的 class 的數據成員。
MutexLock 封裝臨界區的進入和退出，即加鎖和解鎖。MutexLock 一般是個棧上對象，它的作用域剛好等于臨界區域。它的構造函數原型為 explicit MutexLock::MutexLock(Mutex& m);
這兩個 classes 都不允許拷貝構造和賦值。
一個線程安全的 Counter 示例
編寫單個的線程安全的 class 不算太難，只需用同步原語保護其內部狀態。例如下面這個簡單的計數器類 Counter：
class Counter : boost::noncopyable
{
// copy-ctor and assignment should be private by default for a class.
public:
Counter(): value_(0) {}
int64_t value() const;
int64_t increase();
int64_t decrease();
private:
int64_t value_;
mutable Mutex mutex_;
}
int64_t Counter::value() const
{
MutexLock lock(mutex_);
return value_;
}
int64_t Counter::increase()
{
MutexLock lock(mutex_);
int64_t ret = value_++;
return ret;
}
// In a real world, atomic operations are perferred.
// 當然在實際項目中，這個 class 用原子操作更合理，這里用鎖僅僅為了舉例。
這個 class 很直白，一看就明白，也容易驗證它是線程安全的。注意到它的 mutex_ 成員是 mutable 的，意味著 const 成員函數如 Counter::value() 也能直接使用 non-const 的 mutex_。
盡管這個 Counter 本身毫無疑問是線程安全的，但如果 Counter 是動態創建的并透過指針來訪問，前面提到的對象銷毀的 race condition 仍然存在。
2 對象的創建很簡單
對象構造要做到線程安全，惟一的要求是在構造期間不要泄露 this 指針，即
l 不要在構造函數中注冊任何回調
l 也不要在構造函數中把 this 傳給跨線程的對象
l 即便在構造函數的最后一行也不行
之所以這樣規定，是因為在構造函數執行期間對象還沒有完成初始化，如果 this 被泄露 (escape) 給了其他對象（其自身創建的子對象除外），那么別的線程有可能訪問這個半成品對象，這會造成難以預料的后果。
// 不要這么做 Don't do this.
class Foo : public Observer
{
public:
Foo(Observable* s) {
s->register(this); // 錯誤
}
virtual void update();
};
// 要這么做 Do this.
class Foo : public Observer
{
// ...
void observe(Observable* s) { // 另外定義一個函數，在構造之后執行
s->register(this);
}
};
Foo* pFoo = new Foo;
Observable* s = getIt();
pFoo->observe(s); // 二段式構造
這也說明，二段式構造——即構造函數+initialize()——有時會是好辦法，這雖然不符合 C++ 教條，但是多線程下別無選擇。另外，既然允許二段式構造，那么構造函數不必主動拋異常，調用端靠 initialize() 的返回值來判斷對象是否構造成功，這能簡化錯誤處理。
即使構造函數的最后一行也不要泄露 this，因為 Foo 有可能是個基類，基類先于派生類構造，執行完 Foo::Foo() 的最后一行代碼會繼續執行派生類的構造函數，這時 most-derived class 的對象還處于構造中，仍然不安全。
相對來說，對象的構造做到線程安全還是比較容易的，畢竟曝光少，回頭率為 0。而析構的線程安全就不那么簡單，這也是本文關注的焦點。
3 銷毀太難
對象析構，這在單線程里不會成為問題，最多需要注意避免空懸指針（和野指針）。而在多線程程序中，存在了太多的競態條件。對一般成員函數而言，做到線程安全的辦法是讓它們順次執行，而不要并發執行，也就是讓每個函數的臨界區不重疊。這是顯而易見的，不過有一個隱含條件或許不是每個人都能立刻想到：函數用來保護臨界區的互斥器本身必須是有效的。而析構函數破壞了這一假設，它會把互斥器銷毀掉。悲劇啊！
Mutex 不是辦法
Mutex 只能保證函數一個接一個地執行，考慮下面的代碼，它試圖用互斥鎖來保護析構函數：
Foo::~Foo() { MutexLock lock(mutex_); // free internal state (1) } |
void Foo::update() { MutexLock lock(mutex_); // (2) // make use of internal state } |
extern Foo* x; // visible by all threads |
// thread A delete x; x = NULL; // helpless |
// thread B if (x) { x->update(); } |
有 A 和 B 兩個線程，線程 A 即將銷毀對象 x，而線程 B 正準備調用 x->update()。盡管線程 A 在銷毀對象之后把指針置為了 NULL，盡管線程 B 在調用 x 的成員函數之前檢查了指針 x 的值，還是無法避免一種 race condition：
1. 線程 A 執行到了析構函數的 (1) 處，已經持有了互斥鎖，即將繼續往下執行
2. 線程 B 通過了 if (x) 檢測，阻塞在 (2) 處
接下來會發生什么，只有天曉得。因為析構函數會把 mutex_ 銷毀，那么 (2) 處有可能永遠阻塞下去，有可能進入“臨界區”然后 core dump，或者發生其他更糟糕的情況。
這個例子至少說明 delete 對象之后把指針置為 NULL 根本沒用，如果一個程序要靠這個來防止二次釋放，說明代碼邏輯出了問題。
作為數據成員的 Mutex
前面的例子說明，作為 class 數據成員的 Mutex 只能用于同步本 class 的其他數據成員的讀和寫，它不能保護安全地析構。因為成員 mutex 的生命期最多與對象一樣長，而析構動作可說是發生在對象身故之后（或者身亡之時）。另外，對于基類對象，那么調用到基類析構函數的時候，派生類對象的那部分已經析構了，那么基類對象擁有的 mutex 不能保護整個析構過程。再說，析構過程本來也不需要保護，因為只有別的線程都訪問不到這個對象時，析構才是安全的，否則會有第 1 節談到的競態條件發生。
另外如果要同時讀寫本 class 的兩個對象，有潛在的死鎖可能，見 PPT 第 12 頁的 swap() 和 operator=()。
4 線程安全的 Observer 有多難？
一個動態創建的對象是否還活著，光看指針（引用也一樣）是看不出來的。指針就是指向了一塊內存，這塊內存上的對象如果已經銷毀，那么就根本不能訪問 [CCS:99]（就像 free 之后的地址不能訪問一樣），既然不能訪問又如何知道對象的狀態呢？換句話說，判斷一個指針是不是野指針沒有高效的辦法。（萬一原址又創建了一個新的對象呢？再萬一這個新的對象的類型異于老的對象呢？）
在面向對象程序設計中，對象的關系主要有三種：composition, aggregation 和 association。Composition（組合）關系在多線程里不會遇到什么麻煩，因為對象 x 的生命期由其惟一的擁有者 owner 控制，owner 析構的時候，會把 x 也析構掉。從形式上看，x 是 owner 的數據成員，或者 scoped_ptr 成員，或者 owner 持有的容器的元素。
后兩種關系在 C++ 里比較難辦，處理不好就會造成內存泄漏或重復釋放。Association（關聯/聯系）是一種很寬泛的關系，它表示一個對象 a 用到了另一個對象 b，調用了后者的成員函數。從代碼形式上看，a 持有 b 的指針（或引用），但是 b 的生命期不由 a 單獨控制。Aggregation（聚合）關系從形式上看與 association 相同，除了 a 和 b 有邏輯上的整體與部分關系。如果 b 是動態創建的并在整個程序結束前有可能被釋放，那么就會出現第 1 節談到的競態條件。
那么似乎一個簡單的解決辦法是：只創建不銷毀。程序使用一個對象池來暫存用過的對象，下次申請新對象時，如果對象池里有存貨，就重復利用現有的對象，否則就新建一個。對象用完了，不是直接釋放掉，而是放回池子里。這個辦法當然有其自身的很多缺點，但至少能避免訪問失效對象的情況發生。
這種山寨辦法的問題有：
l 對象池的線程安全，如何安全地完整地把對象放回池子里，不會出現“部分放回”的競態？（線程 A 認為對象 x 已經放回了，線程 B 認為對象 x 還活著）
l thread contention，這個集中化的對象池會不會把多線程并發的操作串行化？
l 如果共享對象的類型不止一種，那么是重復實現對象池還是使用類模板？
l 會不會造成內存泄露與分片？因為對象池占用的內存只增不減，而且多個對象池不能共享內存（想想為何）。
回到正題上來，如果對象 x 注冊了任何非靜態成員函數回調，那么必然在某處持有了指向 x 的指針，這就暴露在了 race condition 之下。
一個典型的場景是 Observer 模式。
class Observer
{
public:
virtual ~Observer() { }
virtual void update() = 0;
};
class Observable
{
public:
void register(Observer* x);
void unregister(Observer* x);
void notifyObservers() {
foreach Observer* x { // 這行是偽代碼
x->update(); // (3)
}
}
// ...
}
當 Observable 通知每一個 Observer 時 (3)，它從何得知 Observer 對象 x 還活著？
要不在 Observer 的析構函數里解注冊 (unregister)？恐難奏效。
struct Observer
{
virtual ~Observer() { }
virtual void update() = 0;
void observe(Observable* s) {
s->register(this);
subject_ = s;
}
virtual ~Observer() {
// (4)
subject_->unregister(this);
}
Observable* subject_;
};
我們試著讓 Observer 的析構函數去 unregister(this)，這里有兩個 race conditions。其一：(4) 處如何得知 subject_ 還活著？其二：就算 subject_ 指向某個永久存在的對象，那么還是險象環生：
1. 線程 A 執行到 (4) 處，還沒有來得及 unregister 本對象
2. 線程 B 執行到 (3) 處，x 正好指向是 (4) 處正在析構的對象
那么悲劇又發生了，既然 x 所指的 Observer 對象正在析構，調用它的任何非靜態成員函數都是不安全的，何況是虛函數（C++ 標準對在構造函數和析構函數中調用虛函數的行為有明確的規定，但是沒有考慮并發調用的情況）。更糟糕的是，Observer 是個基類，執行到 (4) 處時，派生類對象已經析構掉了，這時候整個對象處于將死未死的狀態，core dump 恐怕是最幸運的結果。
這些 race condition 似乎可以通過加鎖來解決，但在哪兒加鎖，誰持有這些互斥鎖，又似乎不是那么顯而易見的。要是有什么活著的對象能幫幫我們就好了，它提供一個 isAlive() 之類的程序函數，告訴我們那個對象還在不在。可惜指針和引用都不是對象，它們是內建類型。
5 一些啟發
指向對象的原始指針 (raw pointer) 是壞的，尤其當暴露給別的線程時。Observable 應當保存的不是原始的 Observer*，而是別的什么東西，能分別 Observer 對象是否存活。類似地，如果 Observer 要在析構函數里解注冊（這雖然不能解決前面提到的 race condition，但是在析構函數里打掃戰場還是應該的），那么 subject_ 的類型也不能是原始的 Observable*。
有經驗的 C++ 程序員或許會想到用智能指針，沒錯，這是正道，但也沒那么簡單，有些關竅需要注意。這兩處直接使用 shared_ptr 是不行的，會造成循環引用，直接造成資源泄漏。別著急，后文會一一講到。
圖片請看 PDF 版。
原始指針有何不妥？
有兩個指針 p1 和 p2，指向堆上的同一個對象 Object，p1 和 p2 位于不同的線程中（左圖）。假設線程 A 透過 p1 指針將對象銷毀了（盡管把 p1 置為了 NULL），那么 p2 就成了空懸指針（右圖）。這是一種典型的 C/C++ 內存錯誤。
要想安全地銷毀對象，最好讓在別人（線程）都看不到的情況下，偷偷地做。
一個“解決辦法”
一個解決空懸指針的辦法是，引入一層間接性，讓 p1 和 p2 所指的對象永久有效。比如下圖的 proxy 對象，這個對象，持有一個指向 Object 的指針。（從 C 語言的角度，p1 和 p2 都是二級指針。）
當銷毀 Object 之后，proxy 對象繼續存在，其值變為 0。而 p2 也沒有變成空懸指針，它可以通過查看 proxy 的內容來判斷 Object 是否還活著。要線程安全地釋放 Object 也不是那么容易，race condition 依舊存在。比如 p2 看第一眼的時候 proxy 不是零，正準備去調用 Object 的成員函數，期間對象已經被 p1 銷毀了。
問題在于，何時釋放 proxy 指針呢？
一個更好的解決辦法
為了安全地釋放 proxy，我們可以引入引用計數，再把 p1 和 p2 都從指針變成對象 sp1 和 sp2。proxy 現在有兩個成員，指針和計數器。
1. 一開始，有兩個引用，計數值為 2，
2. sp1 析構了，引用計數的值減為 1，
3. sp2 也析構了，引用計數的值為 0，可以安全地銷毀 proxy 和 Object 了。
慢著！這不就是引用計數型智能指針嗎？
一個萬能的解決方案
引入另外一層間接性，another layer of indirection，用對象來管理共享資源（如果把 Object 看作資源的話），亦即 handle/body 手法 (idiom)。當然，編寫線程安全、高效的引用計數 handle 的難度非凡，作為一名謙卑的程序員，用現成的庫就行。
萬幸，C++ 的 tr1 標準庫里提供了一對神兵利器，可助我們完美解決這個頭疼的問題。
6 神器 shared_ptr/weak_ptr
shared_ptr 是引用計數型智能指針，在 boost 和 std::tr1 里都有提供，現代主流的 C++ 編譯器都能很好地支持。shared_ptr<T> 是一個類模板 (class template)，它只有一個類型參數，使用起來很方便。引用計數的是自動化資源管理的常用手法，當引用計數降為 0 時，對象（資源）即被銷毀。weak_ptr 也是一個引用計數型智能指針，但是它不增加引用次數，即弱 (weak) 引用。
shared_ptr 的基本用法和語意請參考手冊或教程，本文從略，這里談幾個關鍵點。
l shared_ptr 控制對象的生命期。shared_ptr 是強引用（想象成用鐵絲綁住堆上的對象），只要有一個指向 x 對象的 shared_ptr 存在，該 x 對象就不會析構。當指向對象 x 的最后一個 shared_ptr 析構或 reset 的時候，x 保證會被銷毀。
l weak_ptr 不控制對象的生命期，但是它知道對象是否還活著（想象成用棉線輕輕拴住堆上的對象）。如果對象還活著，那么它可以提升 (promote) 為有效的 shared_ptr；如果對象已經死了，提升會失敗，返回一個空的 shared_ptr。“提升”行為是線程安全的。
l shared_ptr/weak_ptr 的“計數”在主流平臺上是原子操作，沒有用鎖，性能不俗。
l shared_ptr/weak_ptr 的線程安全級別與 string 等 STL 容器一樣，后面還會講。
7 插曲：系統地避免各種指針錯誤
我同意孟巖說的“大部分用 C 寫的上規模的軟件都存在一些內存方面的錯誤，需要花費大量的精力和時間把產品穩定下來。”內存方面的問題在 C++ 里很容易解決，我第一次也是最后一次見到別人的代碼里有內存泄漏是在 2004 年實習那會兒，自己寫的C++ 程序從來沒有出現過內存方面的問題。
C++ 里可能出現的內存問題大致有這么幾個方面：
1. 緩沖區溢出
2. 空懸指針/野指針
3. 重復釋放
4. 內存泄漏
5. 不配對的 new[]/delete
6. 內存碎片
正確使用智能指針能很輕易地解決前面 5 個問題，解決第 6 個問題需要別的思路，我會另文探討。
1. 緩沖區溢出 ⇒ 用 vector/string 或自己編寫 Buffer 類來管理緩沖區，自動記住用緩沖區的長度，并通過成員函數而不是裸指針來修改緩沖區。
2. 空懸指針/野指針 ⇒ 用 shared_ptr/weak_ptr，這正是本文的主題
3. 重復釋放 ⇒ 用 scoped_ptr，只在對象析構的時候釋放一次
4. 內存泄漏 ⇒ 用 scoped_ptr，對象析構的時候自動釋放內存
5. 不配對的 new[]/delete ⇒ 把 new[] 統統替換為 vector/scoped_array
正確使用上面提到的這幾種智能指針并不難，其難度大概比學習使用 vector/list 這些標準庫組件還要小，與 string 差不多，只要花一周的時間去適應它，就能信手拈來。我覺得，在現代的 C++ 程序中一般不會出現 delete 語句，資源（包括復雜對象本身）都是通過對象（智能指針或容器）來管理的，不需要程序員還為此操心。
需要注意一點：scoped_ptr/shared_ptr/weak_ptr 都是值語意，要么是棧上對象，或是其他對象的直接數據成員，或是標準庫容器里的元素。幾乎不會有下面這種用法：
shared_ptr<Foo>* pFoo = new shared_ptr<Foo>(new Foo); // WRONG semantic
還要注意，如果這幾種智能指針是對象 x 的數據成員，而它的模板參數 T 是個 incomplete 類型，那么 x 的析構函數不能是默認的或內聯的，必須在 .cpp 文件里邊顯式定義，否則會有編譯錯或運行錯。（原因請見 boost::checked_delete）
8 應用到 Observer 上
既然透過 weak_ptr 能探查對象的生死，那么 Observer 模式的競態條件就很容易解決，只要讓 Observable 保存 weak_ptr<Observer> 即可：
class Observable // not 100% thread safe!
{
public:
void register(weak_ptr<Observer> x);
void unregister(weak_ptr<Observer> x); // 可用 std::remove/vector::erase 實現
void notifyObservers()
{
MutexLock lock(mutex_);
Iterator it = observers_.begin();
while (it != observers_.end()) {
shared_ptr<Observer> obj(it->lock()); // 嘗試提升，這一步是線程安全的
if (obj) {
// 提升成功，現在引用計數值至少為 2 （想想為什么？）
obj->update(); // 沒有競態條件，因為 obj 在棧上，對象不可能在本作用域內銷毀
++it;
} else {
// 對象已經銷毀，從容器中拿掉 weak_ptr
it = observers_.erase(it);
}
}
}
private:
std::vector<weak_ptr<Observer> > observers_; // (5)
mutable Mutex mutex_;
};
就這么簡單。前文代碼 (3) 處的競態條件已經彌補了。
解決了嗎？
把 Observer* 替換為 weak_ptr<Observer> 部分解決了 Observer 模式的線程安全，但還有幾個疑點：
不靈活，強制要求 Observer 必須以 shared_ptr 來管理；
不是完全線程安全，Observer 的析構函數會調用 subject_->unregister(this)，萬一 subject_ 已經不復存在了呢？為了解決它，又要求 Observable 本身是用 shared_ptr 管理的，并且 subject_ 是個 weak_ptr<Observable>；
線程爭用 (thread contention)，即 Observable 的三個成員函數都用了互斥器來同步，這會造成 register 和 unregister 等待 notifyObservers，而后者的執行時間是無上限的，因為它同步回調了用戶提供的 update() 函數。我們希望 register 和 unregister 的執行時間不會超過某個固定的上限，以免即便殃及無辜群眾。
死鎖，萬一 update() 虛函數中調用了 (un)register 呢？如果 mutex_ 是不可重入的，那么會死鎖；如果 mutex_ 是可重入的，程序會面臨迭代器失效（core dump 是最好的結果），因為 vector observers_ 在遍歷期間被無意識地修改了。這個問題乍看起來似乎沒有解決辦法，除非在文檔里做要求。個辦法是：用可重入的 mutex_，把容器換為 std::list，并把 ++it 往前挪一行。）
這些問題留到本文附錄中去探討，每個問題都是能解決的。
我個人傾向于使用不可重入的 Mutex，例如 pthreads 默認提供的那個，因為“要求 Mutex 可重入”本身往往意味著設計上出了問題。Java 的 intrinsic lock 是可重入的，因為要允許 synchronized 方法相互調用，我覺得這也是無奈之舉。
思考：如果把 (5) 處改為 vector<shared_ptr<Observer> > observers_;，會有什么后果？
9 再論 shared_ptr 的線程安全
雖然我們借 shared_ptr 來實現線程安全的對象釋放，但是 shared_ptr 本身不是 100% 線程安全的。它的引用計數本身是安全且無鎖的，但對象的讀寫則不是，因為 shared_ptr 有兩個數據成員，讀寫操作不能原子化。
根據文檔，shared_ptr 的線程安全級別和內建類型、標準庫容器、string 一樣，即：
l 一個 shared_ptr 實體可被多個線程同時讀取；
l 兩個的 shared_ptr 實體可以被兩個線程同時寫入，“析構”算寫操作；
l 如果要從多個線程讀寫同一個 shared_ptr 對象，那么需要加鎖。
請注意，這是 shared_ptr 對象本身的線程安全級別，不是它管理的對象的線程安全級別。
要在多個線程中同時訪問同一個 shared_ptr，正確的做法是：
shared_ptr<Foo> globalPtr;
Mutex mutex; // No need for ReaderWriterLock
void doit(const shared_ptr<Foo>& pFoo);
globalPtr 能被多個線程看到，那么它的讀寫需要加鎖。注意我們不必用讀寫鎖，而只用最簡單的互斥鎖，這是為了性能考慮，因為臨界區非常小，用互斥鎖也不會阻塞并發讀。
void read()
{
shared_ptr<Foo> ptr;
{
MutexLock lock(mutex);
ptr = globalPtr; // read globalPtr
}
// use ptr since here
doit(ptr);
}
寫入的時候也要加鎖：
void write()
{
shared_ptr<Foo> newptr(new Foo);
{
MutexLock lock(mutex);
globalPtr = newptr; // write to globalPtr
}
// use newptr since here
doit(newptr);
}
注意到 read() 和 write() 在臨界區之外都沒有再訪問 globalPtr，而是用了一個指向同一 Foo 對象的棧上 shared_ptr local copy。下面會談到，只要有這樣的 local copy 存在，shared_ptr 作為函數參數傳遞時不必復制，用 reference to const 即可。
10 shared_ptr 技術與陷阱
意外延長對象的生命期。shared_ptr 是強引用（鐵絲綁的），只要有一個指向 x 對象的 shared_ptr 存在，該對象就不會析構。而 shared_ptr 又是允許拷貝構造和賦值的（否則引用計數就無意義了），如果不小心遺留了一個拷貝，那么對象就永世長存了。例如前面提到如果把 (5) 處 observers_ 的類型改為 vector<shared_ptr<Observer> >，那么除非手動調用 unregister，否則 Observer 對象永遠不會析構。即便它的析構函數會調用 unregister，但是不去 unregister 就不會調用析構函數，這變成了雞與蛋的問題。這也是 Java 內存泄露的常見原因。
另外一個出錯的可能是 boost::bind，因為 boost:;bind 會把參數拷貝一份，如果參數是個 shared_ptr，那么對象的生命期就不會短于 boost::function 對象：
class Foo
{
void doit();
};
boost::function<void()> func;
shared_ptr<Foo> pFoo(new Foo);
func = bind(&Foo::doit, pFoo); // long life foo
這里 func 對象持有了 shared_ptr<Foo> 的一份拷貝，有可能會不經意間延長倒數第二行創建的 Foo 對象的生命期。
函數參數。因為要修改引用計數（而且拷貝的時候通常要加鎖），shared_ptr 的拷貝開銷比拷貝原始指針要高，但是需要拷貝的時候并不多。多數情況下它可以以 reference to const 方式傳遞，一個線程只需要在最外層函數有一個實體對象，之后都可以用 reference to const 來使用這個 shared_ptr。例如有幾個個函數都要用到 Foo 對象：
void save(const shared_ptr<Foo>& pFoo);
void validateAccount(const Foo& foo);
bool validate(const shared_ptr<Foo>& pFoo)
{
// ...
validateAccount(*pFoo);
// ...
}
那么在通常情況下，
void onMessage(const string& buf)
{
shared_ptr<Foo> pFoo(new Foo(buf)); // 只要在最外層持有一個實體，安全不成問題
if (validate(pFoo)) {
save(pFoo);
}
}
遵照這個規則，基本上不會遇到反復拷貝 shared_ptr 導致的性能問題。另外由于 pFoo 是棧上對象，不可能被別的線程看到，那么讀取始終是線程安全的。
析構動作在創建時被捕獲。這是一個非常有用的特性，這意味著：
l 虛析構不再是必須的。
l shared_ptr<void> 可以持有任何對象，而且能安全地釋放
l shared_ptr 對象可以安全地跨越模塊邊界，比如從 DLL 里返回，而不會造成從模塊 A 分配的內存在模塊 B 里被釋放這種錯誤。
l 二進制兼容性，即便 Foo 對象的大小變了，那么舊的客戶代碼任然可以使用新的動態庫，而無需重新編譯（這要求 Foo 的頭文件中不出現訪問對象的成員的 inline函數）。
l 析構動作可以定制。
這個特性的實現比較巧妙，因為 shared_ptr<T> 只有一個模板參數，而“析構行為”可以是函數指針，仿函數 (functor) 或者其他什么東西。這是泛型編程和面向對象編程的一次完美結合。有興趣的同學可以參考 Scott Meyers 的文章。
這個技術在后面的對象池中還會用到。
析構所在的線程。對象的析構是同步的，當最后一個指向 x 的 shared_ptr 離開其作用域的時候，x 會同時在同一個線程析構。這個線程不一定是對象誕生的線程。這個特性是把雙刃劍：如果對象的析構比較耗時，那么可能會拖慢關鍵線程的速度（如果最后一個 shared_ptr 引發的析構發生在關鍵線程）；同時，我們可以用一個單獨的線程來專門做析構，通過一個 BlockingQueue<shared_ptr<void> > 把對象的析構都轉移到那個專用線程，從而解放關鍵線程。
現成的 RAII handle。我認為 RAII （資源獲取即初始化）是 C++ 語言區別與其他所有編程語言的最重要的手法，一個不懂 RAII 的 C++ 程序員不是一個合格的 C++ 程序員。原來的 C++ 教條是“new 和 delete 要配對，new 了之后要記著 delete”，如果使用 RAII，要改成“每一個明確的資源配置動作（例如 new）都應該在單一語句中執行，并在該語句中立刻將配置獲得的資源交給 handle 對象（如 shared_ptr），程序中一般不出現 delete”（出處見腳注 1）。shared_ptr 是管理共享資源的利器，需要注意避免循環引用，通常的做法是 owner 持有指向 A 的 shared_ptr，A 持有指向 owner 的 weak_ptr。
對象池
假設有 Stock 類，代表一只股票的價格。每一只股票有一個惟一的字符串標識，比如 Google 的 key 是 "NASDAQ:GOOG"，IBM 是 "NYSE:IBM"。Stock 對象是個主動對象，它能不斷獲取新價格。為了節省系統資源，同一個程序里邊每一只出現的股票只有一個 Stock 對象，如果多處用到同一只股票，那么 Stock 對象應該被共享。如果某一只股票沒有再在任何地方用到，其對應的 Stock 對象應該析構，以釋放資源，這隱含了“引用計數”。
為了達到上述要求，我們可以設計一個對象池 StockFactory。它的接口很簡單，根據 key 返回 Stock 對象。我們已經知道，在多線程程序中，既然對象可能被銷毀，那么返回 shared_ptr 是合理的。自然地，我們寫出如下代碼。（可惜是錯的）
class StockFactory : boost::noncopyable
{ // questionable code
public:
shared_ptr<Stock> get(const string& key);
private:
std::map<string, shared_ptr<Stock> > stocks_;
mutable Mutex mutex_;
};
get() 的邏輯很簡單，如果在 stocks_ 里找到了 key，就返回 stocks_[key]；否則新建一個 Stock，并存入 stocks_[key]。
細心的讀者或許已經發現這里有一個問題，Stock 對象永遠不會被銷毀，因為 map 里存的是 shared_ptr，始終有鐵絲綁著。那么或許應該仿照前面 Observable 那樣存一個 weak_ptr？比如
class StockFactory : boost::noncopyable
{
public:
shared_ptr<Stock> get(const string& key)
{
shared_ptr<Stock> pStock;
MutexLock lock(mutex_);
weak_ptr<Stock>& wkStock = stocks_[key]; // 如果 key 不存在，會默認構造一個
pStock = wkStock.lock(); // 嘗試把棉線提升為鐵絲
if (!pStock) {
pStock.reset(new Stock(key));
wkStock = pStock; // 這里更新了 stocks_[key]，注意 wkStock 是個引用
}
return pStock;
}
private:
std::map<string, weak_ptr<Stock> > stocks_;
mutable Mutex mutex_;
};
這么做固然 Stock 對象是銷毀了，但是程序里卻出現了輕微的內存泄漏，為什么？
因為 stocks_ 的大小只增不減，stocks_.size() 是曾經存活過的 Stock 對象的總數，即便活的 Stock 對象數目降為 0。或許有人認為這不算泄漏，因為內存并不是徹底遺失不能訪問了，而是被某個標準庫容器占用了。我認為這也算內存泄漏，畢竟是戰場沒有打掃干凈。
其實，考慮到世界上的股票數目是有限的，這個內存不會一直泄漏下去，大不了把每只股票的對象都創建一遍，估計泄漏的內存也只有幾兆。如果這是一個其他類型的對象池，對象的 key 的集合不是封閉的，內存會一直泄漏下去。
解決的辦法是，利用 shared_ptr 的定制析構功能。shared_ptr 的構造函數可以有一個額外的模板類型參數，傳入一個函數指針或仿函數 d，在析構對象時執行 d(p)。shared_ptr 這么設計并不是多余的，因為反正要在創建對象時捕獲釋放動作，始終需要一個 bridge。
template<class Y, class D> shared_ptr::shared_ptr(Y* p, D d);
template<class Y, class D> void shared_ptr::reset(Y* p, D d);
那么我們可以利用這一點，在析構 Stock 對象的同時清理 stocks_。
class StockFactory : boost::noncopyable
{
// in get(), change
// pStock.reset(new Stock(key));
// to
// pStock.reset(new Stock(key),
// boost::bind(&StockFactory::deleteStock, this, _1)); (6)
private:
void deleteStock(Stock* stock)
{
if (stock) {
MutexLock lock(mutex_);
stocks_.erase(stock->key());
}
delete stock; // sorry, I lied
}
// assuming FooCache lives longer than all Foo's ...
// ...
這里我們向 shared_ptr<Stock>::reset() 傳遞了第二個參數，一個 boost::function，讓它在析構 Stock* p 時調用本 StockFactory 對象的 deleteStock 成員函數。
警惕的讀者可能已經發現問題，那就是我們把一個原始的 StockFactory this 指針保存在了 boost::function 里 (6)，這會有線程安全問題。如果這個 StockFactory 先于 Stock 對象析構，那么會 core dump。正如 Observer 在析構函數里去調用 Observable::unregister()，而那時 Observable 對象可能已經不存在了。
當然這也是能解決的，用到下一節的技術。
enable_shared_from_this
StockFactory::get() 把原始指針 this 保存到了 boost::function 中 (6)，如果 StockFactory 的生命期比 Stock 短，那么 Stock 析構時去回調 StockFactory::deleteStock 就會 core dump。似乎我們應該祭出慣用的 shared_ptr 大法來解決對象生命期問題，但是 StockFactory::get() 本身是個成員函數，如何獲得一個 shared_ptr<StockFactory> 對象呢？
有辦法，用 enable_shared_from_this。這是一個模板基類，繼承它，this 就能變身為 shared_ptr。
class StockFactory : public boost::enable_shared_from_this<StockFactory>,
boost::noncopyable
{ /* ... */ };
為了使用 shared_from_this()，要求 StockFactory 對象必須保存在 shared_ptr 里。
shared_ptr<StockFactory> stockFactory(new StockFactory);
萬事俱備，可以從 this 變身 shared_ptr<StockFactory> 了。
shared_ptr<Stock> StockFactory::get(const string& key)
{
// change
// pStock.reset(new Stock(key),
// boost::bind(&StockFactory::deleteStock, this, _1));
// to
pStock.reset(new Stock(key),
boost::bind(&StockFactory::deleteStock,
shared_from_this(),
_1));
// ...
這樣一來，boost::function 里保存了一份 shared_ptr<StockFactory>，可以保證調用 StockFactory::deleteStock 的時候那個 StockFactory 對象還活著。
注意一點，shared_from_this() 不能在構造函數里調用，因為在構造 StockFactory 的時候，它還沒有被交給 shared_ptr 接管。
最后一個問題，StockFactory 的生命期似乎被意外延長了。
弱回調
把 shared_ptr 綁 (bind) 到 boost:function 里，那么回調的時候對象始終存在，是安全的。這同時也延長了對象的生命期，使之不短于 boost:function 對象。
有時候我們需要“如果對象還活著，就調用它的成員函數，否則忽略之”的語意，就像 Observable::notifyObservers() 那樣，我稱之為“弱回調”。這也是可以實現的，利用 weak_ptr，我們可以把 weak_ptr 綁到 boost::function 里，這樣對象的生命期就不會被延長。然后在回調的時候先嘗試提升為 shared_ptr，如果提升成功，說明接受回調的對象還健在，那么就執行回調；如果提升失敗，就不必勞神了。
使用這一技術的完整 StockFactory 代碼如下：
class StockFactory : public boost::enable_shared_from_this<StockFactory>,
boost::noncopyable
{
public:
shared_ptr<Stock> get(const string& key)
{
shared_ptr<Stock> pStock;
MutexLock lock(mutex_);
weak_ptr<Stock>& wkStock = stocks_[key];
pStock = wkStock.lock();
if (!pStock) {
pStock.reset(new Stock(key),
boost::bind(&StockFactory::weakDeleteCallback,
boost::weak_ptr<StockFactory>(shared_from_this()),
_1));
// 上面必須強制把 shared_from_this() 轉型為 weak_ptr，才不會延長生命期
wkStock = pStock;
}
return pStock;
}
private:
static void weakDeleteCallback(boost::weak_ptr<StockFactory> wkFactory,
Stock* stock)
{
shared_ptr<StockFactory> factory(wkFactory.lock()); // 嘗試提升
if (factory) { // 如果 factory 還在，那就清理 stocks_
factory->removeStock(stock);
}
delete stock; // sorry, I lied
}
void removeStock(Stock* stock)
{
if (stock) {
MutexLock lock(mutex_);
stocks_.erase(stock->key());
}
}
private:
std::map<string, weak_ptr<Stock> > stocks_;
mutable Mutex mutex_;
};
兩個簡單的測試：
void testLongLifeFactory()
{
shared_ptr<StockFactory> factory(new StockFactory);
{
shared_ptr<Stock> stock = factory->get("NYSE:IBM");
shared_ptr<Stock> stock2 = factory->get("NYSE:IBM");
assert(stock == stock2);
// stock destructs here
}
// factory destructs here
}
void testShortLifeFactory()
{
shared_ptr<Stock> stock;
{
shared_ptr<StockFactory> factory(new StockFactory);
stock = factory->get("NYSE:IBM");
shared_ptr<Stock> stock2 = factory->get("NYSE:IBM");
assert(stock == stock2);
// factory destructs here
}
// stock destructs here
}
這下完美了，無論 Stock 和 StockFactory 誰先掛掉都不會影響程序的正確運行。
當然，通常 Factory 對象是個 singleton，在程序正常運行期間不會銷毀，這里只是為了展示弱回調技術，這個技術在事件通知中非常有用。
11 替代方案？
除了使用 shared_ptr/weak_ptr，要想在 C++ 里做到線程安全的對象回調與析構，可能的辦法有：
1. 用一個全局的 facade 來代理 Foo 類型對象訪問，所有的 Foo 對象回調和析構都通過這個 facade 來做，也就是把指針替換為 objId/handle。這樣理論上能避免 race condition，但是代價很大。因為要想把這個 facade 做成線程安全，那么必然要用互斥鎖。這樣一來，從兩個線程訪問兩個不同的 Foo 對象也會用到同一個鎖，讓本來能夠并行執行的函數變成了串行執行，沒能發揮多核的優勢。當然，可以像 Java 的 ConcurrentHashMap 那樣用多個 buckets，每個 bucket 分別加鎖，以降低 contention。
2. 第 4 節提到的“只創建不銷毀”手法，實屬無奈之舉。
3. 自己編寫引用計數的智能指針。本質上是重新發明輪子，把 shared_ptr 實現一遍。正確實現線程安全的引用計數智能指針不是一件容易的事情，而高效的實現就更加困難。既然 shared_ptr 已經提供了完整的解決方案，那么似乎沒有理由抗拒它。
4. 將來在 C++ 0x 里有 unique_ptr，能避免引用計數的開銷，或許能在某些場合替換shared_ptr。
其他語言怎么辦
有垃圾回收就好辦。Google 的 Go 語言教程明確指出，沒有垃圾回收的并發編程是困難的（Concurrency is hard without garbage collection）。但是由于指針算術的存在，在 C/C++里實現全自動垃圾回收更加困難。而那些天生具備垃圾回收的語言在并發編程方面具有明顯的優勢，Java 是目前支持并發編程最好的主流語言，它的 util.concurrent 庫和內存模型是 C++ 0x 效仿的對象。
12 心得與總結
學習多線程程序設計遠遠不是看看教程了解 API 怎么用那么簡單，這最多“主要是為了讀懂別人的代碼，如果自己要寫這類代碼，必須專門花時間嚴肅認真系統地學習，嚴禁半桶水上陣”（孟巖）。一般的多線程教程上都會提到要讓加鎖的區域足夠小，這沒錯，問題是如何找出這樣的區域并加鎖，本文第 9 節舉的安全讀寫 shared_ptr 可算是一個例子。
據我所知，目前 C++ 沒有好的多線程領域專著，C 語言有，Java 語言也有。《Java Concurrency in Practice》是我讀過的寫得最好的書，內容足夠新，可讀性和可操作性俱佳。C++ 程序員反過來要向 Java 學習，多少有些諷刺。除了編程書，操作系統教材也是必讀的，至少要完整地學習一本經典教材的相關章節，可從《操作系統設計與實現》、《現代操作系統》、《操作系統概念》任選一本，了解各種同步原語、臨界區、競態條件、死鎖、典型的 IPC 問題等等，防止閉門造車。
分析可能出現的 race condition 不僅是多線程編程基本功，也是設計分布式系統的基本功，需要反復歷練，形成一定的思考范式，并積累一些經驗教訓，才能少犯錯誤。這是一個快速發展的領域，要不斷吸收新知識，才不會落伍。單 CPU 時代的多線程編程經驗到了多 CPU 時代不一定有效，因為多 CPU 能做到真正的并發執行，每個 CPU 看到的事件發生順序不一定完全相同。正如狹義相對論所說的每個觀察者都有自己的時鐘，在不違反因果律的前提下，可能發生十分違反直覺的事情。
盡管本文通篇在講如何安全地使用（包括析構）跨線程的對象，但我建議盡量減少使用跨線程的對象，我贊同縉大師說的“用流水線，生產者-消費者，任務隊列這些有規律的機制，最低限度地共享數據。這是我所知最好的多線程編程的建議了。”
不用跨線程的對象，自然不會遇到本文描述的各種險態。如果迫不得已要用，我希望本文能對您有幫助。
總結
l 原始指針暴露給多個線程往往會造成 race condition 或額外的簿記負擔；
l 統一用 shared_ptr/scoped_ptr 來管理對象的生命期，在多線程中尤其重要；
l shared_ptr 是值語意，當心意外延長對象的生命期。例如 boost::bind 和容器；
l weak_ptr 是 shared_ptr 的好搭檔，可以用作弱回調、對象池等；
l 認真閱讀一遍 boost::shared_ptr 的文檔，能學到很多東西：
http://www.boost.org/doc/libs/release/libs/smart_ptr/shared_ptr.htm
l 保持開放心態，留意更好的解決辦法，比如 unique_ptr。忘掉已被廢棄的 auto_ptr。
shared_ptr 是 tr1 的一部分，即 C++ 標準庫的一部分，值得花一點時間去學習掌握，對編寫現代的 C++ 程序有莫大的幫助。我個人的經驗是，一周左右就能基本掌握各種用法與常見陷阱，比學 STL 還快。網絡上有一些對 shared_ptr 的批評，那可以算作故意誤用的例子，就好比故意訪問失效的迭代器來證明 vector 不安全一樣。
正確使用 shared_ptr 這樣的自動化內存/資源管理器，解放大腦，從此告別內存錯誤。
13 附錄：Observer 之謬
本文第 8 節把 shared_ptr/weak_ptr 應用到 Observer 模式中，部分解決了其線程安全問題。我用 Observer 舉例，因為這是一個廣為人知的設計模式，但是它有本質的問題。
Observer 模式的本質問題在于其面向對象的設計。換句話說，我認為正是面向對象 (OO) 本身造成了 Observer 的缺點。Observer 是基類，這帶來了非常強的耦合，強度僅次于友元。這種耦合不僅限制了成員函數的名字、參數、返回值，還限制了成員函數所屬的類型（必須是 Observer 的派生類）。
Observer 是基類，這意味著如果 Foo 想要觀察兩個類型的事件（比如時鐘和溫度），需要使用多繼承。這還不是最糟糕的，如果要重復觀察同一類型的事件（比如 1 秒鐘一次的心跳和 30 秒鐘一次的自檢），就要用到一些伎倆來 work around，因為不能從一個 Base class 繼承兩次。
現在的語言一般可以繞過 Observer 模式的限制，比如 Java 可以用匿名內部類，Java 7 用 Closure，C# 用 delegate，C++ 用 boost::function/ boost::bind，我在另外一篇博客《以 boost::function 和 boost:bind 取代虛函數》里有更多的講解。
在 C++ 里為了替換 Observer，可以用 Signal/Slots，我指的不是 QT 那種靠語言擴展的實現，而是完全靠標準庫實現的 thread safe、race condition free、thread contention free 的 Signal/Slots，并且不強制要求 shared_ptr 來管理對象，也就是說完全解決了第 8 節列出的 Observer 遺留問題。不過這篇文章已經夠長了，留作下次吧。有興趣的同學可以先預習一下《借 shared_ptr 實現線程安全的 copy-on-write》。
14 后記
《C++ 沉思錄》/《Runminations on C++》中文版的附錄是王曦和孟巖對作者夫婦二人的采訪，在被問到“請給我們三個你們認為最重要的建議”時，Koenig 和 Moo 的第一個建議是“避免使用指針”。我 2003 年讀到這段時，理解不深，覺得固然使用指針容易造成內存方面的問題，但是完全不用也是做不到的，畢竟 C++ 的多態要透過指針或引用來起效。6 年之后重新拾起來，發現大師的觀點何其深刻，不免掩卷長嘆。
這本書詳細地介紹了 handle/body idiom，這是編寫大型 C++ 程序的必備技術，也是實現物理隔離的法寶，值得細讀。
目前來看，用 shared_ptr 來管理資源在國內 C++ 界似乎并不是一種主流做法，很多人排斥智能指針，視為洪水猛獸（這或許受了 auto_ptr 的垃圾設計的影響）。據我所知，很多 C++ 項目還是手動管理內存和資源，因此我覺得有必要把我認為好的做法分享出來，讓更多的人嘗試并采納。我覺得 shared_ptr 對于編寫線程安全的 C++ 程序是至關重要的，不然就得土法煉鋼，自己重新發明輪子。這讓我想起了 2001 年前后 STL 剛剛傳入國內，大家也是很猶豫，覺得它性能不高，使用不便，還不如自己造的容器類。近十年過去了，現在 STL 已經是主流，大家也適應了迭代器、容器、算法、適配器、仿函數這些“新”名詞“新”技術，開始在項目中普遍使用（至少用 vector 代替數組嘛）。我希望，幾年之后人們回頭看這篇文章，覺得“怎么講的都是常識”，那我這篇文章的目的也就達到了。
.全文完 2010/Jan/22初稿 Jan 27 修訂.