＜生产者、消费者问题＞——《Linux》

 目录

1. 生产者消费者模型

1.1 为何要使用生产者消费者模型

1.2 生产者消费者模型优点

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

2.1 BlockingQueue

2.2 C++ queue模拟阻塞队列的生产消费模型

3.POSIX信号量

4.基于环形队列的生产消费模型

后记：●由于作者水平有限，文章难免存在谬误之处，敬请读者斧正，俚语成篇，恳望指教！

                                                                           ——By 作者：新晓·故知

1. 生产者消费者模型

1.1 为何要使用生产者消费者模型

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

1.2 生产者消费者模型优点

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

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

2.1 BlockingQueue

在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)

2.2 C++ queue模拟阻塞队列的生产消费模型

代码：

为了便于理解，我们以单生产者，单消费者，来进行讲解。

     

      

       

        

       
       

        

         #include <iostream>
        
       
      

       

        

       
       

        

         #include <queue>
        
       
      

       

        

       
       

        

         #include <stdlib.h>
        
       
      

       

        

       
       

        

         #include <pthread.h>
        
       
      

       

        

       
       

        

         #define NUM 8
        
       
      

       

        

       
       

        

         class 
         BlockQueue
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        

         private:
        
       
      

       

        

       
       

        

             std::queue<
         int> q;
        
       
      

       

        

       
       

        
    
         int cap;
        
       
      

       

        

       
       

        
    
         pthread_mutex_t lock;
        
       
      

       

        

       
       

        
    
         pthread_cond_t full;
        
       
      

       

        

       
       

        
    
         pthread_cond_t empty;
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         private:
        
       
      

       

        

       
       

        
    
         void LockQueue()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_mutex_lock(&lock);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void UnLockQueue()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_mutex_unlock(&lock);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void ProductWait()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_cond_wait(&full, &lock);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void ConsumeWait()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_cond_wait(&empty, &lock);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void NotifyProduct()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_cond_signal(&full);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void NotifyConsume()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_cond_signal(&empty);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         bool IsEmpty()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         return (q.
         size() == 
         0 ? 
         true : 
         false);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         bool IsFull()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         return (q.
         size() == cap ? 
         true : 
         false);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         public:
        
       
      

       

        

       
       

        
    
         BlockQueue(
         int _cap = NUM) : 
         cap(_cap)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_mutex_init(&lock, 
         NULL);
        
       
      

       

        

       
       

        
        
         pthread_cond_init(&full, 
         NULL);
        
       
      

       

        

       
       

        
        
         pthread_cond_init(&empty, 
         NULL);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void PushData(const int &data)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         LockQueue();
        
       
      

       

        

       
       

        
        
         while (
         IsFull())
        
       
      

       

        

       
       

        

                 {
        
       
      

       

        

       
       

        
            
         NotifyConsume();
        
       
      

       

        

       
       

        

                     std::cout << 
         "queue full, notify consume data, product stop." << std::endl;
        
       
      

       

        

       
       

        
            
         ProductWait();
        
       
      

       

        

       
       

        

                 }
        
       
      

       

        

       
       

        

                 q.
         push(data);
        
       
      

       

        

       
       

        
        
         // NotifyConsume();
        
       
      

       

        

       
       

        
        
         UnLockQueue();
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void PopData(int &data)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         LockQueue();
        
       
      

       

        

       
       

        
        
         while (
         IsEmpty())
        
       
      

       

        

       
       

        

                 {
        
       
      

       

        

       
       

        
            
         NotifyProduct();
        
       
      

       

        

       
       

        

                     std::cout << 
         "queue empty, notify product data, consume stop." << std::endl;
        
       
      

       

        

       
       

        
            
         ConsumeWait();
        
       
      

       

        

       
       

        

                 }
        
       
      

       

        

       
       

        

                 data = q.
         front();
        
       
      

       

        

       
       

        

                 q.
         pop();
        
       
      

       

        

       
       

        
        
         // NotifyProduct();
        
       
      

       

        

       
       

        
        
         UnLockQueue();
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        

             ~
         BlockQueue()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_mutex_destroy(&lock);
        
       
      

       

        

       
       

        
        
         pthread_cond_destroy(&full);
        
       
      

       

        

       
       

        
        
         pthread_cond_destroy(&empty);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        

         };
        
       
      

       

        

       
       

        

         void *consumer(void *arg)
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        

             BlockQueue *bqp = (BlockQueue *)arg;
        
       
      

       

        

       
       

        
    
         int data;
        
       
      

       

        

       
       

        
    
         for (;;)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        

                 bqp->
         PopData(data);
        
       
      

       

        

       
       

        

                 std::cout << 
         "Consume data done : " << data << std::endl;
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        

         }
        
       
      

       

        

       
       

        

         // more faster
        
       
      

       

        

       
       

        

         void *producter(void *arg)
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        

             BlockQueue *bqp = (BlockQueue *)arg;
        
       
      

       

        

       
       

        
    
         srand((
         unsigned 
         long)
         time(
         NULL));
        
       
      

       

        

       
       

        
    
         for (;;)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         int data = 
         rand() % 
         1024;
        
       
      

       

        

       
       

        

                 bqp->
         PushData(data);
        
       
      

       

        

       
       

        

                 std::cout << 
         "Prodoct data done: " << data << std::endl;
        
       
      

       

        

       
       

        
        
         // sleep(1);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        

         }
        
       
      

       

        

       
       

        

         int main()
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        

             BlockQueue bq;
        
       
      

       

        

       
       

        
    
         pthread_t c, p;
        
       
      

       

        

       
       

        
    
         pthread_create(&c, 
         NULL, consumer, (
         void *)&bq);
        
       
      

       

        

       
       

        
    
         pthread_create(&p, 
         NULL, producter, (
         void *)&bq);
        
       
      

       

        

       
       

        
    
         pthread_join(c, 
         NULL);
        
       
      

       

        

       
       

        
    
         pthread_join(p, 
         NULL);
        
       
      

       

        

       
       

        
    
         return 
         0;
        
       
      

       

        

       
       

        

         }
        
       
     
    

     
    

