Boost線程詳解

一、創建一個線程

創建線程

    boost::thread myThread(threadFun);

    需要注意的是：參數可以是函數對象或者函數指針。並且這個函數無參數，並返回void類型。

    當一個thread執行完成時，這個子線程就會消失。注意這個線程對象不會消失，它仍然是一個還處在它的生存期的C++對象。同理，當對一個堆上的線程對象的指針調用delete時候，線程對象被銷毀，操作系統的線程並不能保證就消失。

放棄時間片

    boost::thread::yield();

    當前線程放棄余下的時間片。

等待一個線程

    myThread.join();

    調用這個方法的線程進入wait狀態，直到myThread代表的線程完成為止。如果它不結束的話，join方法就不會返回。join是一個等待子線程結束的最好的方法。如果主程序不調用join方法而直接結束，它的子線程有可能沒有執行完成，但是所有的子線程也隨之退出。不調用join方法，主線程就不會等待它的子線程。

#include <iostream>
#include <boost/thread/thread.hpp>
#include <boost/thread/xtime.hpp>
 
struct MyThreadFunc {
   void operator( )( ) {
      // Do something long-running...
   }
} threadFun;
 
int main( ) {
 
   boost::thread myThread(threadFun); // Create a thread that starts
                                      // running threadFun
 
   boost::thread::yield( );           // Give up the main thread's timeslice
                                      // so the child thread can get some work
                                      // done.
 
   // Go do some other work...
 
   myThread.join( );                  // The current (i.e., main) thread will wait
                                      // for myThread to finish before it returns
 
}

線程組

    如果你需要創建幾個線程，考慮使用一個線程組對象thread_group來組織它們。一個thread_group對象可以使用多種方法管理線程。首先，可以使用一個指向動態創建的線程對象的指針作為參數來調用add_thread方法，將這個線程加入線程組。也可以直接使用線程組類的create_thread方法，可不先創建線程而直接把線程加入到線程組中。

    當線程組對象的析構函數被調用時，它將刪除（delete）所有這些通過add_thread方法加入的線程指針。所以，只能將堆上的線程對象指針通過add_thread方法加入線程組。remove_thread方法從線程組刪除某個線程的指針，但是我們仍需負責把線程本身內存釋放掉。

    線程組對象的成員方法join_all方法等待線程組中所有線程結束，才返回。

boost::thread_group grp;
boost::thread *p = new boost::thread(threadFun);
grp.add_thread(p);
//do something...
grp.remove_thread(p);
 
grp.create_thread(threadFun);
grp.create_thread(threadFun);   //Now there are two threads in grp
 
grp.join_all();                 //Wait for all threads to finish

二、使資源是線程安全的

    保證同一時刻多個線程不會同時修改同一個共享資源，那么這個程序是線程安全的，或者是串行化訪問資源的。可以使用mutex類來控制線程的並發問題。

#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狀態，直到這個鎖被解開。

三、讀寫鎖

    mutex有一個美中不足，它不區分讀和寫。線程如果只是進行讀操作，mutex強制線程串行化訪問資源，效率低。而且這種操作不需要排他性訪問。基於這個原因，Boost線程庫提供了read_write_mutex。

#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上掛起的讀鎖都被允許。

     選擇使用哪種策略要慎重，因為使用前兩種的話可能會導致某些鎖始終不能成功，出現餓死的現象。

死鎖、餓死和競態條件

    1）死鎖，是涉及至少2個線程和2個資源的情況。線程A和B，資源X和Y。A鎖定了X，而B鎖定了Y。此時A和B有彼此想要對方的資源，死鎖就出現了。

　　死鎖的預防有兩種方法。一種是，通過小心的按照一定的順序對不同的mutex來加鎖。另一種是，使用Boost提供的try_mutex互斥量和scoped_try_lock。或者使用時間鎖。scoped_try_lock對try_mutex加鎖時，可能成功，也可能失敗，但不會阻塞。時間鎖則有一個超時時間。

bool dequeue(T& x)
{
    boost::try_mutex::scope_try_lock lock(tryMutex_);
    if(!lock.locked())
        return false;
    else{
        if (list_.empty())
            throw "empty!";
        x = list_.front();
        list_.pop_front();
        return true;
    }
}
private:
    boost::try_mutex tryMutex_;

　 2）餓死，如果你正在使用write_priority策略，並且你有很多創建寫鎖的線程，那么讀鎖的線程就可能餓死。

    3）競態條件，

if(q.getFront() == "Cyrus"){
   str = q.dequeue();
   //....
}

       這個代碼在單線程環境中工作很好，因為q在第一行和第二行代碼之間不會被修改。多線程環境中則會出現問題。此為競態條件。解決的方法是為Queue添加一個成員函數dequeueIfEquals，在函數執行過程中始終鎖定互斥量。

四、從一個線程中給另一個線程發送通知

    當需要線程等待某個事物時，可以創建一個condition對象，然后通過這個對象來通知那些等待的線程。

#include <iostream>
#include <boost/thread/thread.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/mutex.hpp>
#include <list>
#include <string>
 
class Request { /*...*/ };
 
