并发编程实现模型之（三）Producer-Consumer模式_并发 producer/consumer-程序员宅基地

技术标签： 并发  Java并发编程  java  编程  

生产者-消费模式，通常有两类线程，即若干个生产者线程和若干个消费者线程。生产者线程负责提交用户请求，消费者线程负责具体处理生产者提交的任务。两者之间通过共享内存缓冲去进行通信。

一、架构模式图：

类图：

生产者：提交用户请求，提取用户任务，并装入内存缓冲区；

消费者：在内存缓冲区中提取并处理任务；

内存缓冲区：缓存生产者提交的任务或数据，供消费者使用；

任务：生产者向内存缓冲区提交的数据结构；

Main：使用生产者和消费者的客户端。


二、代码实现一个基于生产者-消费者模式的求整数平方的并行计算：

（1）Producer生产者线程：

[java]view plain copy print?
  1. package ProducerConsumer;  
  2.   
  3. import java.util.Random;  
  4. import java.util.concurrent.BlockingQueue;  
  5. import java.util.concurrent.TimeUnit;  
  6. import java.util.concurrent.atomic.AtomicInteger;  
  7.   
  8. public class Producer  implements Runnable{  
  9.       
  10.     //Volatile修饰的成员变量在每次被线程访问时，都强迫从共享内存中重读该成员变量的值。  
  11.     //而且，当成员变量发生变化时，强迫线程将变化值回写到共享内存。  
  12.     //这样在任何时刻，两个不同的线程总是看到某个成员变量的同一个值。  
  13.     private volatile  boolean isRunning= true;  
  14.       
  15.     //内存缓冲区  
  16.     private BlockingQueue<PCData> queue;  
  17.       
  18.     //总数，原子操作  
  19.     private static AtomicInteger count = new AtomicInteger();  
  20.        
  21.     private static final int SLEEPTIME=1000;  
  22.       
  23.       
  24.     public Producer(BlockingQueue<PCData> queue) {  
  25.           
  26.         this.queue = queue;  
  27.     }  
  28.   
  29.   
  30.   
  31.   
  32.     @Override  
  33.     public void run() {  
  34.         PCData data=null;  
  35.         Random r  = new Random();  
  36.         System.out.println("start producer id = "+ Thread .currentThread().getId());  
  37.         try{  
  38.             while(isRunning){  
  39.                 Thread.sleep(r.nextInt(SLEEPTIME));  
  40.                 //构造任务数据  
  41.                 data= new PCData(count.incrementAndGet());  
  42.                 System.out.println("data is put into queue ");  
  43.                 //提交数据到缓冲区  
  44.                 if(!queue.offer(data,2,TimeUnit.SECONDS)){  
  45.                     System.out.println("faile to  put data:  "+ data);  
  46.                 }  
  47.             }  
  48.         }catch (InterruptedException e){  
  49.             e.printStackTrace();  
  50.             Thread.currentThread().interrupt();  
  51.               
  52.         }  
  53.           
  54.           
  55.     }  
  56.   
  57.     public void stop(){  
  58.           
  59.         isRunning=false;  
  60.     }  
  61.   
  62.   
  63. }  

（2）Consumer消费者线程：

[java]view plain copy print?
  1. package ProducerConsumer;  
  2.   
  3. import java.text.MessageFormat;  
  4. import java.util.Random;  
  5. import java.util.concurrent.BlockingQueue;  
  6.   
  7. public class Consumer implements Runnable {  
  8.     //缓冲区     
  9.     private BlockingQueue<PCData> queue;  
  10.     private static final int SLEEPTIME=1000;  
  11.       
  12.       
  13.     public Consumer(BlockingQueue<PCData> queue) {          
  14.         this.queue = queue;  
  15.     }  
  16.   
  17.   
  18.     @Override  
  19.     public void run() {  
  20.         System.out.println("start Consumer id= "+ Thread .currentThread().getId());  
  21.         Random r = new Random();  
  22.           
  23.             try {  
  24.                 //提取任务  
  25.                 while(true){  
  26.                     PCData data= queue.take();  
  27.                     if(null!= data){  
  28.                         //计算平方  
  29.                         int re= data.getData()*data.getData();  
  30.                         System.out.println(MessageFormat.format("{0}*{1}={2}",  
  31.                                     data.getData(),data.getData(),re  
  32.                                 ));  
  33.                         Thread.sleep(r.nextInt(SLEEPTIME));  
  34.                                                   
  35.                     }  
  36.                 }  
  37.             } catch (InterruptedException e) {                
  38.                 e.printStackTrace();  
  39.                 Thread.currentThread().interrupt();  
  40.             }  
  41.               
  42.           
  43.           
  44.     }  
  45.       
  46.       
  47.   
  48.       
  49.   
  50. }  