模拟实现：

BlockQueue.hpp:

     

      

       

        

       
       

        

         #pragma once
        
       
      

       

        

       
       

        

         #include <iostream>
        
       
      

       

        

       
       

        

         #include <queue>
        
       
      

       

        

       
       

        

         #include <cstdlib>
        
       
      

       

        

       
       

        

         #include <unistd.h>
        
       
      

       

        

       
       

        

         #include <pthread.h>
        
       
      

       

        

       
       

        

         using 
         namespace std;
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         // 实现新需求: 我只想保存最新的5个任务，如果来了任务，老的任务，我想让他直接被丢弃（自行选择实现）
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         const 
         uint32_t gDefaultCap = 
         5;
        
       
      

       

        

       
       

        

         template <
         class 
         T>
        
       
      

       

        

       
       

        

         class 
         BlockQueue
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        

         public:
        
       
      

       

        

       
       

        
    
         BlockQueue(
         uint32_t cap = gDefaultCap) : 
         cap_(cap)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_mutex_init(&mutex_, 
         nullptr);
        
       
      

       

        

       
       

        
        
         pthread_cond_init(&conCond_, 
         nullptr);
        
       
      

       

        

       
       

        
        
         pthread_cond_init(&proCond_, 
         nullptr);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        

             ~
         BlockQueue()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_mutex_destroy(&mutex_);
        
       
      

       

        

       
       

        
        
         pthread_cond_destroy(&conCond_);
        
       
      

       

        

       
       

        
        
         pthread_cond_destroy(&proCond_);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         public:
        
       
      

       

        

       
       

        
    
         //生产接口
        
       
      

       

        

       
       

        
    
         void push(const T &in) // const &: 纯输入
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         // 加锁
        
       
      

       

        

       
       

        
        
         // 判断->是否适合生产->bq是否为满->程序员视角的条件->1. 满(不生产) 2. 不满(生产)
        
       
      

       

        

       
       

        
        
         // if(满) 不生产,休眠
        
       
      

       

        

       
       

        
        
         // else if(不满) 生产，唤醒消费者
        
       
      

       

        

       
       

        
        
         // 解锁
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        
        
         lockQueue();
        
       
      

       

        

       
       

        
        
         while (
         isFull()) 
         // ifFull就是我们在临界区中设定的条件
        
       
      

       

        

       
       

        

                 {
        
       
      

       

        

       
       

        
            
         // before: 当我等待的时候，会自动释放mutex_
        
       
      

       

        

       
       

        
            
         proBlockWait(); 
         //阻塞等待，等待被唤醒。 被唤醒 ！= 条件被满足(概率虽然很小)，被唤醒 && 条件被满足
        
       
      

       

        

       
       

        
            
         //解决伪唤醒(使用while)
        
       
      

       

        

       
       

        
            
         // after: 当我醒来的时候，我是在临界区里醒来的！！
        
       
      

       

        

       
       

        

                 }
        
       
      

       

        

       
       

        
        
         // 条件满足，可以生产
        
       
      

       

        

       
       

        
        
         pushCore(in); 
         //生产完成
        
       
      

       

        

       
       

        
        
         // wakeupCon(); // 唤醒消费者
        
       
      

       

        

       
       

        
        
         unlockQueue();
        
       
      

       

        

       
       

        
        
         wakeupCon(); 
         // 唤醒消费者
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         //消费接口
        
       
      

       

        

       
       

        
    
         T pop()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         // 加锁
        
       
      

       

        

       
       

        
        
         // 判断->是否适合消费->bq是否为空->程序员视角的条件->1. 空(不消费) 2. 有(消费)
        
       
      

       

        

       
       

        
        
         // if(空) 不消费,休眠
        
       
      

       

        

       
       

        
        
         // else if(有) 消费，唤醒生产者
        
       
      

       

        

       
       

        
        
         // 解锁
        
       
      

       

        

       
       

        
        
         lockQueue();
        
       
      

       

        

       
       

        
        
         while (
         isEmpty())
        
       
      

       

        

       
       

        

                 {
        
       
      

       

        

       
       

        
            
         conBlockwait(); 
         //阻塞等待，等待被唤醒，？
        
       
      

       

        

       
       

        

                 }
        
       
      

       

        

       
       

        
        
         // 条件满足，可以消费
        
       
      

       

        

       
       

        

                 T tmp = 
         popCore();
        
       
      

       

        

       
       

        
        
         unlockQueue();
        
       
      

       

        

       
       

        
        
         wakeupPro(); 
         // 唤醒生产者
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        
        
         return tmp;
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         private:
        
       
      

       

        

       
       

        
    
         void lockQueue()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_mutex_lock(&mutex_);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void unlockQueue()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_mutex_unlock(&mutex_);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         bool isEmpty()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         return bq_.
         empty();
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         bool isFull()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         return bq_.
         size() == cap_;
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void proBlockWait() // 生产者一定是在临界区中的！
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         // 1. 在阻塞线程的时候，会自动释放mutex_锁
        
       
      

       

        

       
       

        
        
         pthread_cond_wait(&proCond_, &mutex_);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        
    
         void conBlockwait() //阻塞等待，等待被唤醒
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         // 1. 在阻塞线程的时候，会自动释放mutex_锁
        
       
      

       

        

       
       

        
        
         pthread_cond_wait(&conCond_, &mutex_);
        
       
      

       

        

       
       

        
        
         // 2. 当阻塞结束，返回的时候，pthread_cond_wait，会自动帮你重新获得mutex_，然后才返回
        
       
      

       

        

       
       

        
        
         // 为什么我们上节课，写的代码，批量退出线程的时候，发现无法退出？ 
        
       
      

       

        

       
       

        
        
         //唤醒时， 调用pthread_cond_wait(&conCond_, &mutex_);，重新去竞争这个锁，多个线程去竞争一个，且那个拥有锁的线程退出没有释放锁，最后导致其他线程阻塞。
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        
    
         void wakeupPro() // 唤醒生产者
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_cond_signal(&proCond_);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void wakeupCon() // 唤醒消费者
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         pthread_cond_signal(&conCond_);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         void pushCore(const T &in)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        

                 bq_.
         push(in); 
         //生产完成
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         T popCore()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        

                 T tmp = bq_.
         front();
        
       
      

       

        

       
       

        

                 bq_.
         pop();
        
       
      

       

        

       
       

        
        
         return tmp;
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         private:
        
       
      

       

        

       
       

        
    
         uint32_t cap_;           
         //容量
        
       
      

       

        

       
       

        

             queue<T> bq_;            
         // blockqueue
        
       
      

       

        

       
       

        
    
         pthread_mutex_t mutex_;  
         //保护阻塞队列的互斥锁
        
       
      

       

        

       
       

        
    
         pthread_cond_t conCond_; 
         // 让消费者等待的条件变量
        
       
      

       

        

       
       

        
    
         pthread_cond_t proCond_; 
         // 让生产者等待的条件变量
        
       
      

       

        

       
       

        

         };
        
       
     
    

     
    

