线程队列C++

Question

“”原帖已编辑“”

如何在C++中为两个for循环创建线程池？我需要对0到6之间的每一个数字运行22次start_thread函数，并且根据我使用的机器的不同，我将有一个灵活的可用线程数。如何创建一个池来将空闲线程分配给嵌套循环的下一个？

for (int t=0; t <22; t++){
    for(int p=0; p<6; p++){
        thread th1(start_thread, p);
        thread th2(start_thread, p);
        th1.join();
        th2.join();
     }
}

Answer 1

不太确定你想要什么，但也许是这样的。

for (int t=0; t <22; t++){
        std::vector<std::thread> th;
        for(int p=0; p<6; p++){
                th.emplace_back(std::thread(start_thread, p));
        }
        for(int p=0; p<6; p++){
                th[i].join();
        }
}

(或者可能改变这两个循环)

如果要控制线程数，编辑

#include <iostream>
#include <thread>
#include <vector>

void
start_thread(int t, int p)
{
  std::cout << "th " << t << ' ' << p << '\n';
}

void
join_all(std::vector<std::thread> &th)
{
  for(auto &e: th)
  {
    e.join();
  }
  th.clear();
}

int
main()
{
  std::size_t max_threads=std::thread::hardware_concurrency();
  std::vector<std::thread> th;
  for(int t=0; t <22; ++t)
  {
    for(int p=0; p<6; ++p)
    {
      th.emplace_back(std::thread(start_thread, t, p));
      if(size(th)==max_threads)
      {
        join_all(th);
      }
    }
  } 
  join_all(th);
  return 0;
}

Answer 2

如果您不希望依赖第三方库，这非常简单。

只需创建一些您喜欢的线程，让它们从某个队列中选择一个“作业”。

例如：

#include <iostream>
#include <mutex>
#include <chrono>
#include <vector>
#include <thread>
#include <queue>

void work(int p)
{
  // do the "work"
  std::this_thread::sleep_for(std::chrono::milliseconds(200));
  std::cout << p << std::endl;
}

std::mutex m;
std::queue<int> jobs;
void worker()
{
  while (true)
  {
    int job(0);
    // sync access to the jobs queue
    {
      std::lock_guard<std::mutex> l(m);
      if (jobs.empty())
        return;
      job = jobs.front();
      jobs.pop();
    }
    work(job);
  }
}

int main()
{
  // queue all jobs
  for (int t = 0; t < 22; t++) {
    for (int p = 0; p < 6; p++) {
      jobs.push(p);
    }
  }

  // create reasonable number of threads
  static const int n = std::thread::hardware_concurrency();
  std::vector<std::thread> threads;
  for (int i = 0; i < n; ++i)
    threads.emplace_back(std::thread(worker));
  // wait for all of them to finish
  for (int i = 0; i < n; ++i)
    threads[i].join();
}

显然，您不希望在生产代码中使用全局变量；这只是一个演示解决方案。

Answer 3

停止编写代码，画出你需要做的事情和你需要做的事情。

您需要一个队列来保存作业，需要一个互斥锁来保护队列，这样线程就不会同时访问它，并且需要N个线程。

每个线程函数都是一个循环，

1. 获取互斥对象，
2. 从队列中获取作业，
3. 释放互斥对象，
4. 处理任务.

在本例中，当第2步中队列中没有作业时，我会通过退出循环和线程来保持简单。在生产过程中，您将拥有线程块并等待队列，因此它仍然可以用于服务稍后添加的作业。

将其封装在一个类中，其中一个函数允许您向队列中添加作业，一个函数可以启动N个线程，还有一个函数可以连接所有正在运行的线程。

main定义了类的一个实例，在作业中提供提要，启动线程池，然后阻塞连接，直到每个人都完成为止。

一旦你把设计变成了你有很高自信的东西，就可以做你需要做的事情了，然后你就开始写代码。在没有计划的情况下编写代码，特别是多线程代码，您就需要进行大量的调试和重写，而这些调试和重写通常会大大超出设计时间。