（C/C++线程）：创建将侦听作业并在需要时并发执行它们的工作线程

Question

假设我们有两个工作者。每名工人的编号为0和1。另外假设我们有工作一直到达，每个工作也有一个标识符0或1，它指定哪个工作人员必须完成这项工作。

我想创建两个最初被锁定的线程，然后当两个工作到达时解锁它们，每个工作完成他们的工作，然后再次锁定它们直到其他工作到达。

我有以下代码：

#include <iostream> 
    #include <thread> 
    #include <mutex> 

    using namespace std; 

    struct job{ 

     thread jobThread; 
     mutex jobMutex; 

    }; 

    job jobs[2]; 


    void executeJob(int worker){ 

     while(true){ 

      jobs[worker].jobMutex.lock(); 

      //do some job 

     } 

    } 

    void initialize(){ 

     int i; 
     for(i=0;i<2;i++){ 
       jobs[i].jobThread = thread(executeJob, i); 
     } 

    } 

    int main(void){ 

     //initialization 
     initialize(); 

     int buffer[2]; 
     int bufferSize = 0; 

     while(true){ 
      //jobs arrive here constantly, 
      //once the buffer becomes full, 
      //we unlock the threads(workers) and they start working 
      bufferSize = 2; 
      if(bufferSize == 2){ 
       for(int i = 0; i<2; i++){ 
        jobs[i].jobMutex.unlock(); 
       } 
      } 
      break; 
    } 

    } 

我开始使用std::thread前几天，我不知道为什么，但Visual Studio中给了我一个错误说abort() has been called。我相信有些东西因为我的无知而丢失，我无法弄清楚是什么。

我希望这段代码实际上

1. 初始化两个线程，然后把它们锁

2. 里面的主要功能解锁两个线程，这两个线程会做他们的工作（在这种情况下没有），然后他们会再次被锁定。

但它给我一个错误，而不是。我究竟做错了什么？

预先感谢您！

Answer 1

为此，您可以使用boost的线程池类。 它效率高，测试良好。开源库，而不是你新写和稳定它。

http://threadpool.sourceforge.net/

main() 
{ 
    pool tp(2); //number of worker threads-currently its 2. 

    // Add some tasks to the pool. 
    tp.schedule(&first_task); 
    tp.schedule(&second_task); 
} 

void first_task() 
{ 
    ... 
} 

void second_task() 
{ 
    ... 
} 

注：

建议您例如： 你并不需要为每个线程单独的互斥对象。单互斥锁对象本身将执行所有线程之间的同步。您正在执行任务函数中锁定一个线程的互斥体，而未解锁另一个线程正在调用带有不同互斥体对象的锁，导致死锁或未定义行为。

也因为你正在调用互斥锁。在while循环中lock（）没有解锁，同一个线程试图用相同的互斥对象无限地锁定自己，导致未定义的行为。

如果您不需要并行执行线程，则可以在executejob函数内部使用一个全局互斥对象来锁定和解锁。

mutex m; 

void executeJob(int worker) 
{ 
    m.lock(); 

    //do some job 

    m.unlock(); 
} 

如果您想执行作业并行使用boost线程池，正如我之前建议的那样。

Answer 2

从cppreference's page on std::mutex::unlock：

互斥锁，必须由成功地锁定它被摧毁之前，所有的线程进行解锁。否则，行为是不确定的。

您有一个线程代表另一个线程解锁互斥体的方法不正确。

您尝试的行为通常会使用std::condition_variable完成。有些例子可以查看成员函数的链接。

Answer 3

一般而言，您可以编写类似于以下的算法。它适用于pthreads。我相信它也可以用于C++线程。

1. 创建线程并让它们等待condition variable，例如， work_exists。
2. 工作到达时，通知所有等待该条件变量的线程。然后在主线程开始等待另一条件变量work_done
3. 在收到work_exists通知，工作线程醒来，并从jobs[worker]抓住他们分配工作，他们执行它，他们在work_done变量发送通知，然后回去等待work_exists条件变量
4. 当主线程收到work_done通知时，它会检查是否所有线程都已完成。如果不是，它会一直等待，直到来自最后完成线程的通知到达。