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

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

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

一、架构模式图:

类图:

生产者:提交用户请求,提取用户任务,并装入内存缓冲区;

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

内存缓冲区:缓存生产者提交的任务或数据,供消费者使用;

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

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


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

(1)Producer生产者线程:

  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消费者线程:

  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共享数据模型:

  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函数:

  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

智能推荐

艾美捷Epigentek DNA样品的超声能量处理方案-程序员宅基地

文章浏览阅读15次。空化气泡的大小和相应的空化能量可以通过调整完全标度的振幅水平来操纵和数字控制。通过强调超声技术中的更高通量处理和防止样品污染,Epigentek EpiSonic超声仪可以轻松集成到现有的实验室工作流程中,并且特别适合与表观遗传学和下一代应用的兼容性。Epigentek的EpiSonic已成为一种有效的剪切设备,用于在染色质免疫沉淀技术中制备染色质样品,以及用于下一代测序平台的DNA文库制备。该装置的经济性及其多重样品的能力使其成为每个实验室拥有的经济高效的工具,而不仅仅是核心设施。

11、合宙Air模块Luat开发:通过http协议获取天气信息_合宙获取天气-程序员宅基地

文章浏览阅读4.2k次,点赞3次,收藏14次。目录点击这里查看所有博文  本系列博客,理论上适用于合宙的Air202、Air268、Air720x、Air720S以及最近发布的Air720U(我还没拿到样机,应该也能支持)。  先不管支不支持,如果你用的是合宙的模块,那都不妨一试,也许会有意外收获。  我使用的是Air720SL模块,如果在其他模块上不能用,那就是底层core固件暂时还没有支持,这里的代码是没有问题的。例程仅供参考!..._合宙获取天气

EasyMesh和802.11s对比-程序员宅基地

文章浏览阅读7.7k次,点赞2次,收藏41次。1 关于meshMesh的意思是网状物,以前读书的时候,在自动化领域有传感器自组网,zigbee、蓝牙等无线方式实现各个网络节点消息通信,通过各种算法,保证整个网络中所有节点信息能经过多跳最终传递到目的地,用于数据采集。十多年过去了,在无线路由器领域又把这个mesh概念翻炒了一下,各大品牌都推出了mesh路由器,大多数是3个为一组,实现在面积较大的住宅里,增强wifi覆盖范围,智能在多热点之间切换,提升上网体验。因为节点基本上在3个以内,所以mesh的算法不必太复杂,组网形式比较简单。各厂家都自定义了组_802.11s

线程的几种状态_线程状态-程序员宅基地

文章浏览阅读5.2k次,点赞8次,收藏21次。线程的几种状态_线程状态

stack的常见用法详解_stack函数用法-程序员宅基地

文章浏览阅读4.2w次,点赞124次,收藏688次。stack翻译为栈,是STL中实现的一个后进先出的容器。要使用 stack,应先添加头文件include<stack>,并在头文件下面加上“ using namespacestd;"1. stack的定义其定义的写法和其他STL容器相同, typename可以任意基本数据类型或容器:stack<typename> name;2. stack容器内元素的访问..._stack函数用法

2018.11.16javascript课上随笔(DOM)-程序员宅基地

文章浏览阅读71次。<li> <a href = "“#”>-</a></li><li>子节点:文本节点(回车),元素节点,文本节点。不同节点树:  节点(各种类型节点)childNodes:返回子节点的所有子节点的集合,包含任何类型、元素节点(元素类型节点):child。node.getAttribute(at...

随便推点

layui.extend的一点知识 第三方模块base 路径_layui extend-程序员宅基地

文章浏览阅读3.4k次。//config的设置是全局的layui.config({ base: '/res/js/' //假设这是你存放拓展模块的根目录}).extend({ //设定模块别名 mymod: 'mymod' //如果 mymod.js 是在根目录,也可以不用设定别名 ,mod1: 'admin/mod1' //相对于上述 base 目录的子目录}); //你也可以忽略 base 设定的根目录,直接在 extend 指定路径(主要:该功能为 layui 2.2.0 新增)layui.exten_layui extend

5G云计算:5G网络的分层思想_5g分层结构-程序员宅基地

文章浏览阅读3.2k次,点赞6次,收藏13次。分层思想分层思想分层思想-1分层思想-2分层思想-2OSI七层参考模型物理层和数据链路层物理层数据链路层网络层传输层会话层表示层应用层OSI七层模型的分层结构TCP/IP协议族的组成数据封装过程数据解封装过程PDU设备与层的对应关系各层通信分层思想分层思想-1在现实生活种,我们在喝牛奶时,未必了解他的生产过程,我们所接触的或许只是从超时购买牛奶。分层思想-2平时我们在网络时也未必知道数据的传输过程我们的所考虑的就是可以传就可以,不用管他时怎么传输的分层思想-2将复杂的流程分解为几个功能_5g分层结构

基于二值化图像转GCode的单向扫描实现-程序员宅基地

文章浏览阅读191次。在激光雕刻中,单向扫描(Unidirectional Scanning)是一种雕刻技术,其中激光头只在一个方向上移动,而不是来回移动。这种移动方式主要应用于通过激光逐行扫描图像表面的过程。具体而言,单向扫描的过程通常包括以下步骤:横向移动(X轴): 激光头沿X轴方向移动到图像的一侧。纵向移动(Y轴): 激光头沿Y轴方向开始逐行移动,刻蚀图像表面。这一过程是单向的,即在每一行上激光头只在一个方向上移动。返回横向移动: 一旦一行完成,激光头返回到图像的一侧,准备进行下一行的刻蚀。

算法随笔:强连通分量-程序员宅基地

文章浏览阅读577次。强连通:在有向图G中,如果两个点u和v是互相可达的,即从u出发可以到达v,从v出发也可以到达u,则成u和v是强连通的。强连通分量:如果一个有向图G不是强连通图,那么可以把它分成躲个子图,其中每个子图的内部是强连通的,而且这些子图已经扩展到最大,不能与子图外的任一点强连通,成这样的一个“极大连通”子图是G的一个强连通分量(SCC)。强连通分量的一些性质:(1)一个点必须有出度和入度,才会与其他点强连通。(2)把一个SCC从图中挖掉,不影响其他点的强连通性。_强连通分量

Django(2)|templates模板+静态资源目录static_django templates-程序员宅基地

文章浏览阅读3.9k次,点赞5次,收藏18次。在做web开发,要给用户提供一个页面,页面包括静态页面+数据,两者结合起来就是完整的可视化的页面,django的模板系统支持这种功能,首先需要写一个静态页面,然后通过python的模板语法将数据渲染上去。1.创建一个templates目录2.配置。_django templates

linux下的GPU测试软件,Ubuntu等Linux系统显卡性能测试软件 Unigine 3D-程序员宅基地

文章浏览阅读1.7k次。Ubuntu等Linux系统显卡性能测试软件 Unigine 3DUbuntu Intel显卡驱动安装,请参考:ATI和NVIDIA显卡请在软件和更新中的附加驱动中安装。 这里推荐: 运行后,F9就可评分,已测试显卡有K2000 2GB 900+分,GT330m 1GB 340+ 分,GT620 1GB 340+ 分,四代i5核显340+ 分,还有写博客的小盒子100+ 分。relaybot@re...

推荐文章

热门文章

相关标签