Linux生产者消费者模型

Posted 2022-08-04 Ustinian%

文章目录

• • • • Linux生产者消费者模型
      • • 生产者消费者的概念
        • 生产者消费者模型的特点
        • 生产者消费者模型的优点
      • 基于BlockingQueue的生产者消费者模型
      • • C++queue模拟阻塞队列实现生产者消费者模型

Linux生产者消费者模型

生产者消费者的概念

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。

生产者和消费者彼此之间不直接通讯，而是通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

生产者消费者模型的特点

生产者消费者模型是多线程同步与互斥的一个经典场景，我们生活中也有许多的消费者与生产者模型，下面我们来举一个生活中的生产者消费者模型的例子——超市

我想可能会有很多同学会认为这里的生产者是超市，但仔细想一想其实并不是哦，我们这里的超市并不生产东西，他只是一个交易场所，真正的生产者是超市背后的供应商。

大家再来想一个问题：为什么生活中要有超市呢？

1. 超市可以收集需求，减少交易成本，从而提高效率
2. 将生产环节与消费环节进行了解耦

下面我们再来分析一下生产者、消费者之间有什么关系？

• 生产者——生产者：竞争，互斥关系
• 消费者——消费者：竞争，互斥关系
• 生产者——消费者：互斥关系与同步关系

可能大家有点不理解这里为什么消费者与消费者之间是互斥关系，我们在生活中去超市里面买东西的时候根本不存在这个互斥关系啊，为什么你又说消费者与消费者之间是互斥关系呢？

我们生活中不存在互斥关系，那是因为超市里面的资源充足，假如说现在是世界末日，超市里面现在就只有一一瓶水，然后你和另外一个人正好要去买这瓶水，这是不是就是互斥关系了呢？所以呀，其实消费者与消费者之间是存在互斥关系的。

大家可能对于生产者与消费者之间存在同步关系也有点疑惑，下面我来给大家举一个例子帮大家理解

假如说我最近想去买个苹果13promax，然后我跑到手机店去问售货员小姐姐，店里面有没有苹果13promax，售货员小姐姐和我说不好意思哈，店里面暂时没货，然后我隔了一天我又去问，隔了一天又去问，得到的结果仍然是不好意思哈，店里面暂时没货。这种效率是非常低下的。因此这个我们可以和小姐姐说：售货员小姐姐啊，这是我的电话，要是店里来货了你就马上给我打电话可以嘛？小姐姐说：没问题。于是隔了几天当店里面来货的时候，小姐姐就给你打电话告诉你店里面有货了，于是你现在就可以直接去手机店里面买苹果13promax。如此一来通过同步的方式便极大的提高了效率。

了解了生产者消费者模型之后，下面我来教大家一个方法让大家快速记住它。

321原则（便于记忆）

• 3种关系：生产者与生产者（互斥关系）、消费者与消费者（互斥关系）、生产者和消费者（互斥关系、同步关系）
• 2种角色：生产者和消费者
• 1个交易场所：比如说上面的超市，通常指的是内存中的一段缓冲区。

生产者消费者模型的优点

• 解耦
• 支持并发
• 支持忙闲不均

基于BlockingQueue的生产者消费者模型

在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。

与普通队列的区别是：

• 当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素
• 当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出

C++queue模拟阻塞队列实现生产者消费者模型

为了便于理解，下面我们以单生产者、单消费者为例进行实现。

下面我们使用C++STL中的queue来模拟实现阻塞队列

#pragma once

#include<iostream>
#include<unistd.h>
#include<queue>
#include<pthread.h>

namespace ns_blockqueue
    const int default_cap = 5;
    template<class T>
    class BlockQueue
    
      private:
        std::queue<T>bq_;//我们的阻塞队列
        int cap_;//队列的元素上限
        pthread_mutex_t mtx_;//保护临界资源的锁
        //1. 当生产满了的时候，就应该不要生产了(不要竞争锁了),而应该让消费者来消费
        //2. 当消费空类，就不应该消费(不要竞争锁了),应该让生产者来进行生产
        pthread_cond_t is_full_;//bq满的，消费者在该条件变量下等待
        pthread_cond_t is_empty_;//bq空的，生产者在该条件变量下等待
      public:
        BlockQueue(int cap = default_cap)
            :cap_(cap)
            
                pthread_mutex_init(&mtx_,nullptr);
                pthread_cond_init(&is_full_,nullptr);
                pthread_cond_init(&is_empty_,nullptr);
            
        ~BlockQueue()
        
            pthread_mutex_destroy(&mtx_);
            pthread_cond_destroy(&is_full_);
            pthread_cond_destroy(&is_empty_);
        
      private:
        bool isFull()
        
            return bq_.size()==cap_;
        
        
        bool isEmpty()
        
            return bq_.size()==0;
        
        
        void LockQueue()
        
            pthread_mutex_lock(&mtx_);
        
        
        void UnLockQueue()
        
            pthread_mutex_unlock(&mtx_);
        
        
        void ProductorWait()
        
            //pthread_cond_wait
            //1. 调用的时候，会首先自动释放mtx_,然后再挂起自己
            //2. 返回的时候，会首先自动竞争锁，获取到锁之后，才能返回!
            pthread_cond_wait(&is_empty_,&mtx_);
        
        
        void ConsumerWait()
        
            pthread_cond_wait(&is_full_);
        
        
        void WakeupConsumer()
        
            pthread_cond_signal(&is_full_);
        
        
        void WakeupProductor()
        
            pthread_cond_signal(&is_empty_);
        
      public:
        //const&:输入型参数
        //*:输出型参数
        //&:输入输出型参数
        void Push(const int& in)
        
            LockQueue();
            //临界区
            while(IsFull())
            
                //等待的，把线程挂起，我们当前是持有锁的
                ProductorWait();
            
            //向队列种放数据，生产函数
            bq_.push(in);
            
            WakeupConsumer();
            UnLockQueue();
        
        
        void Pop(T* out)
        
            LockQueue();
            while(IsEmpty())
            
                //无法消费
                ConsumerWait();
            
            //向队列中拿数据，消费函数
            *out = bq_.front();
            
            WakeupProductor();
            UnLockQueue();
        
    ;

