實(shí)現(xiàn)消息隊(duì)列的關(guān)鍵因素是考量不同線程訪問消息隊(duì)列的同步問題吏垮。本實(shí)現(xiàn)涉及到幾個(gè)知識(shí)點(diǎn)
std::lock_guard 介紹
std::lock_gurad 是 C++11 中定義的模板類。定義如下：
template <class Mutex> class lock_guard;
lock_guard 對(duì)象通常用于管理某個(gè)鎖(Lock)對(duì)象罐旗，因此與 Mutex RAII 相關(guān)膳汪，方便線程對(duì)互斥量上鎖，即在某個(gè) lock_guard 對(duì)象的聲明周期內(nèi)九秀，它所管理的鎖對(duì)象會(huì)一直保持上鎖狀態(tài)遗嗽；而 lock_guard 的生命周期結(jié)束之后，它所管理的鎖對(duì)象會(huì)被解鎖(注：類似 shared_ptr 等智能指針管理動(dòng)態(tài)分配的內(nèi)存資源 )鼓蜒。
模板參數(shù) Mutex 代表互斥量類型痹换，例如 std::mutex 類型，它應(yīng)該是一個(gè)基本的 BasicLockable 類型都弹，標(biāo)準(zhǔn)庫(kù)中定義幾種基本的 BasicLockable 類型娇豫，分別 std::mutex, std::recursive_mutex, std::timed_mutex，std::recursive_timed_mutex 以及 std::unique_lock
std::unique_lock 介紹
lock_guard 最大的缺點(diǎn)也是簡(jiǎn)單畅厢，沒有給程序員提供足夠的靈活度冯痢，因此，C++11 標(biāo)準(zhǔn)中定義了另外一個(gè)與 Mutex RAII 相關(guān)類 unique_lock，該類與 lock_guard 類相似系羞，也很方便線程對(duì)互斥量上鎖郭计，但它提供了更好的上鎖和解鎖控制。
顧名思義椒振，unique_lock 對(duì)象以獨(dú)占所有權(quán)的方式（ unique owership）管理 mutex 對(duì)象的上鎖和解鎖操作昭伸，所謂獨(dú)占所有權(quán)，就是沒有其他的 unique_lock 對(duì)象同時(shí)擁有某個(gè) mutex 對(duì)象的所有權(quán)澎迎。
新創(chuàng)建的 unique_lock 對(duì)象管理 Mutex 對(duì)象 m庐杨，并嘗試調(diào)用 m.lock() 對(duì) Mutex 對(duì)象進(jìn)行上鎖，如果此時(shí)另外某個(gè) unique_lock 對(duì)象已經(jīng)管理了該 Mutex 對(duì)象 m夹供，則當(dāng)前線程將會(huì)被阻塞灵份。
std::condition介紹
當(dāng) std::condition_variable 對(duì)象的某個(gè) wait 函數(shù)被調(diào)用的時(shí)候，它使用 std::unique_lock(通過 std::mutex) 來鎖住當(dāng)前線程哮洽。當(dāng)前線程會(huì)一直被阻塞填渠，直到另外一個(gè)線程在相同的 std::condition_variable 對(duì)象上調(diào)用了 notification 函數(shù)來喚醒當(dāng)前線程。
std::condition_variable 提供了兩種 wait() 函數(shù)鸟辅。當(dāng)前線程調(diào)用 wait() 后將被阻塞(此時(shí)當(dāng)前線程應(yīng)該獲得了鎖（mutex）氛什，不妨設(shè)獲得鎖 lck)，直到另外某個(gè)線程調(diào)用 notify_* 喚醒了當(dāng)前線程匪凉。
在線程被阻塞時(shí)枪眉，該函數(shù)會(huì)自動(dòng)調(diào)用 lck.unlock() 釋放鎖，使得其他被阻塞在鎖競(jìng)爭(zhēng)上的線程得以繼續(xù)執(zhí)行再层。另外贸铜，一旦當(dāng)前線程獲得通知(notified，通常是另外某個(gè)線程調(diào)用 notify_* 喚醒了當(dāng)前線程)聂受，wait() 函數(shù)也是自動(dòng)調(diào)用 lck.lock()蒿秦，使得 lck 的狀態(tài)和 wait 函數(shù)被調(diào)用時(shí)相同。
在第二種情況下（即設(shè)置了 Predicate）饺饭，只有當(dāng) pred 條件為 false 時(shí)調(diào)用 wait() 才會(huì)阻塞當(dāng)前線程渤早，并且在收到其他線程的通知后只有當(dāng) pred 為 true 時(shí)才會(huì)被解除阻塞职车。因此第二種情況類似以下代碼：
while (!pred()) wait(lck);
std::function介紹
使用std::function可以將普通函數(shù)瘫俊，lambda表達(dá)式和函數(shù)對(duì)象類統(tǒng)一起來。它們并不是相同的類型悴灵，然而通過function模板類扛芽，可以轉(zhuǎn)化為相同類型的對(duì)象（function對(duì)象），從而放入一個(gè)vector或其他容器里积瞒，方便回調(diào)川尖。
代碼實(shí)現(xiàn)：
#pragma once
#ifndef MESSAGE_QUEUE_H
#define MESSAGE_QUEUE_H
#include <queue>
#include <mutex>
#include <condition_variable>
template<class Type>
class CMessageQueue
{
public:
CMessageQueue& operator = (const CMessageQueue&) = delete;
CMessageQueue(const CMessageQueue& mq) = delete;
CMessageQueue() :_queue(), _mutex(), _condition(){}
virtual ~CMessageQueue(){}
void Push(Type msg){
std::lock_guard <std::mutex> lock(_mutex);
_queue.push(msg);
//當(dāng)使用阻塞模式從消息隊(duì)列中獲取消息時(shí)，由condition在新消息到達(dá)時(shí)提醒等待線程
_condition.notify_one();
}
//blocked定義訪問方式是同步阻塞或者非阻塞模式
bool Pop(Type& msg, bool isBlocked = true){
if (isBlocked)
{
std::unique_lock <std::mutex> lock(_mutex);
while (_queue.empty())
{
_condition.wait(lock);
}
//注意這一段必須放在if語句中茫孔，因?yàn)閘ock的生命域僅僅在if大括號(hào)內(nèi)
msg = std::move(_queue.front());
_queue.pop();
return true;
}
else
{
std::lock_guard<std::mutex> lock(_mutex);
if (_queue.empty())
return false;
msg = std::move(_queue.front());
_queue.pop();
return true;
}
}
int32_t Size(){
std::lock_guard<std::mutex> lock(_mutex);
return _queue.size();
}
bool Empty(){
std::lock_guard<std::mutex> lock(_mutex);
return _queue.empty();
}
private:
std::queue<Type> _queue;//存儲(chǔ)消息的隊(duì)列
mutable std::mutex _mutex;//同步鎖
std::condition_variable _condition;//實(shí)現(xiàn)同步式獲取消息
};
#endif//MESSAGE_QUEUE_H
線程池可以直接在構(gòu)造函數(shù)中構(gòu)造線程叮喳，并傳入回調(diào)函數(shù)被芳，也可以寫一個(gè)Run函數(shù)顯示調(diào)用。這里我們選擇了第二種馍悟，對(duì)比：
在handler函數(shù)外部做循環(huán)接受消息畔濒，當(dāng)消息到達(dá)后調(diào)用hanlder處理。這種實(shí)現(xiàn)在上層做封裝锣咒，但是會(huì)在線程中頻繁的切換調(diào)用函數(shù)侵状。這種設(shè)計(jì)無法復(fù)用一些資源，如當(dāng)在handler中做數(shù)據(jù)庫(kù)操作時(shí)毅整，需要頻繁的連接和斷開連接趣兄，可以通過定義兩個(gè)虛函數(shù)Prehandler和AfterHandler來實(shí)現(xiàn)。
５考怠Ｍ丁！構(gòu)造函數(shù)中調(diào)用虛函數(shù)并不會(huì)能真正的調(diào)用子類的實(shí)現(xiàn)Ｏ访铩１┣！
雖然可以對(duì)虛函數(shù)進(jìn)行實(shí)調(diào)用辛臊，但程序員編寫虛函數(shù)的本意應(yīng)該是實(shí)現(xiàn)動(dòng)態(tài)聯(lián)編仙粱。在構(gòu)造函數(shù)中調(diào)用虛函數(shù)，函數(shù)的入口地址是在編譯時(shí)靜態(tài)確定的彻舰，并未實(shí)現(xiàn)虛調(diào)用寫一個(gè)Run函數(shù)伐割，將這一部分實(shí)現(xiàn)放在run函數(shù)中，顯示調(diào)用刃唤。
《Effective C++ 》條款9：永遠(yuǎn)不要在構(gòu)造函數(shù)或析構(gòu)函數(shù)中調(diào)用虛函數(shù)
#ifndef THREAD_POOL_H
#define THREAD_POOL_H
#include <functional>
#include <vector>
#include <thread>
#include "MessageQueue.h"
#define MIN_THREADS 1
template<class Type>
class CThreadPool
{
CThreadPool& operator = (const CThreadPool&) = delete;
CThreadPool(const CThreadPool& other) = delete;
public:
CThreadPool(int32_t threads,
std::function<void(Type& record, CThreadPool<Type>* pSub)> handler);
virtual ~CThreadPool();
void Run();
virtual void PreHandler(){}
virtual void AfterHandler(){}
void Submit(Type record);
private:
bool _shutdown;
int32_t _threads;
std::function<void(Type& record, CThreadPool<Type>* pSub)> _handler;
std::vector<std::thread> _workers;
CMessageQueue<Type> _tasks;
};
template<class Type>
CThreadPool<Type>::CThreadPool(int32_t threads,
std::function<void(Type& record, CThreadPool<Type>* pSub)> handler)
:_shutdown(false),
_threads(threads),
_handler(handler),
_workers(),
_tasks()
{
//第一種實(shí)現(xiàn)方案隔心，注意這里的虛函數(shù)調(diào)用不正確
/*if (_threads < MIN_THREADS)
_threads = MIN_THREADS;
for (int32_t i = 0; i < _threads; i++)
{
_workers.emplace_back(
[this]{
PreHandler();
while (!_shutdown){
Type record;
_tasks.Pop(record, true);
_handler(record, this);
}
AfterHandler();
}
);
}*/
}
//第二種實(shí)現(xiàn)方案
template<class Type>
void CThreadPool<Type>::Run()
{
if (_threads < MIN_THREADS)
_threads = MIN_THREADS;
for (int32_t i = 0; i < _threads; i++)
{
_workers.emplace_back(
[this]{
PreHandler();
while (!_shutdown){
Type record;
_tasks.Pop(record, true);
_handler(record, this);
}
AfterHandler();
}
);
}
}
template<class Type>
CThreadPool<Type>::~CThreadPool()
{
for (std::thread& worker : _workers)
worker.join();
}
template<class Type>
void CThreadPool<Type>::Submit(Type record)
{
_tasks.Push(record);
}
#endif // !THREAD_POOL_H
