devbox
繁依Fanyi0
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

多线程之生产者和消费者模型_多线程的生产者与消费者模型

作者：繁依Fanyi0 | 2024-06-02 06:35:08

赞

踩

多线程的生产者与消费者模型

这么经典的问题，要多看几次，多思考。

https://blog.csdn.net/chenxun_2010/article/details/49848865

(1)单消费者单生产者模式

  1. #include <iostream>
  2. #include <mutex>
  3. #include <thread>
  4. #include <condition_variable>
  5.  
  6. std::mutex mtx; //互斥变量
  7. std::condition_variable repo_not_full;//条件变量
  8. std::condition_variable repo_not_empty;//条件变量
  9.  
  10. static const int item_total = 20;//一共要生产的个数
  11. static const int repository_size = 10;//仓库的能够存储的个数
  12.  
  13. int product_position = 0;//buf中Product位置
  14. int consume_position = 0;//buf中consume位置
  15.  
  16. int item_buffer[repository_size]; //仓库buffer
  17.  
  18.  
  19. void produter_item(int i)
  20. {
  21.     std::unique_lock<std::mutex> lck(mtx);//锁住
  22. 	//如果不加一 会直接死锁
  23.     while((product_position + 1) % repository_size == consume_position ){
  24.         std::cout << "The item repository is full;" << std::endl;
  25.         repo_not_full.wait(lck);//等待repo_not_full的notify命令
  26.     }
  27.     item_buffer[product_position] = i;
  28.     product_position++;
  29.     if(product_position == repository_size){
  30.         product_position = 0;
  31.     }
  32.     repo_not_empty.notify_all();//通知别的等待进程，buf内不是空的了
  33.         
  34. }
  35.  
  36. int consumer_item()
  37. {
  38.     std::unique_lock<std::mutex>lck(mtx);
  39.     int data;
  40.     while(product_position == consume_position)
  41.     {
  42.         std::cout << "The item repository is empty." << std::endl;
  43.         repo_not_empty.wait(lck);
  44.     }
  45.     data = item_buffer[consume_position];
  46.     consume_position++;
  47.     if(consume_position == repository_size){
  48.         consume_position = 0;
  49.     }
  50.     repo_not_full.notify_all();
  51.     return data;
  52. }
  53.  
  54.  
  55.  
  56. //生产线程
  57. void Producter_thread()
  58. {
  59.     for(int i = 0;i < item_total; i++){
  60.         std::cout << "生产者生产第"<< i <<"个产品."<< std::endl;
  61.         produter_item(i);//执行producter动作
  62.     }    
  63. }
  64. //消费线程
  65. void Consumer_thread()
  66. {
  67.     int cnt = 0;
  68.     while(1){
  69.         int item = consumer_item();//执行consume动作
  70.         std::cout << "消费者消费第"<< item << "个产品." << std::endl;
  71.         cnt++;//个数+1
  72.         if(cnt == item_total){
  73.         std::cout << "consumer_thread is finish." << std::endl;
  74.             break; //结束
  75.         }
  76.     }
  77.     
  78. }
  79.  
  80. int main ()
  81. {
  82.     std::thread producer(Producter_thread);//开启一个producer线程，入口是Producter_thread函数
  83.     std::thread consumer(Consumer_thread);//同上
  84.     producer.join();//等producer结束
  85.     std::cout << "Producter_thread is end;" << std::endl;
  86.     consumer.join();//等consumer结束再结束main函数
  87.     std::cout << "consumer_thread is end;" << std::endl;
  88. }
  89.

(2) 多消费者，单生产者模型

