C++生产者消费者设计模式+多线程基础扫盲

怼烎@ 2023-02-17 02:24 91阅读 0赞

## 基础扫盲 ##

为什么需要这种模式，这种模式的好处是什么，请转：  
[生产者消费者模式原理，优势][Link 1]

## 1.mutex互斥量、锁 ##

**mutex**互斥量是一个类，这个类有有一个`lock()`方法，和一个`unlock()`方法。如果第一次运行了`lock()`这个方法，而没有运行`unlock()`这个方法，第二次再运行`lock()`这个方法时，程序就会卡停在这里，只有当运行了`unlock()`这个方法运行后，第二个`lock()`方法才会运行通过。就是运用这种“锁”的机制就可以保证两段代码独立运行。

> `lock()和unlock()`必须同时成对出现，不可以多写，也不可以少写，要不认会出现不知名的错误。

std::lock_guard<mutex> lock(mutex mtx);

这个类模板的作用就是替代`lock()和unlock() ,lock()和unlock()`必须同时出现，而`lock_guard()`只需要出现一个就行。

unique_lock<mutex> lock(mutex mtx)

`unique_lock`中的`unique`表示独占所有权。  
`unique_lock`独占的是mutex对象，就是对mutex锁的独占。  
等待程序生命周期借宿自动解锁，或者 `lock_unlock`.

## 2.std::condition\_variable 类介绍（条件变量） ##

condition_variable repo_not_full;//条件变量，只是产品缓冲区不满
    condition_variable repo_not_empty;//指示产品缓冲区不为空

当 `std::condition_variable`对象的某个`wait` 函数被调用的时候，它使用 `std::unique_lock(通过 std::mutex)` 来锁住当前线程。当前线程会一直被阻塞，直到另外一个线程在相同的 `std::condition_variable` 对象上调用了 `notification` 函数来唤醒当前线程。  
`std::condition_variable` 对象通常使用 `std::unique_lock` 来等待.  
**std::condition\_variable::wait() 介绍**  
std::condition\_variable提供了两种 wait() 函数。当前线程调用 wait() 后将被阻塞(此时当前线程应该获得了锁（mutex），不妨设获得锁 lck)，直到另外某个线程调用 notify\_\* 唤醒了当前线程。  
**std::condition\_variable::notify\_all() 介绍**  
唤醒所有的等待(wait)线程。如果当前没有等待线程，则该函数什么也不做。  
**std::condition\_variable::notify\_one() 介绍**  
唤醒某个等待(wait)线程。如果当前没有等待线程，则该函数什么也不做，如果同时存在多个等待线程，则唤醒某个线程是不确定的(unspecified)。

## 3.生产者消费者模型伪代码（单生产单消费） ##

理解这种模式，学以致用；

typedef struct ItemRepository
    {
        
    	int Item_Buffer[10];//项目存储库
    	int Read_Position;//读取标志位
    	int Write_Positin;//写入标志位
    	condition_variable repo_not_full;//缓冲区不为空条件变量
    	condition_variable repo_not_empty;//不为满，两条件变量用来两个线程之间进行通信使用
    }
    ProduceItem(ItemRepository *ir,int item)//生产者线程
    {
        
    	对存储区加锁lock(Item_Buffer)；
    	判断存储区是否为满，满则repo_not_empty.notifiy_all()通知消费线程，赶紧开始消费; 
    	同时repo_not_full.wait(lock);线程开始阻塞等待缓存不满事件的通知，同时别忘了lock.unlock()解锁缓冲区;
    	不满，则写入数据，
    }
    
    ConsumerItem(ItemRepository *ir)//消费者线程
    {
        
    	判断缓冲区是否为空，如空，则repo_not_full.notify_all();通知生产线程开始生产；
    	同时repo_not_empty.wait(lock);阻塞等待生产线程缓存不空事件的通知。同时lock.unlock();
    	不空，则继续读取数据，
    }
    ProduceTask()//循环生产者任务
    ConsumerTask()//循环消费者任务

## 4.详注源码； ##

**thrad\_Lock.h**

#ifndef THREAD_LOCK_H
    #define THREAD_LOCK_H
    #pragma once;
    #include<thread>
    #include<pthread.h>
    #include<mutex>//互斥锁，多线程访问统一资源时，防止资源被同时访问出错
    #include"myheadfile.h"
    #define  kItemRepositorySize 5
    //#define  kItemToProduce 100
    using namespace std;
    
    //typedef是允许结构体使用别名，如果不加的话，结构体不能有别名，会有重定义的错误。
    typedef struct ItemRepository//项目存储库
    {
        
    	int item_buffer[kItemRepositorySize];//缓冲区,缓冲区是生产者和消费者数据交互的渠道。
    										 //生产者生产的数据存入此数组中，然后消费者从此数组中取出数据。
    	size_t read_position;//消费者者读取产品位置
    	size_t write_position;//生产者写入产品位置
    	mutex mutexBuffer;//缓冲区互斥锁，避免多线程同时操作缓冲区
    	condition_variable repo_not_full;//条件变量，只是产品缓冲区不满
    	condition_variable repo_not_empty;//指示产品缓冲区不为空
    }gItemRepository;
    
    void ProduceItem(ItemRepository* ir, int item);//生产者生产
    int ConsumeItem(ItemRepository* ir);//消费者消费
    void ProducerTask(ItemRepository* ir);//生产者任务
    void ConsumerTask(ItemRepository* ir);//消费者任务
    ItemRepository* InitItemRepository(ItemRepository* ir);//项目初始化
    #endif