BlockQueueTest.cc: 

     

      

       

        

       
       

        

         #include "Task.hpp"
        
       
      

       

        

       
       

        

         #include "BlockQueue.hpp"
        
       
      

       

        

       
       

        

         #include <ctime>
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         const std::string ops = 
         "+-*/%";
        
       
      

       

        

       
       

        

         // 并发，并不是在临界区中并发(一般)，而是生产前(before blockqueue)，消费后(after blockqueue)对应的并发
        
       
      

       

        

       
       

        

         void *consumer(void *args)
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        

             BlockQueue<Task> *bqp = 
         static_cast<BlockQueue<Task> *>(args);
        
       
      

       

        

       
       

        
    
         while (
         true)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        

                 Task t = bqp->
         pop(); 
         // 消费任务
        
       
      

       

        

       
       

        
        
         int result = 
         t();    
         //处理任务 --- 任务也是要花时间的！
        
       
      

       

        

       
       

        
        
         int one, two;
        
       
      

       

        

       
       

        
        
         char op;
        
       
      

       

        

       
       

        

                 t.
         get(&one, &two, &op);
        
       
      

       

        

       
       

        

                 cout << 
         "consumer[" << 
         pthread_self() << 
         "] " << (
         unsigned 
         long)
         time(
         nullptr) << 
         " 消费了一个任务: " << one << op << two << 
         "=" << result << endl;
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        

         }
        
       
      

       

        

       
       

        

         void *productor(void *args)
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        

             BlockQueue<Task> *bqp = 
         static_cast<BlockQueue<Task> *>(args);
        
       
      

       

        

       
       

        
    
         while (
         true)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         // 1. 制作任务 --- 要不要花时间？？ -- 网络，磁盘，用户
        
       
      

       

        

       
       

        
        
         int one = 
         rand() % 
         50;
        
       
      

       

        

       
       

        
        
         int two = 
         rand() % 
         20;
        
       
      

       

        

       
       

        
        
         char op = ops[
         rand() % ops.
         size()];
        
       
      

       

        

       
       

        
        
         Task t(one, two, op);
        
       
      

       

        

       
       

        
        
         // 2. 生产任务
        
       
      

       

        

       
       

        

                 bqp->
         push(t);
        
       
      

       

        

       
       

        

                 cout << 
         "producter[" << 
         pthread_self() << 
         "] " << (
         unsigned 
         long)
         time(
         nullptr) << 
         " 生产了一个任务: " << one << op << two << 
         "=?" << endl;
        
       
      

       

        

       
       

        
        
         sleep(
         1);
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        

         }
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         int main()
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        
    
         srand((
         unsigned 
         long)
         time(
         nullptr) ^ 
         getpid());
        
       
      

       

        

       
       

        
    
         // 定义一个阻塞队列
        
       
      

       

        

       
       

        
    
         // 创建两个线程，productor, consumer
        
       
      

       

        

       
       

        
    
         // productor ----- consumer
        
       
      

       

        

       
       

        
    
         // BlockQueue<int> bq;
        
       
      

       

        

       
       

        
    
         // bq.push(10);
        
       
      

       

        

       
       

        
    
         // int a = bq.pop();
        
       
      

       

        

       
       

        
    
         // cout << a << endl;
        
       
      

       

        

       
       

        
    
         // 既然可以使用int类型的数据，我们也可以使用自己封装的类型，包括任务
        
       
      

       

        

       
       

        
    
         // BlockQueue<int> bq;
        
       
      

       

        

       
       

        

             BlockQueue<Task> bq;
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        
    
         pthread_t c, p;
        
       
      

       

        

       
       

        
    
         pthread_create(&c, 
         nullptr, consumer, &bq);
        
       
      

       

        

       
       

        
    
         pthread_create(&p, 
         nullptr, productor, &bq);
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        
    
         pthread_join(c, 
         nullptr);
        
       
      

       

        

       
       

        
    
         pthread_join(p, 
         nullptr);
        
       
      

       

        

       
       

        
    
         return 
         0;
        
       
      

       

        

       
       

        

         }
        
       
     
    

     
    

