低調(diào)做技術(shù)__歡迎移步我的獨(dú)立博客 codemaro.com 微博 kevinlynx
6.25.2008
Kevin Lynx
引言
在一個(gè)高效的系統(tǒng)中，我們經(jīng)常會將一些費(fèi)時(shí)的操作轉(zhuǎn)換為異步操作。例如往數(shù)據(jù)庫中寫日志。如果數(shù)據(jù)庫配置在網(wǎng)絡(luò)上，那么往數(shù)據(jù)庫中插入一些日志信息將非常慢(相對于程序其他部分)。
如何轉(zhuǎn)換為異步？
將類似于以上過程轉(zhuǎn)換為異步操作，一個(gè)典型的做法是：建立一個(gè)單獨(dú)的數(shù)據(jù)庫日志線程，一個(gè)線程安全的隊(duì)列。要寫日志時(shí)，只需要往隊(duì)列里放入數(shù)據(jù)，數(shù)據(jù)庫日志線程則從這個(gè)隊(duì)列里取數(shù)據(jù)然后完成寫操作。
大致的過程類似于：
將他們包裝起來
我們很有可能會在同一個(gè)系統(tǒng)中多次遇到類似需要轉(zhuǎn)換為異步操作的地方，如果每一次都手動去創(chuàng)建一個(gè)隊(duì)列和一個(gè)線程，那將會多么乏味啊！懶惰的程序員喜歡重用各種代碼。所以，我自己覺得很有必要將這一切封裝起來。我們只需要封裝這個(gè)隊(duì)列和創(chuàng)建線程的繁瑣細(xì)節(jié)，讓應(yīng)用層全部專注于具體的邏輯處理：
我利用了已有的組件：線程安全的容器multi_list、包裝任意執(zhí)行體的functor、線程維護(hù)類thread。那么，現(xiàn)在，應(yīng)用層只需要定義隊(duì)列節(jié)點(diǎn)類型，寫應(yīng)用相關(guān)的回調(diào)函數(shù)(任意可被functor包裝的廣義函數(shù))。(見附件例子)
之所以為這個(gè)組件加上init和release，是因?yàn)橛行〇|西(例如COM)需要在線程啟動時(shí)初始化，而在線程快結(jié)束時(shí)釋放，例如對于使用COM的應(yīng)用來說，就需要在線程初始化時(shí)CoInitialize，結(jié)束時(shí)CoUninitialize。
閑說下其他東西
在本文的附件代碼里，你可以獲取到functor、thread、multi_list這些東西，所以我有必要提一下。
關(guān)于functor，你可以參看<實(shí)現(xiàn)functor - 增強(qiáng)型的函數(shù)指針>，基本上可以看成增強(qiáng)版的C回調(diào)函數(shù)；至于multi_list，基本上是一個(gè)container adapter (套用下STL的概念)，使用條件變量參與線程同步，據(jù)說效率要比簡單的互斥高點(diǎn)；至于thread，我需要特別說下：
thread最為重要的就是為其附加了一個(gè)windows的消息隊(duì)列(只要調(diào)用PeekMessage之類的函數(shù)該隊(duì)列就存在)，本意是可以讓其他線程傳送數(shù)據(jù)到該線程，但是目前只用于線程退出，即其他線程可以在任何時(shí)候要求該線程安全地退出(該線程沒有阻塞的情況下，阻塞時(shí)獲取不到消息)。我不知道這個(gè)安全退出策略是否真的有必要存在，但是我討厭看到各種撇腳的退出方法(例如設(shè)置全局標(biāo)志變量，增加額外的--沒封裝前---event對象之類)。
結(jié)束
不知道其他人是如何做這種異步轉(zhuǎn)換操作的，在這里我只是起個(gè)拋磚引玉的作用，歡迎大家提出意見。
posted on 2008-06-25 15:47 Kevin Lynx 閱讀(5062) 評論(29) 編輯 收藏 引用 所屬分類: 通用編程
既然允許阻塞時(shí)不能退出，為什么不能設(shè)全局變量呢？他更及時(shí)，不用參與排隊(duì)，當(dāng)前日志寫完后，就可以退出了，當(dāng)然你要考慮日志是否要保存。 回復(fù) 更多評論
@true 貌似你誤解我意思了;) ’阻塞時(shí)不能退出‘： 例如線程里 WaitFor...INFINITE了，那么它就無法處理消息，也無法判斷退出標(biāo)志變量，代碼卡在這里了。 ’他更及時(shí)，不用參與排隊(duì)‘： 這個(gè)跟上面不是說的同一個(gè)問題，隊(duì)列是用于實(shí)際業(yè)務(wù)處理，例如寫日志，上面那個(gè)說的是線程。 回復(fù) 更多評論
一般這么搞，日志線程退出的及時(shí)：WaitFor。。。設(shè)個(gè)timeout，返回后，一是檢查是否超時(shí)，二檢查全局變量。除非在代碼結(jié)束階段，否則還是少用infinite。這種用法，用個(gè)模式的術(shù)語就是Active Object（感覺這個(gè)概念提出的很強(qiáng)，思路清晰） 回復(fù) 更多評論
撇腳的退出方法，其實(shí)沒什么，主要是個(gè)人喜惡。 我喜歡用標(biāo)記，可以在任何平臺下使用都沒問題。 另外，我不明白為什么用了condition還要用GUARD 那樣不是死鎖了么？ BLOCK等待condition的時(shí)候，container被GUARD鎖死在一個(gè)線程內(nèi)，其他線程調(diào)用PUSH的時(shí)候都會被擋在GUARD那里。 回復(fù) 更多評論
@飯中淹 不會的，condition和guard使用的都是同一個(gè)mutex。guard用于一般的同步，condition用于隊(duì)列元素為0時(shí)的臨界條件。 回復(fù) 更多評論
“如果數(shù)據(jù)庫配置在網(wǎng)絡(luò)上，那么往數(shù)據(jù)庫中插入一些日志信息將非常慢”，沒錯(cuò)，有時(shí)還會發(fā)生網(wǎng)絡(luò)錯(cuò)。 回復(fù) 更多評論
@Kevin Lynx 是啊，當(dāng)元素為0，又正好BLOCK=TRUE的時(shí)候，就會在POP里面死循環(huán)了。 因?yàn)镻OP里的GUARD鎖住了MUTEX，PUSH的GUARD就會等在GUARD的MUTEX那里。這樣的話，POP就永遠(yuǎn)等不到CONDITION的SIGNAL。 回復(fù) 更多評論
正準(zhǔn)備寫一篇類似的東西。被你搶先了……還有沒有必要寫呢？ 回復(fù) 更多評論
@飯中淹 前面我說了，condition和guard使用的是同一個(gè)mutex，在condition的wait里，會先_external_mutex.release();，然后push的時(shí)候，就不會阻塞在guard那里，于是condition.signal(元素為0時(shí)，看下代碼)，然后pop里的condition.wait就過了。 回復(fù) 更多評論
最近準(zhǔn)備把我的軟件渲染器搞成異步的，思路也差不多和LZ一樣。 回復(fù) 更多評論
@Kevin Lynx 這樣會造成很多誤解啊。。。。 WAIT完成，再去LOCK一下MUTEX么？ 回復(fù) 更多評論
@true 同意這種做法 回復(fù) 更多評論
我倒是有個(gè)想法,可以考慮使用一個(gè)"特殊的日志消息",根據(jù)消息內(nèi)容得知是退出消息 回復(fù) 更多評論
例如線程里 WaitFor...INFINITE了，那么它就無法處理消息，也無法判斷退出標(biāo)志變量，代碼卡在這里了。 我是這樣判斷，設(shè)置個(gè)計(jì)數(shù)器，可以是全局/局部，在線程執(zhí)行體里不斷自加，主進(jìn)程隔一段時(shí)間去比較這個(gè)值是否增大。如果沒有，則判斷線程已經(jīng)死掉，這樣就可以避免許多其他的異常而沒有捕捉到的情況。 回復(fù) 更多評論
原文如下：void push_back( const value_type &data ) { guard<mutex_type> g( _mutex ); if( _container.size() == 0 ) { _condition.signal(); } _container.push_back( data ); }不知道此處這樣寫何解？我覺得是不是應(yīng)該寫成這樣：void push_back( const value_type &data ) { guard<mutex_type> g( _mutex ); _container.push_back( data ); _condition.signal(); }？ 回復(fù) 更多評論
@francis 你可以查找‘條件變量’(condition variable)的相關(guān)資料。 signal只在隊(duì)列尺寸為0時(shí)的push操作中調(diào)用，因?yàn)樵趐op中會在隊(duì)列尺寸為0時(shí)wait. 回復(fù) 更多評論
@Kevin Lynx “signal只在隊(duì)列尺寸為0時(shí)的push操作中調(diào)用”沒必要這樣吧？只要push之后signal，信號量加1, pop時(shí)wait信號量減1,為0時(shí)wait就可以了 回復(fù) 更多評論
@francis 關(guān)鍵在于，signal和wait會對外部mutex做操作。參見kl_condition.h相關(guān)代碼。 回復(fù) 更多評論
@Kevin Lynx signal并沒有對mutex操作， void push_back( const value_type &data ) { guard<mutex_type> g( _mutex ); _container.push_back( data ); _condition.signal(); } 這樣寫是不是好些？ 回復(fù) 更多評論
if( _container.size() == 0 ) { _condition.signal(); } 這么寫沒有必要 回復(fù) 更多評論
@francis 我們看一次常規(guī)的同步操作： push_front() { guard<mutex_type> g(_mutex ); ... } pop_front() { guadr<mutex_type> g(_mutex ); ... } 那么，在消費(fèi)者線程里可能會不斷地去檢查隊(duì)列大小是否為0(size操作同樣會涉及到同步)， 這浪費(fèi)了CPU資源。而如果： pop_front() { guadr<mutex_type> g(_mutex ); while( size == 0 ) _cond.wait(); } 當(dāng)大小為0時(shí)，wait操作將阻塞此線程，從而讓出了CPU資源(wait會阻塞)。 另一方面，如果每次push操作都進(jìn)行條件變量的signal，這個(gè)所謂的原語操作開銷有多大？ 查看condition::signal/wait代碼，其內(nèi)部還存在一個(gè)_wait_count的同步。另外，如果 每次push都signal的話，那么pop操作也需要進(jìn)行wait，這樣以來它其實(shí)已經(jīng)不是條件變量， 而是信號量了。 我們所要的效果，就是在隊(duì)列元素為0時(shí)，進(jìn)行pop操作的時(shí)候讓其阻塞而已。 回復(fù) 更多評論
@Kevin Lynx 注意看，想想當(dāng)有線程在wait，同時(shí)container不為空時(shí)的情況吧，明明container里有數(shù)據(jù)，但是wait的線程卻取不出來。。。 這樣看來貌似是存在邏輯錯(cuò)誤的。 回復(fù) 更多評論
@francis while( _container.size() == 0 ) _condition.wait(); 僅當(dāng)為空的情況下才wait的。代碼在邏輯上不存在大問題，因?yàn)檫@個(gè)基礎(chǔ)部件已經(jīng)被用于實(shí)際項(xiàng)目（用于保存驗(yàn)證服務(wù)器端的驗(yàn)證帳號）。 回復(fù) 更多評論
@Kevin Lynx while( _container.size() == 0 ) _condition.wait(); 并不能保證“有線程在wait，同時(shí)container不為空”， 是可以模擬出現(xiàn)的。這樣的話，邏輯就出問題了。 回復(fù) 更多評論
@francis 詳述下這種會出現(xiàn)問題的情況。如果可以模擬出現(xiàn)的話，麻煩基于multi_list這個(gè)類制造這種情況。 回復(fù) 更多評論
:-) 不是太好找，應(yīng)該是沒什么問題的。下面這個(gè)情況有太多假設(shè)，:-)有2線程在等待，thread1.pop ->waitthread2.pop ->wait然后下面運(yùn)行thread3.pushthread3.pushthread3.push。。。thread3 push之后會激活一個(gè)線程,假設(shè)thread1被激活，假設(shè)這個(gè)時(shí)候線程沒有切換，thread3繼續(xù)push之后才切換thread1, 這個(gè)時(shí)候， container被push了兩次，thread1這時(shí)取出一個(gè)數(shù)據(jù)處理，處理完之后切換線程，thread2還是繼續(xù)等待，thread3繼續(xù)push,這個(gè)時(shí)候假設(shè)又push了2個(gè),thread1由于沒有wait可以不停的取數(shù)據(jù)，但是thread2就慘了，不是沒有數(shù)據(jù)可處理，而是它只能wait了。 回復(fù) 更多評論
@francis 我讀了幾遍你的這次回復(fù)，希望我沒有誤解你的意思： thread1被激活，取數(shù)據(jù)，就可能導(dǎo)致container的size為0，一旦size為0，thread3繼續(xù)push的時(shí)候就可能會激活thread2。如果OS一直沒有調(diào)度到thread2，那么，container就可能經(jīng)歷過多次size為0，size不為0，也就是thread1和thread3發(fā)生多次數(shù)據(jù)交互。thread2看起來是慘了，因?yàn)閏ontainer曾經(jīng)有數(shù)據(jù)，但是thread2卻沒取到？為什么呢？thread2根本沒被OS調(diào)度到，從沒有獲取到CPU控制權(quán)，它又有什么理由不wait？ 歸根結(jié)底，你說的這個(gè)問題，只是因?yàn)閠hread2沒有被調(diào)度到而已。 回復(fù) 更多評論
我說的就是會出現(xiàn)“container不空，但是有wait的線程”這種情況,這里就是thread2在wait.如果有更多的線程做push, thread3本身也在不停的push,這個(gè)時(shí)候，處理的線程只有thread1,thread2只能wait, 因?yàn)閏ontainer一直不空，永遠(yuǎn)沒法發(fā)出激活信息。當(dāng)然這樣不會出現(xiàn)什么錯(cuò)誤，只是可能會和實(shí)際想的功能不符合，本意是想兩個(gè)線程處理container里的內(nèi)容，但是實(shí)際只有thread1在處理，thread2一直wait. 回復(fù) 更多評論
@francis thread2沒有獲得CPU權(quán)！thread2沒有被調(diào)度！ 這里的用法完全是條件變量的標(biāo)準(zhǔn)用法，如果你硬是要從理論上認(rèn)為它有問題，那你可以去查條件變量相關(guān)資料。這里的結(jié)構(gòu)和條件變量通用用法一樣。你甚至可以找本操作系統(tǒng)書，在上面找到條件變量的使用結(jié)構(gòu)。你想推翻這一切？建議你大量查閱條件變量相關(guān)資料。如果你還認(rèn)為有問題，你應(yīng)該去對發(fā)明條件變量的某個(gè)可能已經(jīng)死掉的牛人說：你這個(gè)條件變量的結(jié)構(gòu)有問題。 回復(fù) 更多評論
Powered by: C++博客 Copyright © Kevin Lynx