生产者消费者模型

🕗 发布于 2024-11-12 22:01 c++ git 开发语言 网络 linux

线程同步

互斥锁(互斥量)
条件变量
生产/消费者模型

一、互斥锁

C++11提供了四种互斥锁：

mutex：互斥锁。
timed_mutex：带超时机制的互斥锁。
recursive_mutex：递归互斥锁。
recursive_timed_mutex：带超时机制的递归互斥锁。

包含头文件：#include <mutex>

1、mutex类

1）加锁lock()

互斥锁有锁定和未锁定两种状态。

如果互斥锁是未锁定状态，调用lock()成员函数的线程会得到互斥锁的所有权，并将其上锁。

如果互斥锁是锁定状态，调用lock()成员函数的线程就会阻塞等待，直到互斥锁变成未锁定状态。

2）解锁unlock()

只有持有锁的线程才能解锁。

lock和unlock至少满足95%的应用场景！

3）尝试加锁try_lock()

如果互斥锁是未锁定状态，则加锁成功，函数返回true。

如果互斥锁是锁定状态，则加锁失败，函数立即返回false。（线程不会阻塞等待）

2、timed_mutex类

增加了两个成员函数：

bool try_lock_for(时间长度);

bool try_lock_until(时间点);

3、recursive_mutex类

递归互斥锁允许同一线程多次获得互斥锁，可以解决同一线程多次加锁造成的死锁问题。

4、lock_guard类

lock_guard是模板类，可以简化互斥锁的使用，也更安全。

lock_guard的定义如下：

template<class Mutex>
class lock_guard
{
  explicit lock_guard(Mutex& mtx);
}

lock_guard在构造函数中加锁，在析构函数中解锁。

lock_guard采用了RAII思想（在类构造函数中分配资源，在析构函数中释放资源，保证资源在离开作用域时自动释放）。

二、条件变量-生产消费者模型

条件变量

当条件不满足时，相关线程被一直阻塞，直到某种条件出现，这些线程才会被唤醒。
为了保护共享资源，条件变量需要和互斥锁结合一起使用
生产/消费者模型(高速缓存队列)

条件变量是一种线程同步机制。当条件不满足时，相关线程被一直阻塞，直到某种条件出现，这些线程才会被唤醒。

C++11的条件变量提供了两个类：

condition_variable：只支持与普通mutex搭配，效率更高。

condition_variable_any：是一种通用的条件变量，可以与任意mutex搭配（包括用户自定义的锁类型）。

包含头文件：<condition_variable>

1、condition_variable类

主要成员函数：

1）condition_variable() 默认构造函数。

2）condition_variable(const condition_variable &)=delete 禁止拷贝。

3）condition_variable& condition_variable::operator=(const condition_variable &)=delete 禁止赋值。

4）notify_one() 通知一个等待的线程。

5）notify_all() 通知全部等待的线程。

6）wait(unique_lock<mutex> lock) 阻塞当前线程，直到通知到达。

7）wait(unique_lock<mutex> lock,Pred pred) 循环的阻塞当前线程，直到通知到达且谓词满足。

8）wait_for(unique_lock<mutex> lock,时间长度)

9）wait_for(unique_lock<mutex> lock,时间长度,Pred pred)

10）wait_until(unique_lock<mutex> lock,时间点)

11）wait_until(unique_lock<mutex> lock,时间点,Pred pred)

重点（wait(mutex)函数）：

wait(mutex)做了三件事：

把互斥锁解锁。
阻塞，等待被唤醒。
被唤醒后，给互斥锁加锁。

2、unique_lock类

template <class Mutex> class unique_lock是模板类，模板参数为互斥锁类型。

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

生产者消费者模型类示例：

#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);  // 1）把互斥锁解开；2）阻塞，等待被唤醒；3）给互斥锁加锁。
            m_cond.wait(lock, [this] { return !m_q.empty(); });

            // 数据元素出队。
            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(2);      // 生产2个数据。

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

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

流程：

