Linux多线程——基于环形队列的生产消费模型

一时失言乱红尘 2023-01-16 09:53 142阅读 0赞

### Linux多线程——基于环形队列的生产消费模型 ###

*  一、POSIX信号量
 *   *  1.1 什么是POSIX信号量？
     *  1.2 信号量相关操作接口
     *   *  <1> 初始化信号量
         *  <2> 销毁信号量
         *  <3> 等待信号量-P()操作
         *  <4> 发布信号量-V()操作
 *  二、基于环形队列的生产消费者模型
 *   *  2.1 模型原理
     *  2.2 代码实现
     *   *  <1> RingQueue.hpp
         *  <2> test.c
         *  <3> Makefile
         *  <4>实验结果

关于生产消费者模型的介绍，参看本篇博客  
[Linux多线程——生产消费者模型概念和C++模拟实现生产消费者模型][Linux_C]  
这篇博客生产者-消费者的例子是基于queue的,其空间可以动态分配,现在基于固定大小的环形队列重写这个程序（POSIX信号量）

# 一、POSIX信号量 #

POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。 但POSIX可以用于线程间同步。

## 1.1 什么是POSIX信号量？ ##

前面讲过，对于一份临界资源，一次只能有一个线程去访问临界资源，造成临界资源的利用率较低，是否有一种办法能将临界资源划分为若干份，同时让多个线程去访问临界资源，从而提高资源利用率呢？

答案是肯定的，**通过POSIX信号量将临界资源划分为若干个独立的区域，每个区域只能有一个线程访问，而有多少个独立区域就有多少个线程！**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDc2MDIy_size_16_color_FFFFFF_t_70]

从上图我们可以清晰的看出，**POSIX信号量的本质是一个计数器**，其底层通过一个count计数器统计临界区资源

*  当有线程进入临界区资源时，count- -，表示临界区资源-1，该操作称为**P()操作**
 *  当有线程进入临界区资源时，count+ +，表示临界区资源+1，该操作称为**V()操作**

为了保证临界区资源每次只能进入一个线程，因此信号量是原子性的。

## 1.2 信号量相关操作接口 ##

### <1> 初始化信号量 ###

`#include <semaphore.h>`  
`int sem_init(sem_t *sem, int pshared, unsigned int value);`  
参数：

*  pshared:0表示线程间共享，非零表示进程间共享
 *  value：信号量初始值

### <2> 销毁信号量 ###

`int sem_destroy(sem_t *sem);`

### <3> 等待信号量-P()操作 ###

功能：等待信号量，会将信号量的值减1  
`int sem_wait(sem_t *sem); //P()`

### <4> 发布信号量-V()操作 ###

功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1。  
`int sem_post(sem_t *sem);//V()`

# 二、基于环形队列的生产消费者模型 #

## 2.1 模型原理 ##

![在这里插入图片描述][20210506135542101.png]  
前面说过321原则，使用循环队列实现生产消费者模型，生产者只关心格子数量，也就是队列中空格子，消费者只关心数据，也就是队列中数据的容量。

*  生产者执行P操作就是申请空格子，让格子数量-1，执行V操作时就是将数据放入到了格子，归还的是数据
 *  消费者执行P操作就是申请有数据的格子，执行V操作时，数据被取走，归还的是空格子资源
 *  循环队列通过数组实现，当头尾指针相等时，代表队列满或者空。

## 2.2 代码实现 ##

### <1> RingQueue.hpp ###