多消费者模型，需要在单消费者模型上添加一个控制消费产品总数的变量，也是互斥访问的，有线程访问时候就加1。

  1. #include <iostream>
  2. #include <mutex>
  3. #include <thread>
  4. #include <condition_variable>
  5. #include <vector>
  6.  
  7. std::mutex mtx; //互斥变量
  8. std::mutex consumer_count_mtx;//消费者互斥变量
  9.  
  10. std::condition_variable repo_not_full;//条件变量
  11. std::condition_variable repo_not_empty;//条件变量
  12.  
  13. static const int item_total = 20;//一共要生产的个数
  14. static const int repository_size = 10;//仓库的能够存储的个数
  15.  
  16. static std::size_t consumer_items = 0; //消费者线程得到的产品数量
  17. static std::size_t product_position = 0;//buf中Product位置
  18. static std::size_t consume_position = 0;//buf中consume位置
  19.  
  20. int item_buffer[repository_size]; //仓库buffer
  21.  
  22.  
  23. void produter_item(int i)
  24. {
  25.     std::unique_lock<std::mutex> lck(mtx);//锁住
  26. 	//如果不加一 会直接死锁
  27.     while((product_position + 1) % repository_size == consume_position ){
  28.         std::cout << "仓库满了." << std::endl;
  29.         repo_not_full.wait(lck);//等待repo_not_full的notify命令
  30.     }
  31.     item_buffer[product_position] = i;
  32.     product_position++;
  33.     if(product_position == repository_size){
  34.         product_position = 0;
  35.     }
  36.     repo_not_empty.notify_all();//通知别的等待进程，buf内不是空的了
  37.         
  38. }
  39.  
  40. int consumer_item()
  41. {
  42.     std::unique_lock<std::mutex>lck(mtx);
  43.     int data;
  44.     while(product_position == consume_position)
  45.     {
  46.         std::cout << "仓库空了." << std::endl;
  47.         repo_not_empty.wait(lck);
  48.     }
  49.     data = item_buffer[consume_position];
  50.     consume_position++;
  51.     if(consume_position == repository_size){
  52.         consume_position = 0;
  53.     }
  54.     repo_not_full.notify_all();
  55.     return data;
  56. }
  57.  
  58.  
  59.  
  60. //生产线程
  61. void Producter_thread()
  62. {
  63.     for(int i = 0;i < item_total; i++){
  64.         std::cout << "生产者生产第"<< i <<"个产品."<< std::endl;
  65.         produter_item(i);//执行producter动作
  66.     }    
  67. }
  68. //消费线程
  69. void Consumer_thread()
  70. {
  71.     int exit = 0;
  72.     while(1){
  73.         std::unique_lock<std::mutex> lck(consumer_count_mtx);
  74.         consumer_items++;
  75.         if(consumer_items <= item_total){
  76.             int item = consumer_item();//执行consume动作
  77.             std::cout << "消费者消费第"<< item << "个产品." << std::endl;    	
  78.         }else{
  79.             exit = 1;
  80.         }
  81.         if(exit == 1){
  82.         std::cout << "消费完了所有的产品。." << std::endl;
  83.             break; //结束
  84.         }
  85.     }
  86.     std::cout << "消费者线程:" << std::this_thread::get_id() << "结束." << std::endl;
  87.     
  88. }
  89.  
  90. int main ()
  91. {
  92.     std::thread producer(Producter_thread);//开启一个producer线程，入口是Producter_thread函数
  93.     std::cout << "生产者线程id是:" << producer.get_id() << std::endl;
  94.     std::vector<std::thread> thread_vector;//定义一个线程向量
  95.     //将开启5个Consumer_thread线程
  96. 	for(int i = 0; i < 5; i++){
  97.         thread_vector.push_back(std::thread(Consumer_thread));
  98.     }
  99.     producer.join();//等producer结束
  100.     std::cout << "生产者线程结束." << std::endl;
  101.     for(auto &thr:thread_vector){
  102.     thr.join();//等consumer结束再结束main函数
  103.     }
  104. }
  105.  
  106.

