一.简介
progschj/ThreadPool（链接：GitHub - progschj/ThreadPool: A simple C++11 Thread Pool implementation ）是一个用到C++11特性的跨平台线程池，可以在windows，linux上运行。其只用不到100行代码就实现了线程池的基本功能，麻雀虽小五脏俱全，非常适合初学者学习。本文对其源码进行分析。
二.源码
ThreadPool.h
#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 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;
};
// 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 pool
if(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
使用例子：
example.cpp
#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;
}三.总流程分析
上述例子中通过ThreadPool pool(4)创建了4个工作线程，将任务
{
std::cout << "hello " << i << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "world " << i << std::endl;
return i*i;
}
添加到任务队列（ThreadPool的std::queue< std::function<void()> > tasks）中，然后在创建的子线程中启动这些任务。最后通过：
for(auto && result: results)
std::cout << result.get() << ' ';
std::cout << std::endl;阻塞等待子线程执行完毕，打印任务的返回值（i*i的结果）。
四.源码分析
首先我们分析ThreadPool::enqueue函数
（1）
template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args)
-> std::future<typename std::result_of<F(Args...)>::type>template<class F, class... Args>等价于template<typename F, typename... Args>，其中由于有参数class F，所以可以把lambda表达式或函数作为enqueue的第一个参数传进去。class... Args是可变长参数模板，负责把参数传到函数F里面。所以可以在example.cpp中实现：
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;
}上述例子中
[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;
}是lambda表达式，[i]是捕获列表，表示按值捕获i，所以大括号的代码中可以打印i的值
可以用函数替换掉lambda表达式作为参数传到enqueue里面，所以上述例子可以改写成：
#include <iostream>
#include <vector>
#include <chrono>
#include "ThreadPool.h"
int fun(int 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;
}
int main()
{
ThreadPool pool(4);
std::vector< std::future<int> > results;
for (int i = 0; i < 8; ++i) {
results.emplace_back(pool.enqueue(fun, i));
}
for(auto && result: results)
std::cout << "get:" << result.get() << std::endl;
//std::cout << std::endl;
return 0;
}这里面实参fun对应形参F&& f，实参i对应形参Args&&... args 。
（2）
template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args)
-> std::future<typename std::result_of<F(Args...)>::type>这里的“-> std::future<typename std::result_of<F(Args...)>::type>”是返回类型后置，和auto结合起来进行使用，共同完成函数返回值类型的推导，这里enqueue函数的返回值就是std::future<typename std::result_of<F(Args...)>::type>。
由于在example.cpp中
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;
}std::result_of<F(Args...)为该lambda表达式的返回值(i*i）的类型，也就是int。所以这里enqueue函数的返回值为std::future<int>类型。在C++14中可省略箭头返回值部分，直接将函数返回类型设置为auto，所以enqueue函数可以去掉“-> std::future<typename std::result_of<F(Args...)>::type>”这部分直接写成：
auto ThreadPool::enqueue(F&& f, Args&&... args)（3）
using return_type = typename std::result_of<F(Args...)>::type;using用来给typename std::result_of<F(Args...)>::type取别名，相当于typedef
（4）
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();这里创建得到一个packaged_task对象，进而通过它的get_future()成员函数得到已经配对完成的future对象。随后，我们通过ThreadPool::ThreadPool中创建的子线程来执行这个packaged_task，也就是执行线程函数
{
std::cout << "hello " << i << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "world " << i << std::endl;
return i*i;
}开始准备结果数据。与此同时，我们在main函数中使用future对象的get()函数
for(auto && result: results)
std::cout << result.get() << ' ';
std::cout << std::endl;来等待分支线程执行完毕。一旦分支线程的线程函数执行完毕返回结果数据（return i*i)，result数据就会被主线程中的get()函数获得而返回。
(5)
{
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();std::queue< std::function<void()> > tasks是任务队列。std::function等价于函数指针，绑定{ (*task)(); }。这里将任务放到任务队列中，为了避免有多个线程同时对tasks进行操作，所以这里得加锁std::unique_lock<std::mutex> lock(queue_mutex)，保证在任一时刻，只能有一个线程访问它。然后通过condition.notify_one() 通知ThreadPool::ThreadPool中第一个进入阻塞或者等待的线程，取消对线程的阻塞。
我们接着分析inline ThreadPool::ThreadPool(size_t threads)函数
（6）
for(size_t i = 0;i<threads;++i)
workers.emplace_back(
...通过这里创建了threads个子线程
（7）
{
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;
.....这里通过条件变量，让任务队列为空的时候可以阻塞，避免不断轮询判断缓冲区是否为空消耗CPU。
（8）
task = std::move(this->tasks.front());
this->tasks.pop();这里把任务出队列。其用到了移动赋值，避免拷贝，提高了性能。
（9）
task();这里相当于执行了[task](){ (*task)(); }。而(*task)()执行了主函数
{
std::cout << "hello " << i << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "world " << i << std::endl;
return i*i;
}中的内容，从而实现了在子线程中执行任务
