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