參考項目

https://github.com/progschj/ThreadPool

源碼分析

// 常規頭文件保護宏, 避免重復 include
#ifndef THREAD_POOL_H
#define THREAD_POOL_H// 線程池, 存儲線程對象;
#include <vector>// 任務隊列, 雙向都可操作隊列, queue 不能刪除首個元素
#include <queue>// 智能指針
#include <memory>// c++11 線程對象
#include <thread>// 鎖保護隊列多線程任務添加, 刪除的安全;
#include <mutex>// 條件變量用來 condition wait 和 notify; 即事件的通知和阻塞等待
#include <condition_variable>// future 用來獲取更友好的封裝任務(函數), 并獲取返回值;
#include <future>// function 任務隊列
#include <functional>// 非法場景拋出異常
#include <stdexcept>// 為什么不用模板? 因為 enqueue 是模板, 可以兼容幾乎所有場景;
class ThreadPool {
public:// 設置線程池大小ThreadPool(size_t);// 添加函數, function 和 argstemplate<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 themstd::vector< std::thread > workers;// the task queuestd::queue< std::function<void()> > tasks;// synchronizationstd::mutex queue_mutex;std::condition_variable condition;bool 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)// 添加任務, 用 emplace_back 的形勢, 避免某些類型不支持拷貝;workers.emplace_back(// lambda 捕獲 this 對象, 用于操作任務隊列;[this]{// 死循環等待任務for(;;){// 任務獲取 void() 類型 統一封裝, 后面用 packaged_task 封裝不同的std::function<void()> task;{// 構建對象用于 conditionstd::unique_lock<std::mutex> lock(this->queue_mutex);// 等待線程池停止, 或者有任務;this->condition.wait(lock, [this]{ return this->stop || !this->tasks.empty(); });// 如果有任務 則 false, 即使停止也需要執行完任務之后再停止隊列// 如果請求停止, 且沒有任務，則終止;if(this->stop && this->tasks.empty())return;// 獲取第一個任務;task = std::move(this->tasks.front());this->tasks.pop();}// 執行獲取到任務;task();}});
}// 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;// 將對象通過 bind 打包成可調用對象, 并封裝到 packeaged_task;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 poolif(stop)throw std::runtime_error("enqueue on stopped ThreadPool");// 用 lambda 再封裝是為了統一函數格式  void()// task 是 shared_ptr, 避免拷貝;tasks.emplace([task](){ (*task)(); });}// 提醒阻塞線程有新的任務;condition.notify_one();// 返回獲取函數返回值的具柄, 這樣可以獲取返回值;return res;
}// the destructor joins all threads
inline ThreadPool::~ThreadPool()
{{// 加鎖請求停止std::unique_lock<std::mutex> lock(queue_mutex);stop = true;}// 提醒所有阻塞線程, 需要停止線程池condition.notify_all();// join每個 thread 對象; 可以用 std::future 來作為 thread 對象, 讓 std 管理生命周期;// 除非開發者有更加細致的管理, 如 優先級, 棧, 添加屬性之類的操作;for(std::thread &worker: workers)worker.join();
}#endif

使用案例一

// create thread pool with 4 worker threads
ThreadPool pool(4);// enqueue and store future
auto result = pool.enqueue([](int answer) { return answer; }, 42);// get result from future
std::cout << result.get() << std::endl;

使用案例二

#include <iostream>
#include <vector>
#include <chrono>#include "ThreadPool.h"int main()
{ThreadPool pool(4);std::vector< std::future<int> > results;for(int i = 0; i < 8; ++i) {results.emplace_back(pool.enqueue([i] {std::cout << "hello " << i << std::endl;std::this_thread::sleep_for(std::chrono::seconds(1));std::cout << "world " << i << std::endl;return i*i;}));}for(auto && result: results)std::cout << result.get() << ' ';std::cout << std::endl;return 0;
}

無注釋版本

#ifndef THREAD_POOL_H
#define THREAD_POOL_H#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>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 themstd::vector< std::thread > workers;// the task queuestd::queue< std::function<void()> > tasks;// synchronizationstd::mutex queue_mutex;std::condition_variable condition;bool 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();}});
}// 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 poolif(stop)throw std::runtime_error("enqueue on stopped ThreadPool");tasks.emplace([task](){ (*task)(); });}condition.notify_one();return res;
}// 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();
}#endif