在大多数系统中，将每个任务指定给某个线程是不切实际的，不过可以利用现有的并发性，进行并发执行。线程池就提供了这样的功能:

  • 提交到线程池中的任务将并发执行
  • 提交的任务将会挂在任务队列上
  • 队列中的每一个任务都会被池中的工作线程所获取
  • 当任务执行完成后，线程再回到线程池中获取下一个任务


1 最简单的线程池

最简单的线程池中，线程就不需要等待其他线程完成对应任务了。任务没有返回值，不需要阻塞其他线程。

  1. //最简单的线程池，仅用于示例
  2. #ifndef THREAD_POOL
  3. #define THREAD_POOL
  4. #include <thread>
  5. #include <atomic>
  6. #include <vector>
  7. #include "21_threadsafe_queue2.hpp"
  9. class join_threads
  10. {
  11. private:
  12.     std::vector<std::thread>& threads;
  13. public:
  14.     explicit join_threads(std::vector<std::thread>& threads_): threads(threads_)
  15.     {}
  16.     ~join_threads()
  17.     {
  18.         for(unsigned long i=0;i<threads.size();++i)
  19.         {
  20.             // 析构函数中等待线程结束，RAII方式
  21.             if(threads[i].joinable())
  22.                 threads[i].join();
  23.         }
  24.     }
  25. };
  27. class thread_pool
  28. {
  29. private:
  30.     //注意这里声明的顺序，是为了保证析构时能以相反的顺序销毁（很重要）
  31.     std::atomic_bool done;
  32.     threadsafe_queue<std::function<void()> > work_queue;  // 使用线程安全队列保存任务
  33.     std::vector<std::thread> threads;  // 保存pool里的工作线程
  34.     join_threads joiner;  // join_threads使用RAII方式，保证pool销毁前所有线程能执行结束
  36.     void worker_thread()
  37.     {
  38.         while(!done)  // 原子bool设置为true时，保证每个正在运行的线程停止
  39.         {
  40.             std::function<void()> task;
  41.             if(work_queue.try_pop(task))  // 从任务队列获取到一个任务
  42.             {
  43.                 task();  // 执行任务，函数指针
  44.             }
  45.             else
  46.             {
  47.                 std::this_thread::yield();  // 没有任务，当前线程放弃执行，线程休息
  48.             }
  49.         }
  50.     }
  52. public:
  53.     thread_pool(): done(false),joiner(threads)
  54.     {
  55.         // pool中线程个数使用硬件支持的最大个数
  56.         unsigned const thread_count=std::thread::hardware_concurrency();
  58.         try
  59.         {
  60.             for(unsigned i=0;i<thread_count;++i)
  61.             {
  62.                 // 创建工作线程，每个线程都执行worker_thread函数，在此函数中获取任务处理
  63.                 threads.push_back(std::thread(&thread_pool::worker_thread, this));
  64.             }
  65.         }
  66.         catch(...)
  67.         {
  68.             done=true;  // 有异常时，设置done为true
  69.             throw;
  70.         }
  71.     }
  73.     ~thread_pool()
  74.     {
  75.         done=true;  // 11
  76.     }
  77.     // submit将一个要执行的任务包装成std::function函数指针，加入任务队列
  78.     template<typename FunctionType>
  79.     void submit(FunctionType f)
  80.     {
  81.         work_queue.push(std::function<void()>(f));  // 12
  82.     }
  83. };
  84. #endif

2 添加功能：任务返回值

