设置线程数量作为消费者工作

Question

我创建了如下的生产者/消费者代码

class CTest{
public:
    void producer( int i ){
        unique_lock<mutex> l(m);
        q.push(i);
        if( q.size() )
            cnd.notify_all();
    }

    void consumer(int i ){
        unique_lock<mutex> l(m);    
            while( q.empty() ){
                    cnd.wait(l );
                }
            if( q.empty())
                return;
            cout << "IM AWAKE :" << i  << endl;
            int tmp = q.front();
            q.pop();
            l.unlock();
            cout << "Producer got " << tmp << endl;
        }



    void ConsumerInit( int threads ){
        for( int i = 0; i < threads; i++ ){
            thrs.push_back(thread(&CTest::consumer, this ,i));
        }

    }
    void waitForTHreads(){
        for( auto &a : thrs )
            a.join();
    }
    void printQueue(){
        while( ! q.empty()){
            int tmp = q.front();
            q.pop();
            cout << "Queue got " << tmp << endl;
        }
    }
private:
    queue<int> q;
    vector<thread> thrs;
    mutex m;
    condition_variable cnd;
};

和主要的

int main(){
    int x;   
    CTest t;
    int counter = 0;
    while( cin >> x ){
        if( x == 0 ){
            cout << "yay" << endl;;
            break;
        }
        if( x == 1)
            t.producer(counter++);
        if( x == 2 )
            t.ConsumerInit(5);
    }   
    t.waitForTHreads();
    t.printQueue();
    return 0;
}

这段代码是做什么的，当用户输入“1”时，它将向队列添加数字，当用户输入“2”时，会产生5个线程来从队列中检索数据并打印它。但是，当我输入时，我的问题如下 6个数字，由于只产生了5个线程，因此只打印了其中的5个，我想做的是线程从队列中检索数据，打印int，然后再次等待它是否可以打印另一个数据。这样一来，所有 N > 5 个数字都将仅用 5 个线程打印。

我的问题是，如何实现这一目标的标准方法是什么？我读了一些文档，但没有找到/想不出好的解决方案。这样的问题是如何解决的？

当我尝试创建简单的线程池时：

void consumer(int i ){
    while(true){
        {
            unique_lock<mutex> l(m);    
            while( q.empty() ){
                    cnd.wait(l );
                }
            if( q.empty())
                return;
            cout << "IM AWAKE :" << i  << endl;
            int tmp = q.front();
            q.pop();

            cout << "Producer " << i << " got " << tmp << endl;
        }   //consumer(i);
    }
}

并输入N个数字，所有数字都由一个线程处理。 感谢您的帮助！

Answer 1

当前版本的consumer 在退出前只能读取一个值。为了阅读更多内容，它必须循​​环，这会导致您的consumer 的第二个版本有两个问题：

1. 这里的消耗如此之快，以至于队列中的第一个线程可以在其时间片内消耗整个队列（或者无论如何分配 CPU）。插入 yield 或 sleep 以强制操作系统切换任务。
2. 互斥锁未解锁，因此其他线程无法进入。

幸运的是，在您需要线程之前，您不会创建线程，并且它们会在队列为空后终止，因此与 conditional_variable 的整个处理可以结束。

void consumer(int i)
{
    unique_lock<mutex> l(m);
    while (!q.empty())
    {
        int tmp = q.front();
        q.pop();
        cout << i << " got " << tmp << endl;
        // note: In the real world, locking around a cout is gross. cout is slow, 
        // so you want the unlock up one line. But...! This allows multiple threads
        // to write to the consle at the same time and that makes your output 
        // look like it was tossed into a blender, so we'll take the performance hit
        l.unlock(); // let other threads have a turn
        this_thread::yield();
        l.lock(); // lock again so the queue can be safely inspected
    }
}

如果您需要使用线程池方法，事情会变得有点混乱，并且条件变量会返回。

void consumer(int i)
{
    while (true)
    {
        unique_lock<mutex> l(m);
        if (q.empty())
        {
            cnd.wait(l);
        }
        if (!q.empty()) // OK. We got out of the conditional wait, but have 
                        // other threads sucked the queue dry? Better check.
        {
            int tmp = q.front();
            q.pop();
            cout << i << " got " << tmp << endl;
        }
        l.unlock();
        this_thread::yield();
    }
}

atomic<bool> terminated 可能有助于有序关闭 while (true) 不允许。

Answer 2

一般而言，无需深入了解代码细节，就会创建一个线程池并将线程置于等待状态（等待一个或多个事件/信号，或者在您的情况下为 condition_variable cnd;） - 我习惯于工作带有事件，所以我将在下面的文本中使用它，但 condition_variable 应该以类似的方式工作。

当一个任务被添加到队列中时，一个任务事件被设置/触发并且一个或多个线程被唤醒（取决于事件（单/多））。

当线程唤醒时，它会检查（带锁）是否有可用的任务，如果可用，则执行该任务，完成后再次检查（！）是否有更多的任务在等待。 （因为一次添加 8 个任务时，有 5 个线程处于活动状态，所以他们需要在完成第一个任务后检查是否还有更多任务。

如果没有剩余作业，则线程返回等待状态（等待下一个作业或退出事件）。

当退出应用程序时，会为所有线程设置另一个，例如 quit-event（您不能只等待线程完成，因为线程本身正在等待事件执行某些工作）- 或您可以触发相同的事件，并首先设置volatile variable，然后线程应首先检查任何事件以查看它们是否需要退出或执行其他工作。然后你可以等待线程“回家”。

锁应该尽可能短。

至于你的代码：

void producer( int i ){
    unique_lock<mutex> l(m);
    q.push(i);
    if( q.size() )
        cnd.notify_all();
}

这里的锁被持有的时间比需要的时间长（而且可能太长了）。您还只是推送了一个值，因此q 不会为空（无需检查）。由于您只添加了一项（任务），因此应该只唤醒一个线程（所以 notify_one() 在这里应该没问题）。

所以你应该：lock, push, unlock, notify - 你可以将lock 和push 放在括号内，而不是unlock，这将在unique_lock<> 析构函数中触发unlock。

void consumer(int i ){
    unique_lock<mutex> l(m);    
        while( q.empty() ){
                cnd.wait(l );
            }
        if( q.empty())
            return;
        cout << "IM AWAKE :" << i  << endl;
        int tmp = q.front();
        q.pop();
        l.unlock();
        cout << "Producer got " << tmp << endl;
}

这里你应该lock, check queue, pop if there is a task, unlock，如果没有任务，让线程再次进入等待状态，否则使用弹出的值（解锁后），然后再次检查是否还有更多工作要做。通常在数据被锁定时调用cout 不是一个好主意。但是对于一个小测试，你可以逃脱它，特别是因为cout 也需要同步（但同步@ 会更干净987654337@ 独立于您的数据锁）。

void printQueue(){
    while( ! q.empty()){
        int tmp = q.front();
        q.pop();
        cout << "Queue got " << tmp << endl;
    }
}

确保您的数据也被锁定在这里！ （虽然它只是在线程完成后才从 main 调用，但该函数在您的类中，并且数据应该被锁定）。