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