（3）PCData共享数据模型：

[java]view plain copy print?
  1. package ProducerConsumer;  
  2.   
  3. public  final class PCData {  
  4.   
  5.     private final int intData;  
  6.   
  7.     public PCData(int d) {  
  8.         intData=d;  
  9.     }  
  10.       
  11.     public PCData(String  d) {  
  12.         intData=Integer.valueOf(d);  
  13.     }  
  14.       
  15.     public int getData(){  
  16.           
  17.         return intData;  
  18.           
  19.     }  
  20.     @Override  
  21.     public String toString(){  
  22.         return "data:"+ intData ;  
  23.     }  
  24.       
  25. }  

（4）Main函数：

[java]view plain copy print?
  1. package ProducerConsumer;  
  2.   
  3. import java.util.concurrent.BlockingQueue;  
  4. import java.util.concurrent.Executor;  
  5. import java.util.concurrent.ExecutorService;  
  6. import java.util.concurrent.Executors;  
  7. import java.util.concurrent.LinkedBlockingDeque;  
  8.   
  9. public class Main {  
  10.   
  11.     /** 
  12.      * @param args 
  13.      */  
  14.     public static void main(String[] args)  throws InterruptedException{  
  15.         //建立缓冲区  
  16.         BlockingQueue<PCData> queue = new LinkedBlockingDeque<PCData>(10);  
  17.         //建立生产者  
  18.         Producer producer1 = new Producer(queue);  
  19.         Producer producer2 = new Producer(queue);  
  20.         Producer producer3 = new Producer(queue);  
  21.           
  22.         //建立消费者  
  23.         Consumer consumer1 = new Consumer(queue);  
  24.         Consumer consumer2 = new Consumer(queue);  
  25.         Consumer consumer3 = new Consumer(queue);         
  26.                   
  27.         //建立线程池  
  28.         ExecutorService service = Executors.newCachedThreadPool();  
  29.           
  30.         //运行生产者  
  31.         service.execute(producer1);  
  32.         service.execute(producer2);  
  33.         service.execute(producer3);  
  34.         //运行消费者  
  35.         service.execute(consumer1);  
  36.         service.execute(consumer2);  
  37.         service.execute(consumer3);  
  38.       
  39.         Thread.sleep(10*1000);  
  40.           
  41.         //停止生产者  
  42.         producer1.stop();  
  43.         producer2.stop();  
  44.         producer3.stop();  
  45.           
  46.         Thread.sleep(3000);  
  47.         service.shutdown();  
  48.     }  
  49.   
  50. }  

三、注意：

    volatile关键字：Volatile修饰的成员变量在每次被线程访问时，都强迫从共享内存中重读该成员变量的值。而且，当成员变量发生变化时，强迫线程将变化值回写到共享内存。这样在任何时刻，两个不同的线程总是看到某个成员变量的同一个值。

    生产-消费模式的核心组件是共享内存缓冲区，是两者的通信桥梁，起到解耦作用，优化系统整体结构。

    由于缓冲区的存在，生产者和消费者，无论谁在某一局部时间内速度相对较高，都可以使用缓冲区得到缓解，保证系统正常运行，这在一定程度上缓解了性能瓶颈对系统系能的影响。
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://blog.csdn.net/zmx729618/article/details/78517675

智能推荐

基于内核4.19版本的XFRM框架_linux的xfrm框架-程序员宅基地

文章浏览阅读794次，点赞2次，收藏5次。XFRM框架_linux的xfrm框架

织梦常用标签整理_织梦中什么页面用什么标签教学-程序员宅基地

文章浏览阅读774次。DedeCMS常用标签讲解笔记整理 今天我们主要将模板相关内容，在前面的几节课中已经基本介绍过模板标签的相关内容，大家可以下载天工开物老师的讲课记录：http://bbs.dedecms.com/132951.html,这次课程我们主要讲解模板具体的标签使用，并且结合一些实例来介绍这些标签。 先前课程介绍了，网站的模板就如同一件衣服，衣服的好坏直接决定了网站的好坏，很多网站一看界面_织梦中什么页面用什么标签教学

工作中如何编译开源工具（gdb）_gdb编译-程序员宅基地

文章浏览阅读2.5k次，点赞2次，收藏15次。编译是大部分工程师的烦恼，大家普遍喜欢去写业务代码。但我觉得基本的编译流程，我们还是需要掌握的，希望遇到相关问题，不要退缩，尝试去解决。天下文章一大抄，百度能解决我们90%的问题。_gdb编译