上面的任务独立运行，不需要返回值。但是一种特殊的情况是，执行任务的线程需要返回一个结果到主线程上进行处理。可以使用submit()函数提交任务并返回条件变量或future，用来等待任务的完成，返回任务返回值。
在上面例子的基础上，使用packged_task和future代替std::function，可以返回任务返回值：

  1. //线程池，在29线程池中添加功能：可以返回每个任务的返回值
  2. #ifndef thread_pool22
  3. #define thread_pool22
  4. #include <future>
  5. #include <atomic>
  6. #include <vector>
  7. #include "21_threadsafe_queue2.hpp"
  8. #include "29_thread_pool.hpp"
  10. /*std::packaged_task<>实例是不可拷贝的，仅是可移动的，所以不能再使用std::function<>来实现任务队列
  11. 因为std::function<>需要存储可复制构造的函数对象。
  12. 这里包装一个自定义函数，用来处理只可移动的类型，就是一个带有函数操作符的类型擦除类
  13. */
  14. class function_wrapper
  15. {
  16. private:
  17.     struct impl_base
  18.     {
  19.         virtual void call() = 0;
  20.         virtual ~impl_base() {}
  21.     };
  23.     std::unique_ptr<impl_base> impl;
  25.     template <typename F>
  26.     struct impl_type : impl_base
  27.     {
  28.         F f;
  29.         impl_type(F &&f_) : f(std::move(f_)) {}
  30.         void call() { f(); }
  31.     };
  33. public:
  34.     template <typename F>
  35.     function_wrapper(F &&f) : impl(new impl_type<F>(std::move(f)))
  36.     {
  37.     }
  39.     function_wrapper() = default;
  41.     function_wrapper(function_wrapper &&other) : impl(std::move(other.impl))
  42.     {
  43.     }
  45.     function_wrapper &operator=(function_wrapper &&other)
  46.     {
  47.         impl = std::move(other.impl);
  48.         return *this;
  49.     }
  51.     void operator()() { impl->call(); }
  53.     function_wrapper(const function_wrapper &) = delete;
  54.     function_wrapper(function_wrapper &) = delete;
  55.     function_wrapper &operator=(const function_wrapper &) = delete;
  56. };
  58. class thread_pool2
  59. {
  60. private:
  61.     std::atomic_bool done;
  62.     threadsafe_queue<function_wrapper> work_queue; // 使用function_wrapper，而非使用std::function
  63.     std::vector<std::thread> threads;              // 保存pool里的工作线程
  64.     join_threads joiner;                           // join_threads使用RAII方式，保证pool销毁前所有线程能执行结束
  66.     void worker_thread()
  67.     {
  68.         while (!done)
  69.         {
  70.             function_wrapper task;
  71.             if (work_queue.try_pop(task))
  72.             {
  73.                 task();
  74.             }
  75.             else
  76.             {
  77.                 std::this_thread::yield();
  78.             }
  79.         }
  80.     }
  82. public:
  83.     thread_pool2() : done(false), joiner(threads)
  84.     {
  85.         // pool中线程个数使用硬件支持的最大个数
  86.         unsigned const thread_count = std::thread::hardware_concurrency();
  88.         try
  89.         {
  90.             for (unsigned i = 0; i < thread_count; ++i)
  91.             {
  92.                 // 创建工作线程，每个线程都执行worker_thread函数，在此函数中获取任务处理
  93.                 threads.push_back(std::thread(&thread_pool2::worker_thread, this));
  94.             }
  95.         }
  96.         catch (...)
  97.         {
  98.             done = true; // 有异常时，设置done为true
  99.             throw;
  100.         }
  101.     }
  103.     ~thread_pool2()
  104.     {
  105.         done = true;
  106.     }
  107.     // submit返回一个保存任务返回值的future，使用result_of获取函数functionType的返回类型
  108.     template <typename FunctionType>
  109.     std::future<typename std::result_of<FunctionType()>::type>
  110.     submit(FunctionType f)
  111.     {
  112.         // 设置函数functionType的返回类型的别名为result_type
  113.         typedef typename std::result_of<FunctionType()>::type result_type;
  115.         std::packaged_task<result_type()> task(std::move(f)); // 封装一个异步任务，任务执行函数f
  116.         std::future<result_type> res(task.get_future());      // 获取异步任务的future
  117.         work_queue.push(std::move(task));                     // 将任务添加到任务队列中
  118.         return res;                                           // 返回future给submit函数的调用者
  119.     }
  120. };
  121. #endif

