C++之生产者和消费者模型分析（条件变量）

2023-05-16

先看一下生产者消费者模型

概述：

生产者把需要处理的数据放到缓存队列中并向消费者发出信号，然后消费者把数据拿出来处理，这里生产者可以是单线程或者多线程，而消费者一般是多线程，消费者线程集合也称线程池。

下面再举一个生活中的生产者和消费者的例子

例如，在平台接单送外卖，生产者是广大人民，而外卖小哥就是消费者，人发出订单，外卖小哥接单，当没有订单时，外卖小哥就进入等待状态。

条件变量

条件变量是一种线程同步机制。当条件不满足时，相关线程被一直阻塞，直到某种条件出现，这些线程才会被唤醒。
C++11的条件变量提供了两个类：
condition_variable：只支持与普通mutex搭配，效率更高。
condition_variable_any：是一种通用的条件变量，可以与任意mutex搭配（包括用户自定义的锁类型）。

成员函数

下面看一个例子，三个线程处理学生，

#include <iostream>
#include <string>
#include <thread>       // 线程类头文件。
#include <mutex>       // 互斥锁类的头文件。
#include <deque>       // deque容器的头文件。
#include <queue>      // queue容器的头文件。
#include <condition_variable>  // 条件变量的头文件。
using namespace std;
class AA
{
    mutex m_mutex;        // 互斥锁。
    condition_variable m_cond;    // 条件变量。
    queue<string, deque<string>> m_q;   // 缓存队列，底层容器用deque。
public:
    void incache(int num)   // 生产数据，num指定数据的个数。
    {
        lock_guard<mutex> lock(m_mutex);   // 申请加锁。
        for (int ii = 0; ii < num; ii++)
        {
            static int bh = 1;    // 同学编号。
            string message = to_string(bh++) + "号同学";    // 拼接出一个数据。
            m_q.push(message);   // 把生产出来的数据入队。
        }
        //给消费者发出信号
        m_cond.notify_one();   // 唤醒一个被当前条件变量阻塞的线程。
        //m_cond.notify_all();    // 唤醒全部被当前条件变量阻塞的线程。
    }
    void outcache()       // 消费者线程任务函数。
    {
        while (true)
        {
            // 把互斥锁转换成unique_lock<mutex>，并申请加锁。
            unique_lock<mutex> lock(m_mutex);   //构造函数的参数时普通互斥锁

            while (m_q.empty())    // 如果队列空，进入循环，否则直接处理数据。必须用循环，不能用if
            {
                m_cond.wait(lock);  // 等待生产者的唤醒信号。
            }

            // 数据元素出队。
            string message = m_q.front();  m_q.pop();
            cout << "线程：" << this_thread::get_id() << "，" << message << endl;
            lock.unlock();

            // 处理出队的数据（把数据消费掉）。
            this_thread::sleep_for(chrono::milliseconds(1));   // 假设处理数据需要1毫秒。
        }
    }
};

int main()
{
    AA aa;

    thread t1(&AA::outcache, &aa);     // 创建消费者线程t1。
    thread t2(&AA::outcache, &aa);     // 创建消费者线程t2。
    thread t3(&AA::outcache, &aa);     // 创建消费者线程t3。

    this_thread::sleep_for(chrono::seconds(2));    // 休眠2秒。
    aa.incache(3);      // 生产3个数据。

    this_thread::sleep_for(chrono::seconds(3));    // 休眠3秒。
    aa.incache(5);      // 生产5个数据。

    t1.join();   // 回收子线程的资源。
    t2.join();
    t3.join();
}

我们发现第一次生产者生产了3个对象，但是都由一个线程处理了，第二次生产了5个对象，也是由同一个线程处理了。这是为什么呢？我们发现在第25行信号之通知了一个等待的线程，所以问题解开了，开始申请的3个线程进入等待状态后，程序每次只申请了一个，所以我们只需要改成申请全部就OK了。

即把25行注释掉，把第26行注释打开

此时我们为了研究wait函数的作用，对代码进行如下改动