3、多生产者单消费者模型

  1. #include <iostream>
  2. #include <mutex>
  3. #include <thread>
  4. #include <condition_variable>
  5. #include <vector>
  6.  
  7. std::mutex mtx; //互斥变量
  8. std::mutex producer_count_mtx;//消费者互斥变量
  9.  
  10. std::condition_variable repo_not_full;//条件变量
  11. std::condition_variable repo_not_empty;//条件变量
  12.  
  13. static const int item_total = 20;//一共要生产的个数
  14. static const int repository_size = 10;//仓库的能够存储的个数
  15.  
  16. static std::size_t consumer_items = 0; //消费者线程得到的产品数量
  17. static std::size_t producter_items = 0;//生产者线程生产的产品数量
  18. static std::size_t product_position = 0;//buf中Product位置
  19. static std::size_t consume_position = 0;//buf中consume位置
  20.  
  21. int item_buffer[repository_size]; //仓库buffer
  22.  
  23. std::chrono::seconds t(1);
  24.  
  25.  
  26. void produter_item(int i)
  27. {
  28.     std::unique_lock<std::mutex> lck(mtx);//锁住
  29. 	//如果不加一 会直接死锁
  30.     while((product_position + 1) % repository_size == consume_position ){
  31.         std::cout << "仓库满了." << std::endl;
  32.         repo_not_full.wait(lck);//等待repo_not_full的notify命令
  33.     }
  34.     item_buffer[product_position] = i;
  35.     product_position++;
  36.     if(product_position == repository_size){
  37.         product_position = 0;
  38.     }
  39.     repo_not_empty.notify_all();//通知别的等待进程，buf内不是空的了
  40.         
  41. }
  42.  
  43. int consumer_item()
  44. {
  45.     std::unique_lock<std::mutex>lck(mtx);
  46.     int data;
  47.     while(product_position == consume_position)
  48.     {
  49.         std::cout << "仓库空了." << std::endl;
  50.         repo_not_empty.wait(lck);
  51.     }
  52.     data = item_buffer[consume_position];
  53.     consume_position++;
  54.     if(consume_position == repository_size){
  55.         consume_position = 0;
  56.     }
  57.     repo_not_full.notify_all();
  58.     return data;
  59. }
  60.  
  61.  
  62.  
  63. //生产线程
  64. void Producter_thread()
  65. {
  66.     int exit = 0;
  67.  
  68.     while(1){
  69.         std::unique_lock<std::mutex> lck(producer_count_mtx);
  70.         producter_items++;
  71.         if(producter_items <= item_total){
  72.             std::cout << "生产第" << producter_items << "产品." << std::endl;
  73.             produter_item(producter_items - 1);
  74.         }else{
  75.             exit = 1;
  76.         }
  77.         if(exit == 1){
  78.             std::cout << "生产数量已经达到要求." << std::endl;
  79.             break;
  80.         }
  81.     }
  82.     std::cout << "生产者线程" << std::this_thread::get_id() << "结束." << std::endl;    
  83. }
  84. //消费线程
  85. void Consumer_thread()
  86. {
  87.     while(1){
  88.         std::this_thread::sleep_for(t);
  89.         if(consumer_items <= item_total){
  90.             int item = consumer_item();//执行consume动作
  91.             std::cout << "消费者消费"<< item << "产品." << std::endl;    	
  92.         }
  93.         consumer_items++;
  94.         if(consumer_items == item_total){
  95.             break;
  96.         }
  97.     }
  98.     
  99. }
  100.  
  101. int main ()
  102. {
  103.     //定义5个生产者线程和一个消费者线程
  104.     std::vector<std::thread> thread_vector;
  105.     for(int i = 0; i < 5; i++){
  106.         thread_vector.push_back(std::thread(Producter_thread));		
  107.     }
  108.     std::thread consumer(Consumer_thread);
  109.  
  110.     for(auto &thr: thread_vector){
  111.         thr.join();	
  112.     }
  113.     consumer.join();
  114.     std::cout << "消费者线程结束." << std::endl;
  115. }
  116.  
  117.