// A simple job queue class; don't do this, use std::queue
template<typename T>
class JobQueue {
public:
   JobQueue( ) {}
  ~JobQueue( ) {}
 
   void submitJob(const T& x) {
      boost::mutex::scoped_lock lock(mutex_);
      list_.push_back(x);
      workToBeDone_.notify_one( );
   }
 
   T getJob( ) {
      boost::mutex::scoped_lock lock(mutex_);
 
      workToBeDone_.wait(lock); // Wait until this condition is
                                // satisfied, then lock the mutex
      T tmp = list_.front( );
      list_.pop_front( );
      return(tmp);
   }
 
private:
   std::list<T> list_;
   boost::mutex mutex_;
   boost::condition workToBeDone_;
};
 
JobQueue<Request> myJobQueue;
 
void boss( ) {
   for (;;) {
      // Get the request from somewhere
      Request req;
      myJobQueue.submitJob(req);
   }
}
 
void worker( ) {
   for (;;) {
      Request r(myJobQueue.getJob( ));
      // Do something with the job...
   }
}
 
int main( ) {
   boost::thread thr1(boss);
   boost::thread thr2(worker);
   boost::thread thr3(worker);
 
   thr1.join( );
   thr2.join( );
   thr3.join( );
}

boost::mutex::scoped_lock lock(mutex_);

workToBeDone_.wait(lock);

     這兩行代碼，第一行鎖定這個mutex對象。第二行代碼解開這個mutex上的鎖，然后進行等待或者休眠，直到它的條件得到了滿足。這個mutex互斥對象的解鎖讓其他的線程能夠使用這個mutex對象，它們中的某個需要設置這個等待條件，之后通知另外的線程。

    notify_all函數，通知那些所有正在等待某個條件變為真的線程，那些線程隨后進入運行狀態。wait方法做兩件事情：它一直等待直到有人在它正等待的condition上調用notify_one或notify_all，然后它就試圖鎖定相關的mutex。當調用的是notify_all時，盡管多個等待的線程都盡量去獲得下一個鎖，但誰將獲得依賴於這個mutex的類型和使用的優先策略。

    一個condition對象能讓消費者線程休眠，因此在還沒有碰到一個condition時處理器可以去處理別的事情。例如一個web服務器使用一個工作線程池來處理進來的請求。當沒有需求進來時，讓這些子線程處於等待狀態比讓它們循環的查詢或者睡眠然后偶爾喚醒來檢查這個隊列，要好很多。

五、只初始化一次共享資源

#include <iostream>
#include <boost/thread/thread.hpp>
#include <boost/thread/once.hpp>
 
// Some sort of connection class that should only be initialized once
struct Conn {
   static void init( ) {++i_;}
   static boost::once_flag init_;
   static int i_;
   // ...
};
 
int Conn::i_ = 0;
boost::once_flag Conn::init_ = BOOST_ONCE_INIT;
 
void worker( ) {
   boost::call_once(Conn::init, Conn::init_);
   // Do the real work...
}
 
Conn c;  // You probably don't want to use a global, so see the
         // next Recipe
 
int main( ) {
 
   boost::thread_group grp;
 
   for (int i = 0; i < 100; ++i)
      grp.create_thread(worker);
 
   grp.join_all( );
 
   std::cout << c.i_ << '\n'; // c.i_ = 1
}

    一個共享資源不得不在某個地方被初始化，並且你希望第一次使用這個資源的線程來完成初始化工作。一個once_flag類型和call_once函數能夠保證多個線程不會重復的初始化同一個對象。首先，必須使用BOOST_ONCE_INIT宏來初始化這個once_flag對象。boost::once_flag Conn::init_ = BOOST_ONCE_INIT; 之后調用call_once函數，boost::call_once(Conn::init, Conn::init_); 第一個形參是希望被執行一次的初始化函數的地址。

六、給線程函數傳遞一個參數

#include <iostream>
#include <string>
#include <functional>
#include <boost/thread/thread.hpp>
 
// A typedef to make the declarations below easier to read
typedef void (*WorkerFunPtr)(const std::string&);
 
template<typename FunT,   // The type of the function being called
         typename ParamT> // The type of its parameter
struct Adapter {
   Adapter(FunT f, ParamT& p) : // Construct this adapter and set the
      f_(f), p_(&p) {}          // members to the function and its arg
 
   void operator( )( ) { // This just calls the function with its arg
      f_(*p_);        
   }
private:
   FunT    f_;
   ParamT* p_;  // Use the parameter's address to avoid extra copying
};
 
void worker(const std::string& s) {
   std::cout << s << '\n';
}
 
int main( ) {
 
   std::string s1 = "This is the first thread!";
   std::string s2 = "This is the second thread!";
 
   boost::thread thr1(Adapter<WorkerFunPtr, std::string>(worker, s1));
   boost::thread thr2(Adapter<WorkerFunPtr, std::string>(worker, s2));
 
   thr1.join( );
   thr2.join( );
}

    使用這個函數適配器類模板，你就可以給線程函數傳遞參數了。如果你需要傳遞多個參數，僅需要在這個適配器中增加另一個類型和成員變量。

 

原文鏈接： http://www.cnblogs.com/younes/archive/2010/06/06/1752745.html