C ++ 11中的线程池

Question

我如何获得一个线程池以将任务发送到，而不是一遍又一遍地创建和删除它们？这意味着持久性线程无需加入即可重新同步。我有看起来像这样的代码：namespace {  std::vector<std::thread> workers;  int total = 4;  int arr[4] = {0};  void each_thread_does(int i) {    arr[i] += 2;  }}int main(int argc, char *argv[]) {  for (int i = 0; i < 8; ++i) { // for 8 iterations,    for (int j = 0; j < 4; ++j) {      workers.push_back(std::thread(each_thread_does, j));    }    for (std::thread &t: workers) {      if (t.joinable()) {        t.join();      }    }    arr[4] = std::min_element(arr, arr+4);  }  return 0;}与其在每个迭代中创建和加入线程，不如在每个迭代中将任务发送到我的工作线程，并且只创建一次。

Answer 1

线程池意味着您所有的线程一直在运行–换句话说，线程函数永远不会返回。为了使线程有意义，您必须设计一个线程间通信系统，既要告诉线程有事情要做，又要传达实际的工作数据。

通常，这将涉及某种并发数据结构，并且每个线程可能会休眠在某种条件变量上，在有工作要做时会通知该条件变量。收到通知后，一个或几个线程将唤醒，从并发数据结构中恢复任务，对其进行处理，并以类似的方式存储结果。

然后，线程将继续检查是否还有更多工作要做，如果还没有，请回到睡眠状态。

结果是您必须自己设计所有这一切，因为没有一种普遍适用的自然的“工作”概念。这需要大量的工作，并且您必须解决一些细微的问题。（如果您喜欢幕后负责线程管理的系统，则可以在Go中编程。）

Answer 2

希望它能对您有所帮助：

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() || therminate_pool});

            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));

整合这些成分后，您将拥有自己的动态线程池。这些线程始终运行，等待作业完成。

如果出现语法错误，我深表歉意，我键入了这些代码，并且记忆力很差。抱歉，我无法为您提供完整的线程池代码，因为这会破坏我的工作完整性。

编辑：要终止池，请调用shutdown（）方法：

XXXX::shutdown(){

{   unique_lock<mutex> lock(threadpool_mutex);

    terminate_pool = true;} // use this flag in condition.wait

condition.notify_all(); // wake up all threads.

// Join all threads.

for(std::thread &every_thread : thread_vector)

{   every_thread.join();}

thread_vector.empty();  

stopped = true; // use this flag in destructor, if not set, call shutdown() 

}