#pragma once
    #include<iostream>
    #include<vector>
    #include<semaphore.h>
    #include<pthread.h>
    #include<unistd.h>
    #define NUM 10
    
    class RingQueue
    {
        
      private:
        std::vector<int> v; //循环队列
        int _max_cap;
        sem_t sem_blank; //生产者-关心格子
        sem_t sem_data; //消费者-关心数据
    
        int c_index;  //消费者下标
        int p_index;  //消费者下标
    
        void P(sem_t &s)  
        {
        
          sem_wait(&s);
        }
    
        void V(sem_t &s)
        {
        
          sem_post(&s);
        }
    
      public:
        //构造函数
        RingQueue(int cap = NUM)
          :_max_cap(cap)
        {
        
          v.resize(cap);
          sem_init(&sem_blank, 0, _max_cap);
          sem_init(&sem_data, 0, 0);
          c_index = 0;
          p_index = 0;
        }
        
        //取数据
        void Get(int &out)
        {
        
          P(sem_data);
          out = v[c_index];
          c_index++;
          c_index %= _max_cap;
          V(sem_blank);
        }
        
        //放数据
        void Put(const int &in)
        {
        
          P(sem_blank);
          v[p_index] = in;
          p_index++;
          p_index %= _max_cap;
          V(sem_data);
        }
    
        ~RingQueue()
        {
        
          sem_destroy(&sem_blank);
          sem_destroy(&sem_data);
          c_index = p_index = 0;
        }
    };

### <2> test.c ###

#include"RingQueue.hpp"
    
    pthread_mutex_t pro_lock; //生产者组内竞争锁
    pthread_mutex_t con_lock; //消费者组内竞争锁
    
    int count = 0;
    
    void Lock(pthread_mutex_t &lock)
    {
        
      pthread_mutex_lock(&lock);
    }
    
    void UnLock(pthread_mutex_t &unlock)
    {
        
      pthread_mutex_unlock(&unlock);
    }
    
    //消费者取数据
    void *Get(void *arg)
    {
        
      RingQueue *rq = (RingQueue *)arg;
      while(true)
      {
        
        usleep(1);
        int data = 0;
        Lock(con_lock); //组内竞争
        rq->Get(data);
        UnLock(con_lock);
        std::cout << pthread_self() <<"consumer done..." << data << std::endl;
      }
    }
    
    //生产者生产数据
    void *Put(void *arg)
    {
        
      RingQueue *rq = (RingQueue *)arg;
      while(true)
      {
        
        sleep(1);
        Lock(pro_lock); //组内竞争
        rq->Put(count);
        UnLock(pro_lock);
        std::cout << pthread_self() <<"Productor done.." << count << std::endl;
        count++;
        if(count > 10){
        
          count = 0;
        }
      }
    }
    int main()
    {
        
      pthread_t c1, c2, c3, p1, p2, p3;
      RingQueue *rq = new RingQueue();
      //初始化锁
      pthread_mutex_init(&con_lock,nullptr);
      pthread_mutex_init(&pro_lock,nullptr);
      //创建线程
      pthread_create(&c1, nullptr, Get,rq);
      pthread_create(&c2, nullptr, Get,rq);
      pthread_create(&c3, nullptr, Get,rq);
      pthread_create(&p1,nullptr, Put, rq);
      pthread_create(&p2,nullptr, Put, rq);
      pthread_create(&p3,nullptr, Put, rq);
    
      //等待线程
      pthread_join(c1,nullptr);
      pthread_join(c2,nullptr);
      pthread_join(c3,nullptr);
      pthread_join(p1,nullptr);
      pthread_join(p2,nullptr);
      pthread_join(p3,nullptr);
      //销毁锁清空资源
      pthread_mutex_destroy(&con_lock);
      pthread_mutex_destroy(&pro_lock);
      delete rq;
      return 0;
    }

### <3> Makefile ###

test:test.c
    	g++ -std=c++11 -o $@ $^ -lpthread
    .PHONY:clean
    clean:
    	rm test

### <4>实验结果 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDc2MDIy_size_16_color_FFFFFF_t_70 1]

[Linux_C]: https://blog.csdn.net/qq_40076022/article/details/116043050
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDc2MDIy_size_16_color_FFFFFF_t_70]: /images/20221022/a1ddc855deac4614a19d70995c9a2e14.png
[20210506135542101.png]: /images/20221022/200965639a0b44d18dd3e8a660085282.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDc2MDIy_size_16_color_FFFFFF_t_70 1]: /images/20221022/3ae7b0e846e9448692fa1248d4b053e2.png