**Produce\_Consumer\_Mode.cpp**

#include<iostream>
    #include<stdio.h>
    #include<mutex>//互斥锁，多线程访问统一资源时，防止资源被同时访问出错
    #include"myheadfile.h"
    #include"thread_Lock.h"
    
    mutex mutexCout;// 全局多线程输出锁，避免多线程同时访问输出窗口
    
    void ProduceItem(ItemRepository* ir, int item)//生产者生产
    {
        
    	unique_lock<mutex> lock(ir->mutexBuffer);//缓冲区加锁，此时只能生产者操作缓冲区，消费者等待
    	while(ir->write_position == kItemRepositorySize - 1)//?????判断缓冲区是否已满
    	{
        
    		ir->write_position = 0;//标志位 归0
    		{
           //这个大括号说明加锁的作用域就在这个区域
    			lock_guard<mutex> lock(mutexCout);//输出加锁
    			cout << "缓冲区已满，请等待消费者消费" << endl;
    		}
    		(ir->repo_not_empty).notify_all();//通知消费者缓冲区不为空
    		(ir->repo_not_full).wait(lock);//说明缓冲区已满，生产者阻塞在这里，开始等待消费者线程给的通知消息。	
    		lock.unlock();//解锁
    	}
    	(ir->item_buffer)[ir->write_position] = item;//当缓冲区不满，就向其写入数据
    	ir->write_position++;//标志位++
    	
    }
    
    int ConsumeItem(ItemRepository* ir)//消费者消费
    {
        
    	unique_lock<mutex> lock(ir->mutexBuffer);
    	while(ir->read_position == kItemRepositorySize - 1)//判断缓冲区是否为空
    	{
        
    		ir->read_position = 0;
    		{
        
    			lock_guard<mutex> lock(mutexCout);
    			cout << "缓冲区空，等待生产者生成产品" << endl;
    		}
    		(ir->repo_not_full).notify_all();//通知消费者产品库不为满
    		(ir->repo_not_empty).wait(lock);//锁等待，缓冲区不为空这一事件的发生。
    		lock.unlock();
    	}
    	int data = ir->item_buffer[ir->read_position];
    	(ir->read_position)++;
    	return data;
    }
    
    void ProducerTask(ItemRepository* ir)//生产者任务
    {
        
    	int i = 1;
    	while (1)
    	{
        
    		//this_thread::sleep_for(chrono::seconds(1));
    		ProduceItem(ir, i);
    	    {
        
    		  lock_guard<mutex> lock(mutexCout);
    		  cout << "正在生产第" << i << "个产品" << endl;
    		}
    		i++;      
    	}	
    }
    
    void ConsumerTask(ItemRepository* ir)//消费者任务
    {
        
    	static int cnt = 0;
    	while (1)
    	{
        
    		this_thread::sleep_for(chrono::seconds(1));//
    		int item = ConsumeItem(ir);
    		{
        
    			lock_guard<mutex> lock(mutexCout);
    			cout << "消费者正在消费第" << item << "个产品" << endl;
    		}
    		//if (++cnt == kItemRepositorySize) break;
    	}
    }
    
    ItemRepository* InitItemRepository(ItemRepository* ir)//项目初始化
    {
        
    	ir->read_position = 0;
    	ir->write_position = 0;
    	return ir;
    }

**main.cpp**

#include"myheadfile.h"
    #include"thread_Lock.h"
    using namespace std;
    void main()
    {
        
    	ItemRepository* guoItem = new ItemRepository;
    	InitItemRepository(guoItem);
    	thread producer(ProducerTask,guoItem);//创建生产者线程
    	thread consumer(ConsumerTask,guoItem);//创建消费者线程
    	producer.join();
    	producer.join();
    }

## 5.运行结果： ##

循环执行结果：  
![在这里插入图片描述][20200611203807732.png]

[Link 1]: https://blog.csdn.net/w57w57w57/article/details/11290165?utm_medium=distribute.pc_relevant.none-task-blog-OPENSEARCH-3.nonecase&depth_1-utm_source=distribute.pc_relevant.none-task-blog-OPENSEARCH-3.nonecase
[20200611203807732.png]: https://img-blog.csdnimg.cn/20200611203807732.png