在 cpp 中使用互斥锁的生产者消费者问题

Question

我有一个生产者和两个消费者线程试图访问共享缓冲区。消费者和生产者之间使用互斥锁。消费者应该并行运行。如果缓冲区为空，则消费者休眠，生产者必须唤醒它们。如果缓冲区已满，生产者不做任何事情。下面是我正在处理的代码 sn-ps： 生产者线程：

void *writer(void*)

{
     // Initialising the seed
         srand(time(NULL));
     while(1)
     {
         pthread_mutex_lock(&rallow);
         if (Q.size() < MAX && item < MAX)
         {
            // Getting the random number
            int num = rand() % 10 + 1;
            // Pushing the number into queue
            Q.push(num);
            
            item++;
            cout << "Produced: " << num << " item: "<<item<<endl;
            pthread_cond_broadcast(&dataNotProduced); 
         }
         else if (item == MAX) {
            pthread_mutex_unlock(&rallow);
            continue;
        }
        pthread_mutex_unlock(&rallow);
    }
}

消费者 1：

void *reader1(void*)

{
    while(1)
    {
         pthread_mutex_lock(&mread);

         rc++;

         if(rc==1)
            pthread_mutex_lock(&rallow);

         pthread_mutex_unlock(&mread);


         if (Q.size() > 0) {
            // Get the data from the front of queue
            int data = Q.front();
      
            // Pop the consumed data from queue
            Q.pop();

            item--;
            cout << "B thread consumed: " << data <<endl;


            pthread_cond_signal(&dataNotConsumed);
        }
        else
        {
            cout << "B is in wait.." << endl;
            pthread_cond_wait(&dataNotProduced, &rallow);
            cout<<"B woke up"<<endl;
        }

         pthread_mutex_lock(&mread);

         rc--;

         if(rc==0)
            pthread_mutex_unlock(&rallow);

         pthread_mutex_unlock(&mread);
        sleep(1);

    } 
}

消费者2：

void *reader2(void*)

{
    while(1)
    {
         pthread_mutex_lock(&mread);

         rc++;

         if(rc==1)
            pthread_mutex_lock(&rallow);

         pthread_mutex_unlock(&mread);


         if (Q.size() > 0) {
            // Get the data from the front of queue
            int data = Q.front();
  
            // Pop the consumed data from queue
            Q.pop();

            item--;
            cout << "C thread consumed: " << data <<endl;
            pthread_cond_signal(&dataNotConsumed);
        }
        else
        {
            cout << "C is in wait.." << endl;
            pthread_cond_wait(&dataNotProduced, &rallow);
            cout<<"C woke up"<<endl;
        }

         pthread_mutex_lock(&mread);

         rc--;

         if(rc==0)
            pthread_mutex_unlock(&rallow);

         pthread_mutex_unlock(&mread);
        sleep(1);

    }
}

输出看起来像这样：

C is in wait..
B is in wait..
Produced: 8 item: 1
Produced: 4 item: 2
Produced: 2 item: 3
Produced: 4 item: 4
Produced: 2 item: 5
Produced: 8 item: 6
Produced: 5 item: 7
Produced: 2 item: 8
Produced: 10 item: 9
Produced: 3 item: 10
>> Producer is in wait..
B woke up
B thread consumed: 8
B thread consumed: 4
B thread consumed: 2
B thread consumed: 4
B thread consumed: 2
B thread consumed: 8
B thread consumed: 5
B thread consumed: 2
B thread consumed: 10
B thread consumed: 3
B is in wait..
C woke up
C is in wait..
Producer woke up

我怀疑为什么线程 B 和 C 没有显示并行执行。以及为什么生产者一次将值填充到缓冲区 10 中，而不是提供少量，然后消费者消费它，然后再次产生少量。任何线索将不胜感激。

Answer 1

     else if (item == MAX) {
        pthread_mutex_unlock(&rallow);
        cout << ">> Producer is in wait.." << endl;
        pthread_cond_wait(&dataNotConsumed, &rallow);

您解锁互斥锁，然后等待。你不能那样做。这会创建一个窗口，在此窗口中，您等待的事情可能发生在您等待之前。您必须在持有互斥锁的同时调用pthread_cond_wait，以确保您等待的事情不会在您决定等待之后但在您开始等待之前发生。

您的消费者中有另一个巨大的错误。一个线程可以锁定rallow，然后另一个线程可以尝试解锁它。这是不允许的——获取互斥锁的线程必须是释放它的线程。您不需要两个互斥锁 - 只需使用一个保护所有状态的互斥锁即可。

Answer 2

首先，不能保证所有线程都会一直并发运行。如果它们在单核上运行，操作系统会给每个线程几十毫秒的时间片。如果它们在不同的内核上运行，那么在一个调用pthread_cond_broadcast() 的线程和另一个从pthread_cond_wait() 唤醒的线程之间存在延迟。这很容易解释编写器线程能够在另一个线程唤醒之前将 10 个项目推送到队列。

下一个问题是，为什么 B 消耗了所有物品，而 C 什么也得不到？问题是因为这个：

pthread_mutex_lock(&mread);

rc++;

if(rc == 1)
    pthread_mutex_lock(&rallow);

pthread_mutex_unlock(&mread);

考虑线程 B 和 C 一个接一个地执行这个块。两者都可以锁定mread，两者都会增加rc，但只有一个会锁定rallow。接下来发生的事情是不确定的，因为他们都试图访问队列，即使其中一个不会持有锁。

应该没有必要有两个互斥锁。两个消费者线程都应该无条件锁定rallow，检查队列中是否有东西，如果没有则调用pthread_cond_wait()。

由于您使用的是 C++，因此您应该真正使用 C++11 的线程支持，而不是使用 C pthread 函数。您的代码应如下所示：

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

std::mutex rallow;
std::condition_variable dataProduced;
std::condition_variable dataConsumed;

void writer() {
    while(true) {
        // Generate the random number
        int num = rand() % 10 + 1;
        std::cout << "Produced: " << num << "\n";

        // Push it to the queue
        {
            std::lock_guard<std::mutex> lock(rallow);
            dataConsumed.wait(rallow, [](){return Q.size() < MAX;});
            Q.push(num);
        }
    }
}

void reader(int id) {
    while(true) {
        int data;

        // Pop an item from the queue
        {
            std::lock_guard<std::mutex> lock(rallow);
            dataProduced.wait(rallow, [](){return Q.size() > 0;});
            data = Q.front();
            Q.pop();
        }

        // Process the data
        std::cout << "Consumer thread " << id << " consumed: " << data << "\n";
    }
}

您甚至可以创建一个线程安全的队列类来处理互斥锁和条件变量本身，因此生产者和消费者代码将简化为：

void writer() {
    while(true) {
        int num = rand() % 10 + 1;
        std::cout << "Produced: " << num << "\n";
        Q.push(num);
    }
}

void reader(int id) {
    while(true) {
        int data = Q.pop();
        std::cout << "Consumer thread " << id << " consumed: " << data << "\n";
    }
}