【C++多线程】C++11互斥锁和条件变量实现生产者消费者模型

news2024/11/25 21:26:04

先看几个问题，第三个问题可以先看代码然后再理解

Q1：临界区在哪

A1: 队列中元素在「生产者生产（push）」和「消费者消费（pop）」时就是临界区

Q2：同步操作在哪

A2: 很显然，队列只有在存在元素的前提下消费者才能消费，队列中元素满（假设有容量线程）时只有生产者是不能生产的，因此

生产者队列满了就应该通知消费者消费
消费者线程发现队列为空就需要通知生产者线程先生产物品

Q3：为什么消费者在 cv.wait(lck) 的条件是 while 而不是 if

A3: cv.wait(lock) 本质上是阻塞的，它会一直等待，直到接收到 notify 或 notify_all 的通知。但是，在某些情况下，虽然没有收到通知，但 cv.wait(lock) 可能会返回。这种情况被称为虚假唤醒（spurious wakeup）。

虚假唤醒是因为条件变量的实现方式，以及底层操作系统和硬件的影响。条件变量的实现通常依赖于底层的线程库和操作系统，它们可能在某些情况下引发虚假唤醒，这是一种难以避免的情况。

因此，为了编写健壮的多线程代码，通常建议使用循环来检查条件，就像在前面的示例中使用的 while (dataQueue.empty()) 一样。这样，即使发生虚假唤醒，线程也会再次检查条件，确保只有在条件满足时才继续执行。

总之，虽然 cv.wait(lock) 通常是阻塞的，但在多线程环境中，考虑到虚假唤醒是一种良好的编程实践，以确保正确性和可靠性。

总结来说就是 cv.wait(lock)可能会在某些情况下（如操作系统调度或硬件中断等）自行返回。因此，为了保险起见，应该在一个循环中检查条件，如示例中使用的 while (dataQueue.empty())，以防止虚假唤醒导致的错误行为

完整代码如下：

#include <iostream>
#include <queue>
#include <thread>
#include <memory>
#include <condition_variable>

using namespace std;

std::mutex mtx; // 互斥锁实现线程之间的互斥操作
std::condition_variable cv; // 条件变量实现线程之间通信操作

class Queue {
  public:
    void put(int val) {
        unique_lock<std::mutex> lck(mtx);
        if (q.size() == 10) {
            // 生产者队列满了就应该通知消费者消费
            // 生产者线程应该进入 #1 等待状态，并且 #2 把 mtx 释放掉
            cv.wait(lck);
        }
        q.push(val);

        /**
         * @brief 通知所有线程 notify_all()，通知一个线程 notify_one()
         * 通知其他所有的线程，我生产了一个物品你们赶紧地去消费
         * 其他线程得到了该通知就会从等待状态 => 阻塞状态 => 获取互斥锁之后才能继续执行
         */
        cv.notify_all();
        cout << "生产者 生产: " << val << "号物品\n";
    }

    void get() {
        unique_lock<std::mutex> lck(mtx);
        //!NOTE: 这里写成 while 是为了防止 cv.wait 被虚假唤醒
        while (q.empty()) {
            // 消费者线程发现队列为空就需要通知生产者线程先生产物品
            // #1 进入等待状态 #2 释放 mtx
            cv.wait(lck);
        }
        int val = q.front();
        q.pop();

        // 通知其他所有的线程，我消费了一个物品你们赶紧地去生产
        cv.notify_all();
        cout << "消费者 消费: " << val << "号物品\n";
    }

  private:
    queue<int> q;
};

void producer(Queue *q)  // 生产者线程
{
    for (int i = 1; i <= 10; ++i) {
        q->put(i);
        std::this_thread::sleep_for(std::chrono::microseconds(100));
    }
}

void consumer(Queue *q)  // 消费者线程
{
    for (int i = 1; i <= 10; ++i) {
        q->get();
        std::this_thread::sleep_for(std::chrono::microseconds(100));
    }
}

int main() {
    Queue q;

    std::thread t1(producer, &q);
    std::thread t2(consumer, &q);

    t1.join();
    t2.join();

    return 0;
}