程序运行，因为wait把互斥锁解开了，所以三个消费者都能加锁成功，现在wait到了第二步，三个线程都被阻塞在条件变量的wait()函数中，此时互斥锁没有被任何线程占有；
生产者往队列中放完数据后，会发出条件信号；
wait()函数接收到i先弄好之后，不一定立即返回，他还要申请加锁，加锁成功后才会返回；
如果wait()返回了，一定申请到了锁，接下来可以让队列中的数据出队，出对后再解锁。

消费者线程中的while(m_q.empty())循环：

条件变量存在虚假唤醒的情况：消费者线程被唤醒后，缓存队列中没有数据（三个消费者线程，一次生产两个数据，然后notifyall，全部唤醒，肯定有一个线程拿不到数据，被虚假唤醒了）。如果被虚假唤醒了应该继续等待下一次通知，所以用if肯定不行，必须用while。

也可以用wait(unique_lock<mutex> lock,Pred pred) 版本，添加一个谓词：

m_cond.wait(lock, [this] { return !m_q.empty(); });

效果是一样的，本质上这个重载的wait函数中也有个while循环。

原文地址：https://blog.csdn.net/weixin_56520780/article/details/143671309

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：第八章综合案例————优名养生馆
下一篇：Java并发学习总结：ThreadLocal

【C++】构造与析构函数
⽆参构造函数、全缺省构造函数、我们不写构造时编译器默认⽣成的构造函数，都叫做默认构造函数。
阅读更多2024-11-18
#渗透测试#SRC漏洞挖掘#蓝队基础之网络七层杀伤链02
识别Windows典型应用是指通过一定的技术和方法，自动识别和分类Windows操作系统中的各种应用程序。这对于自动化测试、应用管理、性能监控等领域具有重要意义。识别Linux典型应用是指通过一定的技
阅读更多2024-11-18
【网络】什么是交换机？switch
综上所述，交换机是一种重要的网络设备，它通过为网络节点提供独享通路和高效的数据转发机制，提高了网络传输效率和管理便利性。它通过对MAC地址的识别，实现数据帧的转发。功能：交换机能为接入交换机的任意两个
阅读更多2024-11-18
Go开发指南- Goroutine
在java中我们要实现并发编程的时候，通常要自己维护一个线程池，并且需要去包装任务、调度任务和维护上下文切换。这个过程需要消耗大量的精力。Go语言中有一种机制，可以让系统自动把任务分配到CPU上实现并
阅读更多2024-11-18
NX二次开发将刀轨转曲线
获取选择的程序导轨和当前加工坐标系，获取导轨数据点转换到加工坐标系，然后创建样条，创建样条是用UG录制的代码，有更好的方法欢迎交流学习一下。
阅读更多2024-11-18
STM32寄存器结构体详解
2.#define IOMUXC_GPIO1_IO03_GPIO1_IO03 0x020E0068U, 0x5U, 0x00000000U, 0x0U, 0x020E02F4U 的解释。通过宏定义，使
阅读更多2024-11-18
Android 12.0 第三方app授予DeviceOwner权限调用系统reboot,显示隐藏app,锁屏,禁用app等功能系统层部分实现
在12.0的系统rom定制化开发中，在有些客户开发app的功能中，需要系统授予app的DeviceOwner权限，然后app就可以实现重启关机禁用下拉通知栏，显示隐藏app，等功能，首选需要在app和
阅读更多2024-11-18
同构字符串
每个出现的字符都应当映射到另一个字符，同时不改变字符的顺序。不同字符不能映射到同一个字符上，相同字符只能映射到同一个字符上，字符可以映射到自己本身。中的字符可以按某种映射关系替换得到。，那么这两个字符
阅读更多2024-11-18
unity 下拉列表根据内容长度进行下拉列表长度的自适应
unity 下拉列表根据内容长度进行下拉列表长度的自适应
阅读更多2024-11-18
MATLAB实现狼群算法优化柔性车间调度(GWO-fjsp)
MATLAB实现狼群算法优化柔性车间调度(GWO-fjsp)
阅读更多2024-11-18