4、多生产者多消费者模型

  1. #include <iostream>
  2. #include <mutex>
  3. #include <thread>
  4. #include <condition_variable>
  5. #include <vector>
  6.  
  7. std::mutex mtx; //互斥变量
  8. std::mutex consumer_count_mtx;//消费者互斥变量
  9. std::mutex producer_count_mtx;//生产者互斥变量
  10.  
  11. std::condition_variable repo_not_full;//条件变量
  12. std::condition_variable repo_not_empty;//条件变量
  13.  
  14. static const int item_total = 20;//一共要生产的个数
  15. static const int repository_size = 10;//仓库的能够存储的个数
  16.  
  17. static std::size_t producer_items = 0;//生产者线程生产的产品数量
  18. static std::size_t consumer_items = 0; //消费者线程得到的产品数量
  19. static std::size_t product_position = 0;//buf中Product位置
  20. static std::size_t consume_position = 0;//buf中consume位置
  21.  
  22. int item_buffer[repository_size]; //仓库buffer
  23.  
  24. std::chrono::seconds t(1);
  25.  
  26. void produter_item(int i)
  27. {
  28.     std::unique_lock<std::mutex> lck(mtx);//锁住
  29. 	//如果不加一 会直接死锁
  30.     while((product_position + 1) % repository_size == consume_position ){
  31.         std::cout << "仓库满了." << std::endl;
  32.         repo_not_full.wait(lck);//等待repo_not_full的notify命令
  33.     }
  34.     item_buffer[product_position] = i;
  35.     product_position++;
  36.     if(product_position == repository_size){
  37.         product_position = 0;
  38.     }
  39.     repo_not_empty.notify_all();//通知别的等待进程，buf内不是空的了
  40.         
  41. }
  42.  
  43. int consumer_item()
  44. {
  45.     std::unique_lock<std::mutex>lck(mtx);
  46.     int data;
  47.     while(product_position == consume_position)
  48.     {
  49.         std::cout << "仓库空了." << std::endl;
  50.         repo_not_empty.wait(lck);
  51.     }
  52.     data = item_buffer[consume_position];
  53.     consume_position++;
  54.     if(consume_position == repository_size){
  55.         consume_position = 0;
  56.     }
  57.     repo_not_full.notify_all();
  58.     return data;
  59. }
  60.  
  61.  
  62.  
  63. //生产线程
  64. void Producter_thread()
  65. {
  66.     int exit = 0;
  67.     while(1){
  68.         std::unique_lock<std::mutex> lck(producer_count_mtx);
  69.         producer_items ++;
  70.         if(producer_items <= item_total){
  71.             std::cout << "生产者生产第"<< producer_items <<"个产品."<< std::endl;
  72.             produter_item(producer_items);//执行producter动作 
  73.         }else{
  74.             exit = 1;
  75.         }
  76.         if(exit == 1){
  77.             std::cout << "生产完了所有产品." << std::endl;
  78.             break;
  79.         }
  80.     }
  81.     std::cout << "生产者线程:" << std::this_thread::get_id() <<"结束." << std::endl;
  82. }
  83. //消费线程
  84. void Consumer_thread()
  85. {
  86.     int exit = 0;
  87.     while(1){
  88.         std::this_thread::sleep_for(t);
  89.         std::unique_lock<std::mutex> lck(consumer_count_mtx);
  90.         consumer_items++;
  91.         if(consumer_items <= item_total){
  92.             int item = consumer_item();//执行consume动作
  93.             std::cout << "消费者消费第"<< item << "个产品." << std::endl;    	
  94.         }else{
  95.             exit = 1;
  96.         }
  97.         if(exit == 1){
  98.         std::cout << "消费完了所有的产品." << std::endl;
  99.             break; //结束
  100.         }
  101.     }
  102.     std::cout << "消费者线程:" << std::this_thread::get_id() << "结束." << std::endl;
  103.     
  104. }
  105.  
  106. int main ()
  107. {
  108.     std::vector<std::thread> producer_vector;//定义一个生产者线程向量
  109.     std::vector<std::thread> consumer_vector;//定义一个消费者线程向量
  110.     //将开启5个Consumer_thread线程
  111. 	for(int i = 0; i < 5; i++){
  112.         producer_vector.push_back(std::thread(Producter_thread));
  113.         consumer_vector.push_back(std::thread(Consumer_thread));
  114.     }
  115.     //等待生产者线程结束
  116.     for(auto &thr:producer_vector){
  117.         thr.join();
  118.     }
  119.     //等待消费者线程结束
  120.     for(auto &thr:consumer_vector){
  121.         thr.join();//等consumer结束再结束main函数
  122.     }
  123. }
  124.  
  125.  
  126.

总结：

多线程的时候，要在两个以上线程公用的变量前面加锁，阻止访问冲突。要像更加灵活的运用锁，就要和条件变量配合来使用。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/661617
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号