Task.hpp:

     

      

       

        

       
       

        

         #pragma once
        
       
      

       

        

       
       

        

         #include <iostream>
        
       
      

       

        

       
       

        

         #include <string>
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         class 
         Task
        
       
      

       

        

       
       

        

         {
        
       
      

       

        

       
       

        

         public:
        
       
      

       

        

       
       

        
    
         Task() 
        
       
      

       

        

       
       

        

             :
         elemOne_(
         0)
        
       
      

       

        

       
       

        

             ,
         elemTwo_(
         0)
        
       
      

       

        

       
       

        

             ,
         operator_(
         '0')
        
       
      

       

        

       
       

        

             {}
        
       
      

       

        

       
       

        
    
         Task(
         int one, 
         int two, 
         char op) 
        
       
      

       

        

       
       

        

             :
         elemOne_(one)
        
       
      

       

        

       
       

        

             ,
         elemTwo_(two)
        
       
      

       

        

       
       

        

             ,
         operator_(op)
        
       
      

       

        

       
       

        

             {}
        
       
      

       

        

       
       

        
    
         int operator() ()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         return 
         run();
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         int run()
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        
        
         int result = 
         0;
        
       
      

       

        

       
       

        
        
         switch (operator_)
        
       
      

       

        

       
       

        

                 {
        
       
      

       

        

       
       

        
        
         case 
         '+':
        
       
      

       

        

       
       

        

                     result = elemOne_ + elemTwo_;
        
       
      

       

        

       
       

        
            
         break;
        
       
      

       

        

       
       

        
        
         case 
         '-':
        
       
      

       

        

       
       

        

                     result = elemOne_ - elemTwo_;
        
       
      

       

        

       
       

        
            
         break;
        
       
      

       

        

       
       

        
        
         case 
         '*':
        
       
      

       

        

       
       

        

                     result = elemOne_ * elemTwo_;
        
       
      

       

        

       
       

        
            
         break;
        
       
      

       

        

       
       

        
        
         case 
         '/':
        
       
      

       

        

       
       

        

                 {
        
       
      

       

        

       
       

        
            
         if (elemTwo_ == 
         0)
        
       
      

       

        

       
       

        

                     {
        
       
      

       

        

       
       

        

                         std::cout << 
         "div zero, abort" << std::endl;
        
       
      

       

        

       
       

        

                         result = 
         -1;
        
       
      

       

        

       
       

        

                     }
        
       
      

       

        

       
       

        
            
         else
        
       
      

       

        

       
       

        

                     {
        
       
      

       

        

       
       

        

                         result = elemOne_ / elemTwo_;
        
       
      

       

        

       
       

        

                     }
        
       
      

       

        

       
       

        

                 }
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        
        
         break;
        
       
      

       

        

       
       

        
        
         case 
         '%':
        
       
      

       

        

       
       

        

                 {
        
       
      

       

        

       
       

        
            
         if (elemTwo_ == 
         0)
        
       
      

       

        

       
       

        

                     {
        
       
      

       

        

       
       

        

                         std::cout << 
         "mod zero, abort" << std::endl;
        
       
      

       

        

       
       

        

                         result = 
         -1;
        
       
      

       

        

       
       

        

                     }
        
       
      

       

        

       
       

        
            
         else
        
       
      

       

        

       
       

        

                     {
        
       
      

       

        

       
       

        

                         result = elemOne_ % elemTwo_;
        
       
      

       

        

       
       

        

                     }
        
       
      

       

        

       
       

        

                 }
        
       
      

       

        

       
       

        
        
         break;
        
       
      

       

        

       
       

        
        
         default:
        
       
      

       

        

       
       

        

                     std::cout << 
         "非法操作: " << operator_ << std::endl;
        
       
      

       

        

       
       

        
            
         break;
        
       
      

       

        

       
       

        

                 }
        
       
      

       

        

       
       

        
        
         return result;
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        
    
         int get(int *e1, int *e2, char *op)
        
       
      

       

        

       
       

        

             {
        
       
      

       

        

       
       

        

                 *e1 = elemOne_;
        
       
      

       

        

       
       

        

                 *e2 = elemTwo_;
        
       
      

       

        

       
       

        

                 *op = operator_;
        
       
      

       

        

       
       

        

             }
        
       
      

       

        

       
       

        

         private:
        
       
      

       

        

       
       

        
    
         int elemOne_;
        
       
      

       

        

       
       

        
    
         int elemTwo_;
        
       
      

       

        

       
       

        
    
         char operator_;
        
       
      

       

        

       
       

        

         };
        
       
     
    

     
    