注意：

• 我们这里实现的是单生产者、单消费者的生产者消费者模型，因此我们不需要维护生产者和生产者之间的关系，也不需要维护消费者和消费者之间的关系，我们只需要维护生产者和消费者之间的同步与互斥关系即可。
• 我们这个阻塞队列存储的数据模板化，这样以后使用起来就更方便。
• 阻塞队列是会被生产者和消费者同时访问的临界资源，因此我们需要用一把互斥锁将其保护起来。
• 生产者如果想向阻塞队列里面放数据，前提是阻塞队列里面还有空间，如果阻塞队列已经满了，此时我们的生产者就应该挂起等待，等阻塞队列里面有空间了再将其唤醒。
• 消费者如果想从阻塞队列里面拿数据，前提是阻塞队列里面还有数据，如果阻塞队列已经为空，此时我们的消费者应该挂起等待，等阻塞队列里面有数据了再将其唤醒。
• 不论是生产者还是消费者，它们都是先申请到锁进入临界区后再判断是否满足生产或消费条件的，如果对应条件不满足，那么对应线程就会被挂起。但此时该线程是拿着锁的，为了避免死锁问题，在调用pthread_cond_wait函数时就需要传入当前线程手中的互斥锁，此时当该线程被挂起时就会自动释放手中的互斥锁，而当该线程被唤醒时又会自动获取到该互斥锁。

在主函数中我们创建一个生产者线程和一个消费者线程，让生产者线程不断生产数据，消费者线程不断消费数据。

#include"BlockQueue.hpp"
#include<time.h>
#include<cstdlib>

using namespace ns_blockqueue;

void* Consumer(void* args)

    BlockQueu<int>* bq = (BlockQueue<int>*)args;
    while(true)
    
        sleep(1);
        int data = 0;
        bq->Pop(&data);
        std::cout<<"消费者消费了一个数据: "<<std::endl;
    


void* Productor(void* args)

    BlockQueue<int>* bq = (BlockQueue<int>*)args;
    while(true)
    
        sleep(1);
        int data = rand()%20;
        std::cout<<"生产者生产数据: "<<std::endl;
        bq->Push(data);
    


int main()

    //种一颗随机数种子
    srand((long long)time(nullptr));
    BlockQueue<int>* bq = new BlockQueue<int>();
    
    pthread_t c,p;
    pthread_create(&c,nullptr,Consumer,(void*)bq);
    pthread_create(&p,nullptr,Productor,(void*)bq);
    
    pthread_join(c,nullptr);
    pthread_join(p,nullptr);
    
    return 0;

生产者和消费者步调一致

我们上面主函数的代码里面生产者是每隔一秒生产一个数据，消费者每隔一秒消费一个数据，因此代码运行后我们可以看到生产者和消费者的执行步调是一致的。

生产者生产的快，消费者消费的慢

我们把上面代码稍微改一下，我们让生产者每隔一秒生产一个数据，消费者每隔两秒消费一个数据

void* Consumer(void* args)

    BlockQueu<int>* bq = (BlockQueue<int>*)args;
    while(true)
    
        sleep(2);
        int data = 0;
        bq->Pop(&data);
        std::cout<<"消费者消费了一个数据: "<<std::endl;
    


void* Productor(void* args)

    BlockQueue<int>* bq = (BlockQueue<int>*)args;
    while(true)
    
        sleep(1);
        int data = rand()%20;
        std::cout<<"生产者生产数据: "<<std::endl;
        bq->Push(data);
    

