C++11 中的线程池

相关问题:

关于 C++11：

• C++11：std::线程池？ https://stackoverflow.com/questions/12993451/c11-stdthread-pooled
• C++11 中的 async(launch::async) 是否会使线程池过时，以避免昂贵的线程创建？ https://stackoverflow.com/questions/14351352/will-asynclaunchasync-in-c11-make-thread-pools-obsolete-for-avoiding-expen

关于升压：

• C++ boost 线程重用线程 https://stackoverflow.com/questions/2659065/c-boost-thread-reusing-threads
• boost::thread 并创建它们的池！ https://stackoverflow.com/questions/3458397/boostthread-and-creating-a-pool-of-them

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

我的代码如下所示：

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

我不想在每次迭代中创建和加入线程，而是更愿意在每次迭代中将任务发送到我的工作线程，并且只创建一次。

这是改编自我的答案 https://stackoverflow.com/a/32593766/6909078到另一个非常相似的帖子。

让我们构建一个ThreadPool class:

class ThreadPool {
public:
    void Start();
    void QueueJob(const std::function<void()>& job);
    void Stop();
    bool 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
    for (uint32_t ii = 0; ii < num_threads; ++ii) {
        threads.emplace_back(std::thread(&ThreadPool::ThreadLoop,this))
    }
}

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

3. 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

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

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

如果有一些语法错误，我深表歉意，我输入了这段代码，但我的记忆力很差。抱歉我无法提供 您完整的线程池代码；这会违反我的职业操守。

Notes:

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