makefile： 

     

      

       

        

       
       

        

         CC=g++
        
       
      

       

        

       
       

        

         FLAGS=-std=c++11
        
       
      

       

        

       
       

        

         LD=-lpthread
        
       
      

       

        

       
       

        

         bin=blockQueue
        
       
      

       

        

       
       

        

         src=BlockQueueTest.cc
        
       
      

       

        

       
       

        
 
        
       
      

       

        

       
       

        

         $(bin):$(src)
        
       
      

       

        

       
       

        

         	$(CC) -o 
         $@ $^ $(LD) $(FLAGS)
        
       
      

       

        

       
       

        

         .PHONY:clean
        
       
      

       

        

       
       

        

         clean:
        
       
      

       

        

       
       

        
	
         rm -f $(bin)
        
       
     

3.POSIX信号量

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

初始化信号量

      

       

        

         

        
        

         

          #include <semaphore.h>
         
        
       

        

         

        
        

         

          int sem_init(sem_t *sem, int pshared, unsigned int value);
         
        
       

        

         

        
        

         

          参数：
         
        
       

        

         

        
        

         

           pshared:
          0表示线程间共享，非零表示进程间共享
         
        
       

        

         

        
        

         

           value：信号量初始值
         
        
      

销毁信号量

int sem_destroy(sem_t *sem);