python简易爬虫v1.0-程序员宅基地

文章浏览阅读1.8k次，点赞4次，收藏6次。python简易爬虫v1.0作者：William Ma (the_CoderWM)进阶python的首秀，大部分童鞋肯定是做个简单的爬虫吧，众所周知，爬虫需要各种各样的第三方库，例如scrapy, bs4, requests, urllib3等等。此处，我们先从最简单的爬虫开始。首先，我们需要安装两个第三方库：requests和bs4。在cmd中输入以下代码：pip install requestspip install bs4等安装成功后，就可以进入pycharm来写爬虫了。爬

安装flask后vim出现：error detected while processing /home/zww/.vim/ftplugin/python/pyflakes.vim:line 28_freetorn.vim-程序员宅基地

文章浏览阅读2.6k次。解决方法：解决方法可以去github重新下载一个pyflakes.vim。执行如下命令git clone --recursive git://github.com/kevinw/pyflakes-vim.git然后进入git克降目录，./pyflakes-vim/ftplugin,通过如下命令将python目录下的所有文件复制到~/.vim/ftplugin目录下即可。cp -R ...._freetorn.vim

HIT CSAPP大作业：程序人生—Hello‘s P2P-程序员宅基地

文章浏览阅读210次，点赞7次，收藏3次。本文简述了hello.c源程序的预处理、编译、汇编、链接和运行的主要过程，以及hello程序的进程管理、存储管理与I/O管理，通过hello.c这一程序周期的描述，对程序的编译、加载、运行有了初步的了解。_hit csapp

随便推点

挑战安卓和iOS！刚刚，华为官宣鸿蒙手机版，P40搭载演示曝光！高管现场表态：我们准备好了...-程序员宅基地

文章浏览阅读472次。点击上方 "程序员小乐"关注,星标或置顶一起成长后台回复“大礼包”有惊喜礼包！关注订阅号「程序员小乐」，收看更多精彩内容每日英文Sometimes you play a..._挑战安卓和ios!华为官宣鸿蒙手机版,p40搭载演示曝光!高管表态:我们准备好了

精选了20个Python实战项目(附源码)，拿走就用！-程序员宅基地

文章浏览阅读3.8w次，点赞107次，收藏993次。点击上方“Python爬虫与数据挖掘”，进行关注回复“书籍”即可获赠Python从入门到进阶共10本电子书今日鸡汤昔闻洞庭水，今上岳阳楼。大家好，我是小F。Python是目前最好的编程语言之一。由于其可读性和对初学者的友好性，已被广泛使用。那么要想学会并掌握Python，可以实战的练习项目是必不可少的。接下来，我将给大家介绍20个非常实用的Python项目，帮助大家更好的..._python项目

android在线图标生成工具,图标在线生成工具Android Asset Studio的使用-程序员宅基地

文章浏览阅读1.3k次。在网站的导航资源里看到了一个非常好用的东西：Android Asset Studio，可以在线生成各种图标。之前一直在用一个叫做Android Icon Creator的插件，可以直接在Android Studio的插件里搜索，这个工具的优点是可以生成适应各种分辨率的一套图标，有好几种风格的图标资源，遗憾的是虽然有很多套图标风格，毕竟是有限的。Android Asset Studio可以自己选择其..._在线 android 图标

android 无限轮播的广告位_轮播广告位-程序员宅基地

文章浏览阅读514次。无限轮播广告位没有录屏，将就将就着看，效果就是这样主要代码KsBanner.java/** * 广告位 * * Created by on 2016/12/20. */public class KsBanner extends FrameLayout implements ViewPager.OnPageChangeListener { private List

echart省会流向图（物流运输、地图）_java+echart地图+物流跟踪-程序员宅基地

文章浏览阅读2.2k次，点赞2次，收藏6次。继续上次的echart博客，由于省会流向图是从echart画廊中直接取来的。所以直接上代码<!DOCTYPE html><html><head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width,initial-scale=1,minimum-scale=1,maximum-scale=1,user-scalable=no" /&_java+echart地图+物流跟踪

Ceph源码解析：读写流程_ceph 发送数据到其他副本的源码-程序员宅基地

文章浏览阅读1.4k次。一、OSD模块简介1.1 消息封装：在OSD上发送和接收信息。cluster_messenger -与其它OSDs和monitors沟通client_messenger －与客户端沟通1.2 消息调度：Dispatcher类，主要负责消息分类1.3 工作队列：1.3.1 OpWQ: 处理ops（从客户端）和sub ops（从其他的OSD）。运行在op_tp线程池。1...._ceph 发送数据到其他副本的源码

推荐文章

热门文章

相关标签