我们可以看到刚开始的时候，生产者生产的快，消费者消费的慢，但是过了一段时间后，阻塞队列被塞满了数据，此时生产者要进行等待同时通知消费者来消费，此时消费者消费一个数据，然后生产者被唤醒进而继续生产数据。此时就会变成生产者每生成一个数据，消费者消费就会消费一个数据，所以后面我们就看到了生产者和消费者步调又一致了。

生产者生产的慢，消费者消费的快

我们把上面代码再来改一下，我们让生产者每隔两秒生产一个数据，消费者每隔一秒消费一个数据

void* Consumer(void* args)

    BlockQueu<int>* bq = (BlockQueue<int>*)args;
    while(true)
    
        sleep(1);
        int data = 0;
        bq->Pop(&data);
        std::cout<<"消费者消费了一个数据: "<<std::endl;
    


void* Productor(void* args)

    BlockQueue<int>* bq = (BlockQueue<int>*)args;
    while(true)
    
        sleep(2);
        int data = rand()%20;
        std::cout<<"生产者生产数据: "<<std::endl;
        bq->Push(data);
    

因为这一次我们生产者每隔两秒生产一个数据，消费者每隔一秒消费一个数据，由于生产者生产的慢，此时阻塞队列里面没数据，所以消费者就需要进行等待，直达到有数据了才可以进行消费，因此消费者它的步调会随着生产者走。

满足某一条件时再唤醒对应的生产者或消费者

我们可以当阻塞队列当中存储的数据大于队列容量的一半时，再唤醒消费者线程进行消费；当阻塞队列当中存储的数据小于队列容器的一半时，再唤醒生产者线程进行生产。

      void Push(const int& in)
        
            LockQueue();
            //临界区
            while(IsFull())
            
                //等待的，把线程挂起，我们当前是持有锁的
                ProductorWait();
            
            //向队列种放数据，生产函数
            bq_.push(in);
            if(bq_.size()>=cap_/2)
            
                WakeupConsumer();   
            
            UnLockQueue();
        
        
        void Pop(T* out)
        
            LockQueue();
            while(IsEmpty())
            
                //无法消费
                ConsumerWait();
            
            //向队列中拿数据，消费函数
            *out = bq_.front();
            if(bq_.size()<=cap_/2)
            
                WakeupProductor();
            
            UnLockQueue();
        

这里我们让生产者一直生产数据，消费者每隔一秒消费一个数据，下面我们来运行一下代码，看看会有什么结果

可以看到这一次和上面的任何一次都不一样，我们这一次只有在队列的数据小于等于cap的一半时才会通知生产者进行生产数据，只有在队列的数据大于等于cap的一半时才会通知消费者进行消费数据。

基于计算任务的生产者消费者模型

我们的生产者消费者模型可不是只能像上面那样生产者生成一个数据，消费者消费一个数据仅仅打印一个数字而已。

我们还可以自己定义一个Task的类，然后实现一个基于计算任务的生产者消费者模型，生产者生产任务，消费者去处理这个任务然后计算出答案。

#pragma once
#include<iostream>

class Task

  public:
    Task(int x = 0,int y = 0,char op = ' ')
        :_x(x)
            ,_y(y)
            ,_op(op)
        
    ~Task()
    
    
    void Run()
    
        int result = 0;
        switch(_op)
        
            case '+':
                result = _x+_y;
                break;
            case '-':
                result = _x-_y;
                break;
            case '*':
                result = _x*_y;
                break;
            case '/':
                if(_y==0)
                
                    std::cout<<"warning: div zero"<<std::endl;
                
                else
                
                    result = _x/_y;
                
                break;
            case '%':
                if(_y==0)
                
                    std::cout<<"warning: mod zero!"<<std::endl;
                
                else
                
                    result = _x%_y;
                
                break;
            default:
                std::cerr<<"operator error"<<std::endl;
                break;
        
        std::cout<<_x<<_op<<_y<<"="<<result<<std::endl;
    
  private:
    int _x;
    int _y;
    char _op;
;

此时生产者放入阻塞队列的数据就是一个Task对象，而消费者从阻塞队列拿到Task对象后，就可以用该对象调用Run成员函数进行数据处理并计算出结果。

void* Productor(void* arg)

	BlockQueue<Task>* bq = (BlockQueue<Task>*)arg;
	const char* msg = "+-*/%";
	//生产者不断进行生产
	while (true)
		int x = rand() % 100+1;
		int y = rand() % 100+1;
		char op = arr[x % 5];
		Task t(x, y, op);
		bq->Push(t); //生产数据
	

void* Consumer(void* arg)

	BlockQueue<Task>* bq = (BlockQueue<Task>*)arg;
	//消费者不断进行消费
	while (true)
		sleep(1);
		Task t;
		bq->Pop(&t); //消费数据
		t.Run(); //处理数据获得结果
	

下面我们来看一下运行结果吧：

可以看到如此以来，我们的生产者不断的往阻塞队列里面放一个又一个的Task对象，然后消费者拿到一个又一个的Task对象之后对他们进行处理从而得到运行结果。