等待信号量

功能：等待信号量，会将信号量的值减1

int sem_wait(sem_t *sem);

发布信号量

功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1。

int sem_post(sem_t *sem);

上面的生产者-消费者的例子是基于queue的,其空间可以动态分配,现在基于固定大小的环形队列重写这个程序 （POSIX信号量）

信号量是一个计数器,描述临界资源数量的计数器。二元信号量==互斥锁

4.基于环形队列的生产消费模型

• 环形队列采用数组模拟，用模运算来模拟环状特性

• 环形结构起始状态和结束状态都是一样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留一个空的位置，作为满的状态

•  但是我们现在有信号量这个计数器，就很简单的进行多线程间的同步过程

 模拟实现：

   

    

     

      

     
     

      

       #include <iostream>
      
     
    

     

      

     
     

      

       #include <vector>
      
     
    

     

      

     
     

      

       #include <stdlib.h>
      
     
    

     

      

     
     

      

       #include <semaphore.h>
      
     
    

     

      

     
     

      

       #include <unistd.h>
      
     
    

     

      

     
     

      

       #include <pthread.h>
      
     
    

     

      

     
     

      

       #define NUM 16
      
     
    

     

      

     
     

      

       class 
       RingQueue
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

       private:
      
     
    

     

      

     
     

      

           std::vector<
       int> q;
      
     
    

     

      

     
     

      
    
       int cap;
      
     
    

     

      

     
     

      
    
       sem_t data_sem;
      
     
    

     

      

     
     

      
    
       sem_t space_sem;
      
     
    

     

      

     
     

      
    
       int consume_step;
      
     
    

     

      

     
     

      
    
       int product_step;
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       public:
      
     
    

     

      

     
     

      
    
       RingQueue(
       int _cap = NUM) : 
       q(_cap), 
       cap(_cap)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       sem_init(&data_sem, 
       0, 
       0);
      
     
    

     

      

     
     

      
        
       sem_init(&space_sem, 
       0, cap);
      
     
    

     

      

     
     

      

               consume_step = 
       0;
      
     
    

     

      

     
     

      

               product_step = 
       0;
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      
    
       void PutData(const int &data)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       sem_wait(&space_sem); 
       // P
      
     
    

     

      

     
     

      

               q[consume_step] = data;
      
     
    

     

      

     
     

      

               consume_step++;
      
     
    

     

      

     
     

      

               consume_step %= cap;
      
     
    

     

      

     
     

      
        
       sem_post(&data_sem); 
       // V
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      
    
       void GetData(int &data)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       sem_wait(&data_sem);
      
     
    

     

      

     
     

      

               data = q[product_step];
      
     
    

     

      

     
     

      

               product_step++;
      
     
    

     

      

     
     

      

               product_step %= cap;
      
     
    

     

      

     
     

      
        
       sem_post(&space_sem);
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

           ~
       RingQueue()
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       sem_destroy(&data_sem);
      
     
    

     

      

     
     

      
        
       sem_destroy(&space_sem);
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

       };
      
     
    

     

      

     
     

      

       void *consumer(void *arg)
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           RingQueue *rqp = (RingQueue *)arg;
      
     
    

     

      

     
     

      
    
       int data;
      
     
    

     

      

     
     

      
    
       for (;;)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      

               rqp->
       GetData(data);
      
     
    

     

      

     
     

      

               std::cout << 
       "Consume data done : " << data << std::endl;
      
     
    

     

      

     
     

      
        
       sleep(
       1);
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

       }
      
     
    

     

      

     
     

      

       // more faster
      
     
    

     

      

     
     

      

       void *producter(void *arg)
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           RingQueue *rqp = (RingQueue *)arg;
      
     
    

     

      

     
     

      
    
       srand((
       unsigned 
       long)
       time(
       NULL));
      
     
    

     

      

     
     

      
    
       for (;;)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       int data = 
       rand() % 
       1024;
      
     
    

     

      

     
     

      

               rqp->
       PutData(data);
      
     
    

     

      

     
     

      

               std::cout << 
       "Prodoct data done: " << data << std::endl;
      
     
    

     

      

     
     

      
        
       // sleep(1);
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

       }
      
     
    

     

      

     
     

      

       int main()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           RingQueue rq;
      
     
    

     

      

     
     

      
    
       pthread_t c, p;
      
     
    

     

      

     
     

      
    
       pthread_create(&c, 
       NULL, consumer, (
       void *)&rq);
      
     
    

     

      

     
     

      
    
       pthread_create(&p, 
       NULL, producter, (
       void *)&rq);
      
     
    

     

      

     
     

      
    
       pthread_join(c, 
       NULL);
      
     
    

     

      

     
     

      
    
       pthread_join(p, 
       NULL);
      
     
    

     

      

     
     

      

       }
      
     
   
  

   
  