#include <iostream>
#include <string>
#include <thread>       // 线程类头文件。
#include <mutex>       // 互斥锁类的头文件。
#include <deque>       // deque容器的头文件。
#include <queue>      // queue容器的头文件。
#include <condition_variable>  // 条件变量的头文件。
using namespace std;
class AA
{
    mutex m_mutex;        // 互斥锁。
    condition_variable m_cond;    // 条件变量。
    queue<string, deque<string>> m_q;   // 缓存队列，底层容器用deque。
public:
    void incache(int num)   // 生产数据，num指定数据的个数。
    {
        lock_guard<mutex> lock(m_mutex);   // 申请加锁。
        for (int ii = 0; ii < num; ii++)
        {
            static int bh = 1;    // 同学编号。
            string message = to_string(bh++) + "号同学";    // 拼接出一个数据。
            m_q.push(message);   // 把生产出来的数据入队。
        }
        //m_cond.notify_one();   // 唤醒一个被当前条件变量阻塞的线程。
        m_cond.notify_all();    // 唤醒全部被当前条件变量阻塞的线程。
    }

    void outcache()       // 消费者线程任务函数。
    {
        while (true)
        {
            // 把互斥锁转换成unique_lock<mutex>，并申请加锁。
            cout << "线程：" << this_thread::get_id() << "，" << "申请加锁" << endl;
            unique_lock<mutex> lock(m_mutex);   //构造函数的参数时普通互斥锁
            cout << "线程：" << this_thread::get_id() << "，" << "加锁成功" << endl;

            //this_thread::sleep_for(chrono::hours(1));   //让线程休眠一小时

             // 条件变量虚假唤醒：消费者线程被唤醒后，缓存队列中没有数据。
            while (m_q.empty())    // 如果队列空，进入循环，否则直接处理数据。必须用循环，不能用if
            {
                m_cond.wait(lock);  // 等待生产者的唤醒信号。
            }//1，把互斥锁解锁  2，阻塞，等待被唤醒 3，给互斥锁加锁 

            // 数据元素出队。
            string message = m_q.front();  m_q.pop();
            cout << "线程：" << this_thread::get_id() << "，" << message << endl;
            lock.unlock();

            // 处理出队的数据（把数据消费掉）。
            this_thread::sleep_for(chrono::milliseconds(1));   // 假设处理数据需要1毫秒。
        }
    }
};

int main()
{
    AA aa;

    thread t1(&AA::outcache, &aa);     // 创建消费者线程t1。
    thread t2(&AA::outcache, &aa);     // 创建消费者线程t2。
    thread t3(&AA::outcache, &aa);     // 创建消费者线程t3。


    this_thread::sleep_for(chrono::seconds(2));    // 休眠2秒。
    //aa.incache(3);      // 生产3个数据。
    //aa.incache(2);

    this_thread::sleep_for(chrono::seconds(3));    // 休眠3秒。
    //aa.incache(5);      // 生产5个数据。

    t1.join();   // 回收子线程的资源。
    t2.join();
    t3.join();
}

我们发现每个都加锁成功了，可是事实真的如此吗？？

我们打开第37行代码，发现只有一个线程加锁成功，其他的线程都卡在申请加锁那步，所以，wait函数不只是等待生产者信号那么简单，它还有一个功能是把互斥锁解开了，所以后面的线程才能加锁成功。

条件变量的wait()函数

作用

1. 把互斥锁解锁
2. 阻塞，等待被唤醒
3. 给互斥锁加锁

不知道有没有注意到

unique_lock<mutex> lock(m_mutex);   //构造函数的参数时普通互斥锁

为什么要转换为unique_lock锁

unique_lock类

template <class Mutex> class unique_lock是模板类，模板参数为互斥锁类型。
unique_lock和lock_guard都是管理锁的辅助类，都是RAII风格（在构造时获得锁，在析构时释放锁）。它们的区别在于：为了配合condition_variable，unique_lock还有lock()和unlock()成员函数。

条件变量虚假唤醒问题

消费者线程被唤醒后，缓存队列没有数据

例如在main函数第一次生产2个学生，这是由三个线程，所以有一个线程一定是被虚假唤醒了的。

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

之生产者和消费者模型分析

条件变量