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