文章目录

• 九、多线程
• • 7. 生产者消费者模型
  • • 生产者消费者模型的简单代码
    • 结果演示
• 未完待续

九、多线程

7. 生产者消费者模型

生产者消费者模型的简单代码

Makefile：

cp:Main.cc 	g++ -o $@ $^ -std=c++11 -lpthread .PHONY:clean clean: 	rm -f cp 

Thread.hpp：

#ifndef __THREAD_HPP__ #define __THREAD_HPP__  #include  #include  #include  #include  #include   namespace ThreadModule {     template     using func_t = std::function;      template     class Thread     {     public:         void Excute()         {             _func(_data);         }     public:         Thread(func_t func, T& data, const std::string &name="none-name")             : _func(func)             , _data(data)             , _threadname(name)             , _stop(true)         {}          static void *threadroutine(void *args)         {             Thread *self = static_cast *>(args);             self->Excute();             return nullptr;         }          bool Start()         {             int n = pthread_create(&_tid, nullptr, threadroutine, this);             if(!n)             {                 _stop = false;                 return true;             }             else             {                 return false;             }         }          void Detach()         {             if(!_stop)             {                 pthread_detach(_tid);             }         }          void Join()         {             if(!_stop)             {                 pthread_join(_tid, nullptr);             }         }          std::string name()         {             return _threadname;         }          void Stop()         {             _stop = true;         }          ~Thread() {}     private:         pthread_t _tid;         std::string _threadname;         T& _data;         func_t _func;         bool _stop;     }; }  #endif 

BlockQueue.hpp：

#ifndef __BLOCKQUEUE_HPP__ #define __BLOCKQUEUE_HPP__  #include  #include  #include  #include   template class BlockQueue { private:     bool IsFull() const     {         return _block_queue.size() == _cap;     }      bool IsEmpty() const     {         return _block_queue.empty();     } public:     BlockQueue(int cap)         :_cap(cap)     {         _productor_wait_num = 0;         _consumer_wait_num = 0;         // 初始化互斥锁         pthread_mutex_init(&_mutex, nullptr);         // 初始化条件变量         pthread_cond_init(&_productor_cond, nullptr);          // 初始化条件变量         pthread_cond_init(&_consumer_cond, nullptr);     }      // 生产者使用的入队列接口     void Enqueue(const T& in)     {         // 加锁         pthread_mutex_lock(&_mutex);         while (IsFull())         {             // 生产者等待数量加1             _productor_wait_num++;             // 等待条件变量通知唤醒并竞争到互斥锁             pthread_cond_wait(&_productor_cond, &_mutex);             // 生产者等待数量减1             _productor_wait_num--;         }         // 生产的数据入资源队列         _block_queue.push(in);         // 解锁         pthread_mutex_unlock(&_mutex);         // 通知消费者可以从等待队列中出队列         if (_consumer_wait_num > 0) pthread_cond_signal(&_consumer_cond);     }      // 消费者使用的出队列接口     void Pop(T* out)     {         // 加锁         pthread_mutex_lock(&_mutex);         while (IsEmpty())         {             // 消费者等待数量加1             _consumer_wait_num++;             // 等待条件变量通知唤醒并竞争到互斥锁             pthread_cond_wait(&_consumer_cond, &_mutex);             // 消费者等待数量减1             _consumer_wait_num--;         }         // 获取数据         *out = _block_queue.front();         // 数据出队列         _block_queue.pop();         // 解锁         pthread_mutex_unlock(&_mutex);         // 通知生产者可以从等待队列中出队列         if (_productor_wait_num > 0) pthread_cond_signal(&_productor_cond);     }      ~BlockQueue()     {         // 销毁互斥锁         pthread_mutex_destroy(&_mutex);         // 销毁条件变量         pthread_cond_destroy(&_productor_cond);         // 销毁条件变量         pthread_cond_destroy(&_consumer_cond);     } private:     std::queue _block_queue;     // 容量上限     int _cap;     // 互斥锁     pthread_mutex_t _mutex;     // 条件变量，用于通知生产者可以入队列     pthread_cond_t _productor_cond;     // 条件变量，用于通知消费者可以出队列     pthread_cond_t _consumer_cond;     // 生产者等待数量     int _productor_wait_num;     // 消费者等待数量     int _consumer_wait_num; };  #endif 

Task.hpp：

#pragma once  #include  #include    class Task { public:     Task()     {}      Task(int a, int b)         :_a(a)         ,_b(b)         ,_result(0)     {}      // 执行任务     void Execute()     {         _result = _a + _b;     }      std::string ResultToString()     {         return std::to_string(_a) + " + " + std::to_string(_b) + " = " + std::to_string(_result);     }      std::string DebugToString()     {         return std::to_string(_a) + " + " + std::to_string(_b) + " = ?";     } private:     int _a;     int _b;     int _result; }; 

Main.cc：

#include "BlockQueue.hpp" #include "Thread.hpp" #include "Task.hpp" #include  #include  #include   using namespace ThreadModule; // 创建类型别名 using blockqueue_t = BlockQueue;  // 消费者线程 void Consumer(blockqueue_t& bq) {     while (true)     {         Task t;         // 从阻塞队列中获取任务资源         bq.Pop(&t);         // 执行任务         t.Execute();         // 输出结果         std::cout << "Consumer: " << t.ResultToString() << std::endl;     } }  // 生产者线程 void Productor(blockqueue_t& bq) {     srand(time(nullptr)^pthread_self());     while (true)     {         // 分配任务         int a = rand() % 10 + 1;         usleep(1234);         int b = rand() % 20 + 1;         Task t(a, b);         // 任务放入阻塞队列         bq.Enqueue(t);         // 输出任务信息         std::cout << "Productor: " << t.DebugToString() << std::endl;         sleep(1);     } }  // 启动线程 void StartComm(std::vector>* threads, int num, blockqueue_t& bq, func_t func) {     for (int i = 0; i < num; i++)     {         // 创建一批线程         std::string name = "thread-" + std::to_string(i + 1);         threads->emplace_back(func, bq, name);         threads->back().Start();     } }  // 创建消费者线程 void StartConsumer(std::vector>* threads, int num, blockqueue_t& bq) {     StartComm(threads, num, bq, Consumer); }  // 创建生产者线程 void StartProductor(std::vector>* threads, int num, blockqueue_t& bq) {     StartComm(threads, num, bq, Productor); }  // 等待所有线程结束 void WaitAllThread(std::vector>& threads) {     for (auto& thread : threads)     {         thread.Join();     } }  int main() {     // 创建阻塞队列，容量为5     blockqueue_t* bq = new blockqueue_t(5);     // 创建线程     std::vector> threads;     // 创建 1个消费者线程     StartConsumer(&threads, 1, *bq);     // 创建 1个生产者线程     StartProductor(&threads, 1, *bq);      // 等待所有线程结束     WaitAllThread(threads);      return 0; } 

结果演示

这里使用的是单生产者和单消费者，当然也可以在主函数处创建多生产者和多消费者的模型。

未完待续