当选择“因为能并发执行，最小工作块值的一试”时，就需要谨慎了。向线程池提交任务有一定的开销；让工作线程执行这个任务，并且将返回值保存在std::future<>中，对于太小的任务，这样的开销不划算。如果任务块太小，使用线程池的速度可能都不及单线程。

3 添加功能：手动触发任务执行

最简单的方法就是在thread_pool中添加一个新函数，来执行任务队列上的任务，并对线程池进行管理。高级线程池的实现可能会在等待函数中添加逻辑，或等待其他函数来处理这个任务，优先的任务会让其他的任务进行等待。下面清单中的实现，就展示了一个新run_pending_task()函数：

  1. /*
  2. run_pending_task()的实现去掉了在worker_thread()函数的主循环。
  3. 函数任务队列中有任务的时候执行任务，要是没有的话就会让操作系统对线程进行重新分配。
  4. */
  5. void run_pending_task()
  6. {
  7.     function_wrapper task;
  8.     if (work_queue.try_pop(task))
  9.     {
  10.         task();
  11.     }
  12.     else
  13.     {
  14.         std::this_thread::yield();
  15.     }
  16. }

4 避免队列中的任务竞争

随着处理器的增加，任务队列上就会有很多的竞争（添加任务和多线程获取任务），这会让性能下降。使用无锁队列会让任务没有明显的等待，但乒乓缓存会消耗大量的时间。
为了避免乒乓缓存，每个线程建立独立的任务队列。这样，每个线程就会将新任务放在自己的任务队列上，并且当线程上的任务队列没有任务时，去全局的任务列表中取任务。

  1. class thread_pool3
  2. {
  3. private:
  4.     std::atomic_bool done;
  5.     threadsafe_queue<function_wrapper> work_queue; // 使用function_wrapper，而非使用std::function
  6.     typedef std::queue<function_wrapper> local_queue_type;
  7.     // unique_ptr指向每个线程本地（thread_local)的工作队列
  8.     static thread_local std::unique_ptr<local_queue_type> local_work_queue;
  9.     std::vector<std::thread> threads;                                       // 保存pool里的工作线程
  10.     // join_threads使用RAII方式，保证pool销毁前所有线程能执行结束
  11.     join_threads joiner;
  13.     void worker_thread()
  14.     {
  15.         local_work_queue.reset(new local_queue_type); // 3
  16.         while (!done)
  17.         {
  18.             run_pending_task();
  19.         }
  20.     }
  22. public:
  23.     thread_pool3() : done(false), joiner(threads)
  24.     {
  25.         // pool中线程个数使用硬件支持的最大个数
  26.         unsigned const thread_count = std::thread::hardware_concurrency();
  28.         try
  29.         {
  30.             for (unsigned i = 0; i < thread_count; ++i)
  31.             {
  32.                 // 创建工作线程，每个线程都执行worker_thread函数，在此函数中获取任务处理
  33.                 threads.push_back(std::thread(&thread_pool3::worker_thread, this));
  34.             }
  35.         }
  36.         catch (...)
  37.         {
  38.             done = true; // 有异常时，设置done为true
  39.             throw;
  40.         }
  41.     }
  43.     ~thread_pool3()
  44.     {
  45.         done = true;
  46.     }
  47.     // submit返回一个保存任务返回值的future，使用result_of获取函数functionType的返回类型
  48.     template <typename FunctionType>
  49.     std::future<typename std::result_of<FunctionType()>::type>
  50.     submit(FunctionType f)
  51.     {
  52.         // 设置函数functionType的返回类型的别名为result_type
  53.         typedef typename std::result_of<FunctionType()>::type result_type;
  55.         std::packaged_task<result_type()> task(std::move(f)); // 封装一个异步任务，任务执行函数f
  56.         std::future<result_type> res(task.get_future());      // 获取异步任务的future
  57.         if (local_work_queue)                                 // 检查当前线程是否具有一个工作队列，如果有则将任务放入本地队列
  58.         {
  59.             local_work_queue->push(std::move(task));
  60.         }
  61.         else
  62.         {
  63.             work_queue.push(std::move(task)); // 将任务添加到全局任务队列中
  64.         }
  66.         return res; // 返回future给submit函数的调用者
  67.     }
  68.     /*
  69.     run_pending_task()的实现去掉了在worker_thread()函数的主循环。
  70.     函数任务队列中有任务的时候执行任务，要是没有的话就会让操作系统对线程进行重新分配。
  71.     */
  72.     void run_pending_task()
  73.     {
  74.         function_wrapper task;
  75.         if (local_work_queue && !(local_work_queue->empty())) // 如果本地队列有任务，则优先处理本地队列的任务
  76.         {
  77.             task = std::move(local_work_queue->front());
  78.             local_work_queue->pop();
  79.             task();
  80.         }
  81.         else if (work_queue.try_pop(task)) //否则，在全局队列获取任务
  82.         {
  83.             task();
  84.         }
  85.         else
  86.         {
  87.             std::this_thread::yield();
  88.         }
  89.     }
  90. };

