c++ - C++11 中的线程池

Question

相关问题:

关于 C++11:

• C++11: std::thread pooled?
• Will async(launch::async) in C++11 make thread pools obsolete for avoiding expensive thread creation?

关于 Boost:

• C++ boost thread reusing threads
• boost::thread and creating a pool of them!

---

我如何获得一个线程池来向发送任务，而不是一遍又一遍地创建和删除它们？这意味着持久线程无需加入即可重新同步。

---

我的代码如下所示:

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

本文改编自 my answer到另一个非常相似的帖子。

让我们构建一个 ThreadPool 类:

class ThreadPool {
public:
    void Start();
    void QueueJob(const std::function<void()>& job);
    void Stop();
    void busy();

private:
    void ThreadLoop();

    bool should_terminate = false;           // Tells threads to stop looking for jobs
    std::mutex queue_mutex;                  // Prevents data races to the job queue
    std::condition_variable mutex_condition; // Allows threads to wait on new jobs or termination 
    std::vector<std::thread> threads;
    std::queue<std::function<void()>> jobs;
};

1. ThreadPool::Start

为了高效的线程池实现，一旦根据num_threads创建线程，最好不要创建新的或销毁旧的(通过加入)。会有性能损失，甚至可能使您的应用程序比串行版本慢。因此，我们保留了一个可以随时使用的线程池(如果它们尚未运行作业)。

每个线程都应该运行自己的无限循环，不断等待新任务的抓取和运行。

void ThreadPool::Start() {
    const uint32_t num_threads = std::thread::hardware_concurrency(); // Max # of threads the system supports
    threads.resize(num_threads);
    for (uint32_t i = 0; i < num_threads; i++) {
        threads.at(i) = std::thread(ThreadLoop);
    }
}

1. ThreadPool::ThreadLoop

无限循环功能。这是一个等待任务队列打开的 while (true) 循环。

void ThreadPool::ThreadLoop() {
    while (true) {
        std::function<void()> job;
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            mutex_condition.wait(lock, [this] {
                return !jobs.empty() || should_terminate;
            });
            if (should_terminate) {
                return;
            }
            job = jobs.front();
            jobs.pop();
        }
        job();
    }
}

1. ThreadPool::QueueJob

向池中添加新作业；使用锁，以免发生数据争用。

void ThreadPool::QueueJob(const std::function<void()>& job) {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        jobs.push(job);
    }
    mutex_condition.notify_one();
}

使用它:

thread_pool->QueueJob([] { /* ... */ });

4. ThreadPool::busy

void ThreadPool::busy() {
    bool poolbusy;
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        poolbusy = jobs.empty();
    }
    return poolbusy;
}

busy()函数可以用在while循环中，这样主线程就可以在调用线程池析构函数之前等待线程池完成所有任务。

5. ThreadPool::Stop

停止游泳池。

void ThreadPool::Stop() {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        should_terminate = true;
    }
    mutex_condition.notify_all();
    for (std::thread& active_thread : threads) {
        active_thread.join();
    }
    threads.clear();
}

集成这些成分后，您就拥有了自己的动态线程池。这些线程总是运行，等待工作要做。

如果有一些语法错误，我深表歉意，我输入了这段代码，但我记性不好。抱歉我不能提供你完整的线程池代码；这会违反我的职业操守。

注意事项:

• 使用匿名代码块，以便在它们退出时，在其中创建 std::unique_lock 变量超出范围，解锁互斥锁。
• ThreadPool::Stop 不会终止任何当前正在运行的作业，它只是通过 active_thread.join() 等待它们完成。