C++11 动态线程池

Question

最近，我一直在寻找一个用于线程并发任务的库。理想情况下，是一个在线程上调用函数的简单接口。任何时候都有 n 个线程，有些线程完成得比其他线程快，并且到达的时间不同。

首先我在尝试 Rx，它在 C++ 中很棒。我还研究了 Blocks 和 TBB，但它们都依赖于平台。对于我的原型，我需要保持平台独立，因为我们还不知道它将在什么上运行，并且可以在做出决定时进行更改。

C++11 有很多关于线程和并发的东西，我发现了很多类似线程池的例子。

https://github.com/bilash/threadpool

类似的项目对 std::thread 和 std::mutex 使用相同的 lambda 表达式。

这看起来非常适合我的需要。有一些问题。池以定义数量的线程启动，任务将排队，直到线程空闲。

如何添加新线程？删除过期的线程？（。加入（）？？）

显然，这对于已知数量的线程要容易得多，因为它们可以在 ctor 中初始化，然后在 dtor 中 join()。

有 C++ 并发经验的人在这里有什么提示或指示吗？

Answer 1

1. 从系统可以支持的最大线程数开始：

   int Num_Threads =  thread::hardware_concurrency();
   

2. 对于一个高效的线程池实现，一旦根据 Num_Threads 创建了线程，最好不要创建新的，也不要销毁旧的（通过加入）。会有性能损失，甚至可能使您的应用程序运行速度比串行版本慢。

   每个 C++11 线程都应该以无限循环的方式在其函数中运行，不断等待新任务的获取和运行。

   以下是如何将此类函数附加到线程池：

   int Num_Threads = thread::hardware_concurrency();
   vector<thread> Pool;
   for(int ii = 0; ii < Num_Threads; ii++)
   {  Pool.push_back(thread(Infinite_loop_function));}
   

3. Infinite_loop_function

   这是一个等待任务队列的“while(true)”循环

   void The_Pool:: Infinite_loop_function()
   {
       while(true)
       {
           {
               unique_lock<mutex> lock(Queue_Mutex);
   
               condition.wait(lock, []{return !Queue.empty()});
               Job = Queue.front();
               Queue.pop();
           }
           Job(); // function<void()> type
       }
   };
   

4. 创建一个函数将作业添加到您的队列中

   void The_Pool:: Add_Job(function<void()> New_Job)
   {
       {
           unique_lock<mutex> lock(Queue_Mutex);
           Queue.push(New_Job);
       }
       condition.notify_one();
   }
   

5. 将任意函数绑定到您的队列

   Pool_Obj.Add_Job(std::bind(&Some_Class::Some_Method, &Some_object));
   

一旦您集成了这些成分，您就拥有了自己的动态线程池。这些线程始终运行，等待作业完成。