本地任务队列能有效的避免竞争，不过当任务分配不均时，造成的结果就是：某个线程本地队列中有很多任务的同时，其他线程无所事事。例如：举一个快速排序的例子，一开始的数据块能在线程池上被处理，因为剩余部分会放在工作线程的本地队列上进行处理，这样的使用方式也违背使用线程池的初衷。

5 从别的线程队列中窃取任务

上面任务分配不均时的困境时有解的，本地工作队列和全局工作队列上没有任务时，可从别的线程队列中窃取任务。为了让没有任务的线程能从其他线程的任务队列中获取任务，就需要本地任务列表可以被其他线程访问，这样才能让run_pending_tasks()窃取任务。需要每个线程在线程池队列上进行注册，或由线程池指定一个线程。同样，还需要保证数据队列中的任务适当的被同步和保护，这样队列的不变量就不会被破坏。

  1. /*线程池，在30线程池基础上增加功能，本篇最终版本：
  2. 1. 为每个线程建立独立的任务队列，配合全局任务队列使用
  3. 2. 支持从其他线程的本地队列窃取任务
  4. */
  5. #ifndef THREAD_POOL4
  6. #define THREAD_POOL4
  7. #include <future>
  8. #include <atomic>
  9. #include <vector>
  10. #include <queue>
  11. #include "21_threadsafe_queue2.hpp"
  13. // 统一管理线程池所有线程，负责等待线程结束和销毁
  14. class join_threads
  15. {
  16. private:
  17.     std::vector<std::thread> &threads;
  19. public:
  20.     explicit join_threads(std::vector<std::thread> &threads_) : threads(threads_)
  21.     {
  22.     }
  23.     ~join_threads()
  24.     {
  25.         for (unsigned long i = 0; i < threads.size(); ++i)
  26.         {
  27.             // 析构函数中等待线程结束，RAII方式
  28.             if (threads[i].joinable())
  29.                 threads[i].join();
  30.         }
  31.     }
  32. };
  34. /*std::packaged_task<>实例是不可拷贝的，仅是可移动的，所以不能再使用std::function<>来实现任务队列
  35. 因为std::function<>需要存储可复制构造的函数对象。
  36. 这里包装一个自定义函数，用来处理只可移动的类型，就是一个带有函数操作符的类型擦除类
  37. */
  38. class function_wrapper
  39. {
  40. private:
  41.     struct impl_base
  42.     {
  43.         virtual void call() = 0;
  44.         virtual ~impl_base() {}
  45.     };
  47.     std::unique_ptr<impl_base> impl;
  49.     template <typename F>
  50.     struct impl_type : impl_base
  51.     {
  52.         F f;
  53.         impl_type(F &&f_) : f(std::move(f_)) {}
  54.         void call() { f(); }
  55.     };
  57. public:
  58.     template <typename F>
  59.     function_wrapper(F &&f) : impl(new impl_type<F>(std::move(f)))
  60.     {
  61.     }
  63.     function_wrapper() = default;
  65.     function_wrapper(function_wrapper &&other) : impl(std::move(other.impl))
  66.     {
  67.     }
  69.     function_wrapper &operator=(function_wrapper &&other)
  70.     {
  71.         impl = std::move(other.impl);
  72.         return *this;
  73.     }
  75.     void operator()() { impl->call(); }
  77.     function_wrapper(const function_wrapper &) = delete;
  78.     function_wrapper(function_wrapper &) = delete;
  79.     function_wrapper &operator=(const function_wrapper &) = delete;
  80. };
  82. /*
  83. 基于锁的任务窃取队列，代替普通的线程安全队列
  84. 该队列支持从前端和后端获取数据，即先进先出和后进先出两种模式，后进先出可以用于其他线程窃取任务
  85. */
  86. class work_stealing_queue
  87. {
  88. private:
  89.     typedef function_wrapper data_type;
  90.     std::deque<data_type> the_queue; // 使用deque队列保存实际function_wrapper
  91.     mutable std::mutex the_mutex; // 互斥锁用于控制对the_mutex的访问
  93. public:
  94.     work_stealing_queue() {}
  95.     // 拷贝和赋值构造函数不创建
  96.     work_stealing_queue(const work_stealing_queue& other) = delete;
  97.     work_stealing_queue& operator=(const work_stealing_queue& other) = delete;
  98.     // 由互斥锁控制在队列前端插入数据
  99.     void push(data_type data)
  100.     {
  101.         std::lock_guard<std::mutex> lock(the_mutex);
  102.         the_queue.push_front(std::move(data));
  103.     }
  105.     bool empty() const
  106.     {
  107.         std::lock_guard<std::mutex> lock(the_mutex);
  108.         return the_queue.empty();
  109.     }
  110.     // 由互斥锁控制在队列前端获取数据
  111.     bool try_pop(data_type& res)
  112.     {
  113.         std::lock_guard<std::mutex> lock(the_mutex);
  114.         if(the_queue.empty())
  115.         {
  116.             return false;
  117.         }
  119.         res=std::move(the_queue.front());
  120.         the_queue.pop_front();
  121.         return true;
  122.     }
  123.     // 由互斥锁控制在队列后端获取数据
  124.     bool try_steal(data_type& res)
  125.     {
  126.         std::lock_guard<std::mutex> lock(the_mutex);
  127.         if(the_queue.empty())
  128.         {
  129.             return false;
  130.         }
  132.         res=std::move(the_queue.back());
  133.         the_queue.pop_back();
  134.         return true;
  135.     }
  136. };
  138. class thread_pool4
  139. {
  140. private:
  141.     typedef function_wrapper task_type;
  143.     // 注意这里声明的顺序，是为了保证析构时能以相反的顺序销毁（很重要）
  144.     std::atomic_bool done;
  145.     threadsafe_queue<task_type> pool_work_queue; // 全局任务队列
  146.     std::vector<std::unique_ptr<work_stealing_queue>> queues; // 保存每个线程任务队列的全局队列
  147.     // 保存pool里的工作线程
  148.     std::vector<std::thread> threads;
  149.     // join_threads使用RAII方式，保证pool销毁前所有线程能执行结束
  150.     join_threads joiner;
  152.     static thread_local work_stealing_queue* local_work_queue; // 每个线程都有一个可以窃取的任务队列，不再使用普通的队列
  153.     static thread_local unsigned my_index;
  155.     void worker_thread(unsigned my_index_)
  156.     {
  157.         my_index = my_index_;
  158.         // 每个线程根据自己的序号从全局队列中获取自己的任务队列。这意味着根据index可以获取任意线程的任务队列
  159.         local_work_queue = queues[my_index].get();
  160.         while (!done)
  161.         {
  162.             run_pending_task();
  163.         }
  164.     }
  166.     bool pop_task_from_local_queue(task_type& task)
  167.     {
  168.         return local_work_queue && local_work_queue->try_pop(task);
  169.     }
  171.     bool pop_task_from_pool_queue(task_type& task)
  172.     {
  173.         return pool_work_queue.try_pop(task);
  174.     }
  176.     bool pop_task_from_other_thread_queue(task_type& task)  // 从其他线程窃取任务
  177.     {
  178.         for(unsigned i=0;i < queues.size();++i)
  179.         {
  180.             // 根据index从其他线程的任务队列后端窃取任务
  181.             unsigned const index = (my_index + i + 1) % queues.size();
  182.             if(queues[index]->try_steal(task))
  183.             {
  184.                 return true;
  185.             }
  186.         }
  187.         return false;
  188.     }
  190. public:
  191.     thread_pool4() : done(false), joiner(threads)
  192.     {
  193.         // pool中线程个数使用硬件支持的最大个数
  194.         unsigned const thread_count = std::thread::hardware_concurrency();
  196.         try
  197.         {
  198.             for (unsigned i = 0; i < thread_count; ++i)
  199.             {
  200.                 // 当每个线程被创建，就创建了一个属于自己的工作队列，放入全局队列中
  201.                 queues.push_back(std::unique_ptr<work_stealing_queue>(new work_stealing_queue));
  202.                 // 创建工作线程，每个线程都执行worker_thread函数，在此函数中获取任务处理
  203.                 threads.push_back(std::thread(&thread_pool4::worker_thread, this, i));//传入参数i为my_index
  204.             }
  205.         }
  206.         catch (...)
  207.         {
  208.             done = true; // 有异常时，设置done为true
  209.             throw;
  210.         }
  211.     }
  213.     ~thread_pool4()
  214.     {
  215.         done = true;
  216.     }
  218.     // submit返回一个保存任务返回值的future，使用result_of获取函数functionType的返回类型
  219.     template <typename FunctionType>
  220.     std::future<typename std::result_of<FunctionType()>::type>
  221.     submit(FunctionType f)
  222.     {
  223.         // 设置函数functionType的返回类型的别名为result_type
  224.         typedef typename std::result_of<FunctionType()>::type result_type;
  226.         std::packaged_task<result_type()> task(std::move(f)); // 封装一个异步任务，任务执行函数f
  227.         std::future<result_type> res(task.get_future());      // 获取异步任务的future
  228.         if (local_work_queue)                                 // 检查当前线程是否具有一个工作队列，如果有则将任务放入本地队列
  229.         {
  230.             local_work_queue->push(std::move(task)); // 线程本地任务队列
  231.         }
  232.         else
  233.         {
  234.             pool_work_queue.push(std::move(task)); // 将任务添加到全局任务队列中
  235.         }
  237.         return res; // 返回future给submit函数的调用者
  238.     }
  239.     /*
  240.     run_pending_task()的实现去掉了在worker_thread()函数的主循环。
  241.     函数任务队列中有任务的时候执行任务，要是没有的话就会让操作系统对线程进行重新分配。
  242.     */
  243.     void run_pending_task()
  244.     {
  245.         task_type task;
  246.         if (pop_task_from_local_queue(task) || // 第一优先级：从线程自己的队列获取任务
  247.             pop_task_from_pool_queue(task) || // 第二优先级：从线程池队列获取任务
  248.             pop_task_from_other_thread_queue(task)) // 最低优先级：窃取其他线程的任务
  249.         {
  250.             task();
  251.         }
  252.         else
  253.         {
  254.             std::this_thread::yield();
  255.         }
  256.     }
  257. };
  258. #endif

pop_task_from_other_thread_queue()会遍历池中所有线程的任务队列，然后尝试窃取任务。为了避免每个线程都尝试从列表中的第一个线程上窃取任务，每一个线程都会从下一个线程开始遍历，通过自身的线程序号来确定开始遍历的线程序号。

使用线程池有很多好处，还有很多的方式能为某些特殊用法提升性能。特别是还没有探究动态变换大小的线程池，即使线程被阻塞的时候(例如：I/O或互斥锁)，程序都能保证CPU最优的使用率。