RingQueue.hpp:

 

   

    

     

      

     
     

      

       #pragma once
      
     
    

     

      

     
     

      

       #include <iostream>
      
     
    

     

      

     
     

      

       #include <vector>
      
     
    

     

      

     
     

      

       #include <string>
      
     
    

     

      

     
     

      

       #include <semaphore.h>
      
     
    

     

      

     
     

      

       using 
       namespace std;
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       const 
       int gCap = 
       10;
      
     
    

     

      

     
     

      

       template <
       class 
       T>
      
     
    

     

      

     
     

      

       class 
       RingQueue
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

       public:
      
     
    

     

      

     
     

      
    
       RingQueue(
       int cap = gCap)
      
     
    

     

      

     
     

      

           :
       ringqueue_(cap)
      
     
    

     

      

     
     

      

           ,
       pIndex_(
       0)
      
     
    

     

      

     
     

      

           ,
       cIndex_(
       0)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       // 生产
      
     
    

     

      

     
     

      
        
       sem_init(&roomSem_, 
       0, ringqueue_.
       size());
      
     
    

     

      

     
     

      
        
       // 消费
      
     
    

     

      

     
     

      
        
       sem_init(&dataSem_, 
       0, 
       0);
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
        
       pthread_mutex_init(&pmutex_ ,
       nullptr);
      
     
    

     

      

     
     

      
        
       pthread_mutex_init(&cmutex_ ,
       nullptr);
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      
    
       // 生产
      
     
    

     

      

     
     

      
    
       void push(const T &in)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       sem_wait(&roomSem_); 
       //无法被多次的申请
      
     
    

     

      

     
     

      
        
       pthread_mutex_lock(&pmutex_);
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

               ringqueue_[pIndex_] = in; 
       //生产的过程
      
     
    

     

      

     
     

      

               pIndex_++;   
       // 写入位置后移
      
     
    

     

      

     
     

      

               pIndex_ %= ringqueue_.
       size(); 
       // 更新下标，保证环形特征
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
        
       pthread_mutex_unlock(&pmutex_);
      
     
    

     

      

     
     

      
        
       sem_post(&dataSem_);
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      
    
       // 消费
      
     
    

     

      

     
     

      
    
       T pop()
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       sem_wait(&dataSem_);
      
     
    

     

      

     
     

      
        
       pthread_mutex_lock(&cmutex_);
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

               T temp = ringqueue_[cIndex_];
      
     
    

     

      

     
     

      

               cIndex_++;
      
     
    

     

      

     
     

      

               cIndex_ %= ringqueue_.
       size();
       // 更新下标，保证环形特征
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
        
       pthread_mutex_unlock(&cmutex_);
      
     
    

     

      

     
     

      
        
       sem_post(&roomSem_);
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
        
       return temp;
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

           ~
       RingQueue()
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       sem_destroy(&roomSem_);
      
     
    

     

      

     
     

      
        
       sem_destroy(&dataSem_);
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
        
       pthread_mutex_destroy(&pmutex_);
      
     
    

     

      

     
     

      
        
       pthread_mutex_destroy(&cmutex_);
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

       private:
      
     
    

     

      

     
     

      

           vector<T> ringqueue_; 
       // 唤醒队列
      
     
    

     

      

     
     

      
    
       sem_t roomSem_;       
       // 衡量空间计数器，productor
      
     
    

     

      

     
     

      
    
       sem_t dataSem_;       
       // 衡量数据计数器，consumer
      
     
    

     

      

     
     

      
    
       uint32_t pIndex_;     
       // 当前生产者写入的位置， 如果是多线程，pIndex_也是临界资源
      
     
    

     

      

     
     

      
    
       uint32_t cIndex_;     
       // 当前消费者读取的位置，如果是多线程，cIndex_也是临界资源
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
    
       pthread_mutex_t pmutex_;
      
     
    

     

      

     
     

      
    
       pthread_mutex_t cmutex_;
      
     
    

     

      

     
     

      

       };
      
     
   
  

   
  

