模板類高效執行緒安全的實現Queue
保證同一時刻多個執行緒不會同時修改同一個共享資源，那麼這個程式是執行緒安全的，或者是序列化訪問資源的。可以使用mutex類來控制執行緒的併發問題。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
#include <iostream>
#include <boost/thread/thread.hpp> #include <string> // A simple queue class; don't do this, use std::queue template<typename T> class Queue { public: Queue( ) {} ~Queue( ) {} void enqueue(const T& x) { // Lock the mutex for this queue boost::mutex::scoped_lock lock(mutex_); list_.push_back(x); // A scoped_lock is automatically destroyed (and thus unlocked) // when it goes out of scope } T dequeue( ) { boost::mutex::scoped_lock lock(mutex_); if (list_.empty( )) throw "empty!"; // This leaves the current scope, so the T tmp = list_.front( ); // lock is released list_.pop_front( ); return(tmp); } // Again: when scope ends, mutex_ is unlocked private: std::list<T> list_; boost::mutex mutex_; }; Queue<std::string> queueOfStrings; void sendSomething( ) { std::string s; for (int i = 0; i < 10; ++i) { queueOfStrings.enqueue("Cyrus"); } } void recvSomething( ) { std::string s; for (int i = 0; i < 10; ++i) { try {s = queueOfStrings.dequeue( );} catch(...) {} } } int main( ) { boost::thread thr1(sendSomething); boost::thread thr2(recvSomething); thr1.join( ); thr2.join( ); } |
mutex物件本身並不知道它代表什麼，它僅僅是被多個消費者執行緒使用的資源訪問的鎖定解鎖標誌。在某個時刻，只有一個執行緒可以鎖定這個mutex物件，這就阻止了同一時刻有多個執行緒併發訪問共享資源。一個mutex就是一個簡單的訊號機制。
給mutex加解鎖有多種策略，最簡單的是使用scoped_lock類，它使用一個mutex引數來構造，並一直鎖定這個mutex直到物件被銷燬。如果這個正在被構造的mutex已經被別的執行緒鎖定的話，當前執行緒就會進入wait狀態，直到這個鎖被解開。
利用read_write_mutex對上述的進行改進：
mutex有一個美中不足，它不區分讀和寫。執行緒如果只是進行讀操作，mutex強制執行緒序列化訪問資源，效率低。而且這種操作不需要排他性訪問。基於這個原因，Boost執行緒庫提供了read_write_mutex。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
#include <iostream> #include <boost/thread/thread.hpp> #include <boost/thread/read_write_mutex.hpp> #include <string> template<typename T> class Queue { public: Queue( ) : // Use a read/write mutex and give writers priority rwMutex_(boost::read_write_scheduling_policy::writer_priority){} ~Queue( ) {} void enqueue(const T& x) { // Use a r/w lock since enqueue updates the state boost::read_write_mutex::scoped_write_lock writeLock(rwMutex_); list_.push_back(x); } T dequeue( ) { // Again, use a write lock boost::read_write_mutex::scoped_write_lock writeLock(rwMutex_); if (list_.empty( )) throw "empty!"; T tmp = list_.front( ); list_.pop_front( ); return(tmp); } T getFront( ) { // This is a read-only operation, so you only need a read lock boost::read_write_mutex::scoped_read_lock readLock(rwMutex_); if (list_.empty( )) throw "empty!"; return(list_.front( )); } private: std::list<T> list_; boost::read_write_mutex rwMutex_; }; Queue<std::string> queueOfStrings; void sendSomething( ) { std::string s; for (int i = 0; i < 10; ++i) { queueOfStrings.enqueue("Cyrus"); } } void checkTheFront( ) { std::string s; for (int i = 0; i < 10; ++i) { try {s = queueOfStrings.getFront( );} catch(...) {} } } int main( ) { boost::thread thr1(sendSomething); boost::thread_group grp; grp.create_thread(checkTheFront); grp.create_thread(checkTheFront); grp.create_thread(checkTheFront); grp.create_thread(checkTheFront); thr1.join( ); grp.join_all( ); } |
注意Queue的建構函式中隊讀寫鎖rwMutex的初始化。同一時刻，可能有多個讀寫執行緒要鎖定一個read_write_mutex，而這些鎖的排程策略依賴於構造這個mutex時選定的排程策略。Boost庫中提供了四種排程策略：
1）reader_priority：等待讀鎖的執行緒優先於等待寫鎖的執行緒
2）writer_priority：等待寫鎖的執行緒優先於等待讀鎖的執行緒
3）alternating_single_read：在讀鎖和寫鎖之間交替
4）alternating_many_reads：在讀鎖和寫鎖之間交替，這個策略將在兩個寫鎖之間使得所有的在這個queue上掛起的讀鎖都被允許。
選擇使用哪種策略要慎重，因為使用前兩種的話可能會導致某些鎖始終不能成功，出現餓死的現象。