最后，向大家介紹muduo庫中對于線程池的封裝的理解。
最重要的想法就是線程池將線程看為自己可執行的最小并且可隨時增加的單位。
整個線程池對象維持兩個任務隊列，threads_表示目前正在運行中的線程池，queue_表示位于存儲隊列中的等待線程。
thread_在運行的過程中使用while循環+條件變量判斷當前的活動線程池中是否有空位，以及新的等待線程進入線程池。
線程池從一開始就確定了自己將要運行的線程數目，不能在后面的運行中更改。
/***ThreadPool.h***/class ThreadPool : boost::noncopyable{ public: typedef boost::function<void ()> Task;//將線程池中的線程作為可替換的任務，以線程為基本單位放在線程池中運行 explicit ThreadPool(const string& nameArg = string("ThreadPool")); ~ThreadPool(); // Must be called before start(). // 設置線程池運行的最大的負載以及線程池中將要運行的線程 void setMaxQueueSize(int maxSize) { maxQueueSize_ = maxSize; }// void setThreadInitCallback(const Task& cb) { threadInitCallback_ = cb; } void start(int numThreads);//啟動一定數量的線程 void stop();//線程池運算停止 const string& name() const { return name_; } size_t queueSize() const;//返回正在排隊等待的線程任務 // Could block if maxQueueSize > 0 void run(const Task& f);//將一個想要運行的線程放入線程池的任務隊列#ifdef __GXX_EXPERIMENTAL_CXX0X__ void run(Task&& f);//C++11的移動方法，用于節省資源#endif PRivate: bool isFull() const;//判斷線程隊列是否已經滿了 void runInThread();//真正讓線程跑起來的函數 Task take();//獲得任務隊列的首個線程 mutable MutexLock mutex_;//互斥鎖 Condition notEmpty_;//條件變量 Condition notFull_; string name_; Task threadInitCallback_;//線程池中執行的線程對象 boost::ptr_vector<muduo::Thread> threads_;//線程池 std::deque<Task> queue_;//排隊執行的線程對象隊列 size_t maxQueueSize_;//等待隊列的最大數 bool running_;//是否已經啟動};每一個加入線程池的線程都帶有一個while循環，保證線程等待隊列中的線程不會等待太久。即所有將加入線程池的線程都會進入線程等待隊列接受檢查。
start()：線程池啟動函數保證在調用時啟動一定數量的線程。
stop()：保證所有的正在運行的線程停止
queueSize()：返回此時線程等待隊列中的個數，用于判斷線程等待隊列是否為空
run()：如果線程池為空，直接跑傳入的線程。如果線程池等待隊列滿了，則當前控制流（線程）在notFull_上等待；否則將傳入的線程加入線程等待隊列，并且使用條件變量notEmpty_通知一條阻塞在該條件變量上的控制流（線程）。
take()：如果當前線程等待隊列為空并且線程池正在跑，則控制流（線程）阻塞在notEmpty_條件變量上。當條件變量被激活時（有線程對象加入呆線程等待隊列），判斷是否可以從線程等待隊列中拿出一個線程對象，如果可以，則將使用條件變量notFull_通知run()中阻塞在–想加入隊列但是隊列沒有空余位置的變量上。
isFull()：返回在線程等待隊列中的個數，用于判斷是否可以將想要運行的線程放到線程等待隊列中。
runInThread()：如果線程啟動函數不為空，則在此將線程的控制流交給用于初始化線程池的線程對象。當此線程對象運行結束的時候，并且此時的線程池還在運行，則線程池離開初始化模式，進入線程池的循環線程補充模式。這種模式控制著線程池中的線程數量：當有新的線程對象進入線程池，則當前的線程控制流交給將要執行的線程對象。也就是說線程池中的線程對象要么主動結束自己的‘life’，然后由線程池的線程補充模式決定將要進入線程池運行的線程對象。然后在后面的take()中使用條件變量完成新的線程進入線程池的同步。
/***ThreadPool.cc***/ThreadPool::ThreadPool(const string& nameArg) : mutex_(), notEmpty_(mutex_), notFull_(mutex_), name_(nameArg), maxQueueSize_(0), running_(false){}ThreadPool::~ThreadPool(){ if (running_) { stop(); }}void ThreadPool::start(int numThreads){ assert(threads_.empty());//首次啟動，斷言線程池為空 running_ = true; threads_.reserve(numThreads);//預分配空間 for (int i = 0; i < numThreads; ++i) { char id[32]; snprintf(id, sizeof id, "%d", i+1); threads_.push_back(new muduo::Thread( boost::bind(&ThreadPool::runInThread, this), name_+id)); threads_[i].start();//直接啟動線程 } if (numThreads == 0 && threadInitCallback_)//只啟動一條線程 { threadInitCallback_(); }}void ThreadPool::stop(){ { MutexLockGuard lock(mutex_); running_ = false; notEmpty_.notifyAll();//目的在于通知notempty條件變量 } for_each(threads_.begin(), threads_.end(), boost::bind(&muduo::Thread::join, _1));//使用STL算法依次關閉線程池中運行的線程}size_t ThreadPool::queueSize() const{//獲得排隊任務執行隊列的隊列長度 MutexLockGuard lock(mutex_); return queue_.size();}void ThreadPool::run(const Task& task){ if (threads_.empty())//如果線程池為空，直接跑這條線程 { task(); } else { MutexLockGuard lock(mutex_); while (isFull())//如果線程等待隊列滿了，在notfull條件變量上等待 { notFull_.wait(); } assert(!isFull()); queue_.push_back(task);//現在線程池執行任務隊列中有空位了 notEmpty_.notify();//notempty條件變量通知信息 }}#ifdef __GXX_EXPERIMENTAL_CXX0X__void ThreadPool::run(Task&& task){ if (threads_.empty()) { task(); } else { MutexLockGuard lock(mutex_); while (isFull()) { notFull_.wait(); } assert(!isFull()); queue_.push_back(std::move(task)); notEmpty_.notify(); }}#endifThreadPool::Task ThreadPool::take(){ MutexLockGuard lock(mutex_); // always use a while-loop, due to spurious wakeup while (queue_.empty() && running_)//如果線程隊列為空并且線程池正在跑 {//在notempty條件變量上等待 notEmpty_.wait();//當前線程停下來等待，當隊列不為空了繼續跑 }//然后獲得新任務 Task task;//創建一個新的任務 if (!queue_.empty()) { task = queue_.front();//獲得隊列中的頭任務 queue_.pop_front();//彈出隊列中的頭任務 if (maxQueueSize_ > 0)//如果隊列最大長度大于0 { notFull_.notify();//通知線程可以跑了 } } return task;//返回任務}bool ThreadPool::isFull() const{//用來判斷線程隊列是否已經滿了 mutex_.assertLocked(); return maxQueueSize_ > 0 && queue_.size() >= maxQueueSize_;}void ThreadPool::runInThread()//生成一個threadFunc對象{ try { if (threadInitCallback_)//如果線程啟動函數不為空，直接啟動 { threadInitCallback_(); } while (running_)//while循環保證線程任務等待隊列中沒有多余的任務 { Task task(take()); if (task) { task(); } } } catch (const Exception& ex) //異常捕捉過程 { fprintf(stderr, "exception caught in ThreadPool %s/n", name_.c_str()); fprintf(stderr, "reason: %s/n", ex.what()); fprintf(stderr, "stack trace: %s/n", ex.stackTrace()); abort(); } catch (const std::exception& ex) { fprintf(stderr, "exception caught in ThreadPool %s/n", name_.c_str()); fprintf(stderr, "reason: %s/n", ex.what()); abort(); } catch (...) { fprintf(stderr, "unknown exception caught in ThreadPool %s/n", name_.c_str()); throw; // rethrow }}碩爺的條件變量用的很精髓，非常精髓。
新聞熱點
疑難解答