RingQueueTest.cc:

   

    

     

      

     
     

      

       #include "RingQueue.hpp"
      
     
    

     

      

     
     

      

       #include <ctime>
      
     
    

     

      

     
     

      

       #include <unistd.h>
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       // 我们是单生产者，单消费者
      
     
    

     

      

     
     

      

       // 多生产者，多消费者？？代码怎么改？
      
     
    

     

      

     
     

      

       // 为什么呢？？？多生产者，多消费者？
      
     
    

     

      

     
     

      

       // 不要只关心把数据或者任务，从ringqueue 放拿的过程，获取数据或者任务，处理数据或者任务，也是需要花时间的！
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       void *productor(void *args)
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           RingQueue<
       int> *rqp = 
       static_cast<RingQueue<
       int> *>(args);
      
     
    

     

      

     
     

      
    
       while(
       true)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       int data = 
       rand()%
       10;
      
     
    

     

      

     
     

      

               rqp->
       push(data);
      
     
    

     

      

     
     

      

               cout << 
       "pthread[" << 
       pthread_self() << 
       "]" << 
       " 生产了一个数据: " << data << endl;
      
     
    

     

      

     
     

      
        
       sleep(
       1);
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

       }
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       void *consumer(void *args)
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           RingQueue<
       int> *rqp = 
       static_cast<RingQueue<
       int> *>(args);
      
     
    

     

      

     
     

      
    
       while(
       true)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       //sleep(10);
      
     
    

     

      

     
     

      
        
       int data = rqp->
       pop();
      
     
    

     

      

     
     

      

               cout << 
       "pthread[" << 
       pthread_self() << 
       "]" << 
       " 消费了一个数据: " << data << endl;
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

       }
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       int main()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      
    
       srand((
       unsigned 
       long)
       time(
       nullptr)^
       getpid());
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

           RingQueue<
       int> rq;
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
    
       pthread_t c1,c2,c3, p1,p2,p3;
      
     
    

     

      

     
     

      
    
       pthread_create(&p1, 
       nullptr, productor, &rq);
      
     
    

     

      

     
     

      
    
       pthread_create(&p2, 
       nullptr, productor, &rq);
      
     
    

     

      

     
     

      
    
       pthread_create(&p3, 
       nullptr, productor, &rq);
      
     
    

     

      

     
     

      
    
       pthread_create(&c1, 
       nullptr, consumer, &rq);
      
     
    

     

      

     
     

      
    
       pthread_create(&c2, 
       nullptr, consumer, &rq);
      
     
    

     

      

     
     

      
    
       pthread_create(&c3, 
       nullptr, consumer, &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);
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
    
       return 
       0;
      
     
    

     

      

     
     

      

       }
      
     
   
  

   
  

 makefile:

 

   

    

     

      

     
     

      

       CC=g++
      
     
    

     

      

     
     

      

       FLAGS=-std=c++
       11
      
     
    

     

      

     
     

      

       LD=-lpthread
      
     
    

     

      

     
     

      

       bin=ringQueue
      
     
    

     

      

     
     

      

       src=RingQueueTest.cc
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       $(bin):$(src)
      
     
    

     

      

     
     

      

       	$(CC) -o $@ $^ $(LD) $(FLAGS)
      
     
    

     

      

     
     

      

       .PHONY:clean
      
     
    

     

      

     
     

      

       clean:
      
     
    

     

      

     
     

      

       	rm -f $(bin)
      
     
   

 

后记：
●由于作者水平有限，文章难免存在谬误之处，敬请读者斧正，俚语成篇，恳望指教！

                                                                           ——By 作者：新晓·故知