码迷，mamicode.com
首页 > 编程语言 > 详细

多线程：多线程设计模式（四）：生产者-消费模式

时间：2017-09-07 17:10:49      阅读：320      评论：0      收藏：0      [点我收藏+]

标签：数据结构   message   sdn   lan   put   dex   random   fail   port   

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

一、架构模式图：

技术分享

类图：

技术分享

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

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

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

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

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

 

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

（1）Producer生产者线程：

 

[java] view plain copy
 
  1. <span style="font-size:18px;">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. }  
  64. </span>  

 

 

 

 

（2）Consumer消费者线程：

 

[java] view plain copy
 
  1. <span style="font-size:18px;">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. }  
  51. </span>  

 

 

 

 

（3）PCData共享数据模型：

[java] view plain copy
 
  1. <span style="font-size:18px;">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. }  
  26. </span>  

 

（4）Main函数：

[java] view plain copy
 
  1. <span style="font-size:18px;">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. }  
  51. </span>  

 

三、注意：

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

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

    由于缓冲区的存在，生产者和消费者，无论谁在某一局部时间内速度相对较高，都可以使用缓冲区得到缓解，保证系统正常运行，这在一定程度上缓解了性能瓶颈对系统系能的影响。

转：http://blog.csdn.net/lmdcszh/article/details/39699261

多线程：多线程设计模式（四）：生产者-消费模式

标签：数据结构   message   sdn   lan   put   dex   random   fail   port   

原文地址：http://www.cnblogs.com/dengyungao/p/7490370.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！