C++11 實(shí)現(xiàn)的 100行 線程池
C++線程池一直都是各位程序員們?cè)燧喿拥氖走x項(xiàng)目之一。今天，帶大家一起來看看這個(gè)輕量的線程池，本線程池是header-only的，并且整個(gè)文件只有100行，其中C++的高級(jí)用法有很多，很值得我們學(xué)習(xí)，一起來看看吧。
以下是正文
線程池
C++帶有線程操作，異步操作，就是沒有線程池，至于線程池的概念，我先搜一下別人的解釋：
一般而言,線程池有以下幾個(gè)部分:
1. 完成主要任務(wù)的一個(gè)或多個(gè)線程。
2. 用于調(diào)度管理的管理線程。
3. 要求執(zhí)行的任務(wù)隊(duì)列。
我來講講人話：你的函數(shù)需要在多線程中運(yùn)行，但是你又不能每來一個(gè)函數(shù)就開啟一個(gè)線程，所以你就需要固定的N個(gè)線程來跑執(zhí)行，但是有的線程還沒有執(zhí)行完，有的又在空閑，如何分配任務(wù)呢，你就需要封裝一個(gè)線程池來完成這些操作，有了線程池這層封裝，你就只需要告訴它開啟幾個(gè)線程，然后直接塞任務(wù)就行了，然后通過一定的機(jī)制獲取執(zhí)行結(jié)果。
這里有一個(gè)100行實(shí)現(xiàn)線程池的操作：
https://github.com/progschj/ThreadPool/blob/master/ThreadPool.h
分析源代碼 頭文件：
#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>
vector,queue,momory 都沒啥說的，thread線程相關(guān)，mutex 互斥量，解決資源搶占問題，condition_variable 條件量，用于喚醒線程和阻塞線程,future 從使用的角度出發(fā)，它是一個(gè)獲取線程數(shù)據(jù)的函數(shù)。functional 函數(shù)子，可以理解為規(guī)范化的函數(shù)指針。stdexcept 就跟它的名字一樣，標(biāo)準(zhǔn)異常。
class ThreadPool {
public:
ThreadPool(size_t);
template<class F, class... Args>
auto enqueue(F&& f, Args&&... args)
-> std::future<typename std::result_of<F(Args...)>::type>;
~ThreadPool();
private:
// need to keep track of threads so we can join them
std::vector< std::thread > workers;
// the task queue
std::queue< std::function<void()> > tasks;
// synchronization
std::mutex queue_mutex;
std::condition_variable condition;
bool stop;
};
線程池的聲明，構(gòu)造函數(shù)，一個(gè)enqueue模板函數(shù) 返回std::future<type>, 然后這個(gè)type又利用了運(yùn)行時(shí)檢測(cè)（還是編譯時(shí)檢測(cè)?）推斷出來的，非常的amazing啊。成功的使用一行代碼反復(fù)套娃，這高階的用法就是大佬的水平嗎，i了i了。
workers 是vector<std::thread> 俗稱工作線程。
std::queue<std::function<void()>> tasks 俗稱任務(wù)隊(duì)列。
那么問題來了，這個(gè)任務(wù)隊(duì)列的任務(wù)只能是void() 類型的嗎？感覺沒那么簡(jiǎn)單，還得接著看吶。
mutex,condition_variable 沒啥講的,stop 控制線程池停止的。
// the constructor just launches some amount of workers
inline ThreadPool::ThreadPool(size_t threads)
: stop(false)
{
for(size_t i = 0;i<threads;++i)
workers.emplace_back(
[this]
{
for(;;)
{
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(this->queue_mutex);
this->condition.wait(lock,
[this]{ return this->stop || !this->tasks.empty(); });
if(this->stop && this->tasks.empty())
return;
task = std::move(this->tasks.front());
this->tasks.pop();
}
task();
}
}
);
}
大佬寫的注釋就是這么樸實(shí)無華，說這個(gè)構(gòu)造函數(shù)僅僅是把一定數(shù)量的線程塞進(jìn)去，我是看了又看才悟出來這玩意是什么意思……雖然本質(zhì)上的確是它說的只是把線程塞進(jìn)去，但是這個(gè)線程也太繞了。
workers.emplace_back 參數(shù)是一個(gè)lambda表達(dá)式，不會(huì)阻塞，也就是說最外層的是一個(gè)異步函數(shù)，每個(gè)線程里面的事情才是重點(diǎn)。
labmda表達(dá)式中最外層是一個(gè)死循環(huán)，至于為什么是for(;;)而不是while(1) 這雖然不是重點(diǎn)，不過大佬的用法還是值得揣摩的，我估計(jì)是效率會(huì)更高？
task 申明后，緊跟著一個(gè)大括號(hào)，這個(gè){}里面的部分，是一個(gè)同步操作，至于為什么用this->lock 而不是直接使用[&]來捕獲參數(shù)，想來也是處于內(nèi)存考慮。精打細(xì)算的風(fēng)格像極了摳門的地主，i了i了。
緊接著一個(gè)wait(lock,condtion)的操作，像極了千層餅的套路。
第一層：這TM不是要鎖死自己啊？這樣不是構(gòu)造都得卡死？
第二層：我們看到它emplace_back了一個(gè)線程，不會(huì)阻塞，但是等開鎖，鎖不就在它自己的線程里面嘛？那不得鎖死了啊？
第三層：我們看到這個(gè)lock其實(shí)只是個(gè)包裝，真正的鎖是外層的mutex，所以從這里是不存在死鎖的。但是你的wait的condition怎么可能不懂呢，必須要 stop 或者 !empty 才wait嗎？
第四層：我們查資料發(fā)現(xiàn)后面的condition是返回false才會(huì)wait，也就是說要!stop && empty才會(huì)wait，就是說這個(gè)線程池是 運(yùn)行態(tài)，并且沒有任務(wù)才才會(huì)執(zhí)行等待操作！否則就不等了，直接沖！
第五層：既然你判斷了上面判斷了stop和非空，為啥下面還要判斷stop和空才退出呢？不顯得冗余？
第六層：要確定它的確是被置為stop了，且隊(duì)列執(zhí)行空了，它才能夠光榮退休。有沒有問題呢，有，最后所有線程都阻塞了，你stop置為true它們也不知道啊……
我估計(jì)它的stop會(huì)有喚醒所有線程的操作，不過如果有的在執(zhí)行，有的在等待，應(yīng)該沒辦法都通知到位，但是在執(zhí)行的在下一次判斷的時(shí)候也能正常退出。
因?yàn)橛辛艘苫螅覀兙拖肟磗top相關(guān)的操作，結(jié)果發(fā)現(xiàn)放在了析構(gòu)函數(shù)里面……
// the destructor joins all threads
inline ThreadPool::~ThreadPool()
{
{
std::unique_lock<std::mutex> lock(queue_mutex);
stop = true;
}
condition.notify_all();
for(std::thread &worker: workers)
worker.join();
}{}里面上鎖進(jìn)行了stop為true的操作，至于為什么不用原子操作，我也不知道，但是仔細(xì)想了下大概是因?yàn)楸緛砭陀幸话焰i了，再用原子就不是內(nèi)味兒了。然后它果然通知了所有，并且還把工作線程join了。也就是等它們結(jié)束。
結(jié)束了千層餅の解析之后，我們看看最重要的入隊(duì)操作
// add new work item to the pool
template<class F, class... Args>
auto ThreadPool:
:enqueue(F&& f, Args&&... args)
-> std::future<typename std::result_of<F(Args...)>::type>
{
using return_type = typename std::result_of<F(Args...)>::type;
auto task = std::make_shared< std::packaged_task<return_type()> >(
std::bind(std::forward<F>(f), std::forward<Args>(args)...)
);
std::future<return_type> res = task->get_future();
{
std::unique_lock<std::mutex> lock(queue_mutex);
// don't allow enqueueing after stopping the pool
if(stop)
throw std::runtime_error("enqueue on stopped ThreadPool");
tasks.emplace([task](){ (*task)(); });
}
condition.notify_one();
return res;
}
typename std::result_of<F(Args...)>::type中的typename 應(yīng)該是為消除歧義的，或者因?yàn)榍短滓蕾嚸值年P(guān)系，做為一個(gè)堅(jiān)決不寫模板的普通程序員，這段代碼太難了……-> type 我倒是知道怎么回事，就是指明它的返回類型的一種方式result_of<F(Args...)> 應(yīng)該是指明了F是一個(gè)函數(shù)，簽名為Args...這個(gè)變參，Args是啥它不關(guān)系，它關(guān)心的是返回值的參數(shù)類型 所以有個(gè)type。
至于為什么函數(shù)入口是一個(gè)右值引用那就超出我的理解范圍了。難道說functional 必須要右值引用？那它的銷毀誰來管呢？這個(gè)線程來管嗎？這些坑我以后慢慢填。
前面我們說了tasks 只能接收void() 的函數(shù)類型，這里使用std::packaged_task<return_type()>完成對(duì)函數(shù)類型的推導(dǎo)，至于為什么不用 function<return_type()> ，因?yàn)檫@還不是最終放入tasks的對(duì)象，它要承接一個(gè)返回future<T>的工作，而package_task就是來打包返回future<T>的……
然后就是加鎖入隊(duì)+通知工作線程+返回future<T>的操作。本來是線程池最難理解的部分，反而顯得平淡無奇了，因?yàn)榍懊婺切┗ɡ锖诘牟僮饕呀?jīng)很好的打通了我們的理解能力。對(duì)于這個(gè)操作本來就有一點(diǎn)概念的，就顯得有種“就這？”的感覺……
