java线程深度解析（五）——并发模型（生产者-消费者）

本文主要是介绍java线程深度解析（五）——并发模型（生产者-消费者），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

http://blog.csdn.net/Daybreak1209/article/details/51378055

三、生产者-消费者模式

在经典的多线程模式中，生产者-消费者为多线程间协作提供了良好的解决方案。基本原理是两类线程，即若干个生产者和若干个消费者，生产者负责提交用户请求任务（到内存缓冲区），消费者线程负责处理任务（从内存缓冲区中取任务进行处理），两类线程之间通过共享内存缓冲区进行通信。

共享内存缓冲区的存在避免生产者和消费者直接通信，且允许消费者和生产者执行速度上存在时间差，无论谁快谁慢，都可以通过缓冲区缓解，确保系统正常运行。

生产者消费者模式中主要角色

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

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

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

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

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

下面用代码实现一个基于本模式的求整数平方和的并行计算。

具体实现采用BlockingQueue充当缓冲区，创建一个任务类PCData，生产者负责创建PCData对象放入缓冲区，消费者负责处理从缓冲区中取出PCData对象进行处理。

（1）Producer生产者线程

[html]  view plain copy  
 /*  
  * 负责创建数据对象PCD并提交到内存缓冲区中  
  */  
 public class Producer implements Runnable {  
     private volatile boolean isRunning=true;  
     private BlockingQueue<PCData> queue;  
     private static AtomicInteger count=new AtomicInteger();//总数，原子操作      
     private static final int SLEEPTIME=1000;      
     public  Producer(BlockingQueue<PCData> queue)  
     {  
         this.queue=queue;  
     }  
       
     @Override  
     public void run() {  
         PCData data=null;  
         Random r=new Random();  
         System.out.println("生产者当前线程"+Thread.currentThread().getId());  
           
         try{  
             while(isRunning)  
             {  
                 Thread.sleep(r.nextInt(SLEEPTIME));  
                 data=new PCData(count.incrementAndGet());//构造任务数据  
               
                 System.out.println(count.incrementAndGet());  
                 System.out.println(data +"已进入缓存区");  
                 if(!queue.offer(data,2,TimeUnit.SECONDS))  
                 {  
                     //提交数据到缓冲区  
                     System.err.println(data+"存入失败");  
                 }  
             }  
         }catch(Exception e)  
         {  
             Thread.currentThread().interrupt();  
         }         
     }  
     public void stop()  
     {  
         isRunning=false;  
     }      
 }  

（2）Consumer消费者线程：

[html]  view plain copy  
 /*  
  * 从缓冲区中获取PCData对象  
  */  
 public class Consumer implements Runnable{  
     private BlockingQueue<PCData> queue;  
     private static final int SLEEPTIME=1000;  
       
     public Consumer(BlockingQueue<PCData> queue)  
     {  
         this.queue=queue;  
     }  
       
     public void run()  
     {  
         System.out.println("消费者开始取数据，当前线程ID:"+Thread.currentThread().getId());  
         Random r=new Random();  
         try  
         {  
             while(true)  
             {  
                 PCData data=queue.take();  
                 if(null!=data)  
                 {  
                     System.out.println("从缓冲区中获取数据"+data.getData());  
                     int re=data.getData()*data.getData();//计算平方  
                     System.out.println(MessageFormat.format("{0}*{1}={2}",data.getData(),data.getData(),re));  
                     System.out.println("本数据对象处理完毕");  
                     Thread.sleep(r.nextInt(SLEEPTIME));  
                 }  
                           
             }  
         }catch(Exception e)  
         {  
             Thread.currentThread().interrupt();  
         }  
     }  
 }  

（3）PCData共享数据模型：

[html]  view plain copy  
 public final class PCData {  
     private final int intData;  
     public PCData(int d)  
     {  
         intData=d;  
     }  
     public PCData(String d)  
     {  
         intData=Integer.valueOf(d);  
     }  
       
     public int getData()  
     {  
         return intData;  
     }  
       
     @Override  
     public String toString()  
     {  
         return "data:"+intData;  
     }     
 }  

（4）Main函数：

[html]  view plain copy  
 public class Client {  
     public static void main(String[] args) throws InterruptedException {  
         //建立缓冲区  
         BlockingQueue<PCData> queue=new LinkedBlockingDeque<PCData>(10);  
         //建立3个生产者  
         Producer p1=new Producer(queue);  
         Producer p2=new Producer(queue);  
         Producer p3=new Producer(queue);  
         //建立3个消费者  
         Consumer c1=new Consumer(queue);  
         Consumer c2=new Consumer(queue);  
         Consumer c3=new Consumer(queue);  
           
         //创建线程池  
         ExecutorService threadPool=Executors.newCachedThreadPool();  
         threadPool.execute(p1);//启动生产者线程  
         threadPool.execute(p2);  
         threadPool.execute(p3);  
           
         threadPool.execute(c1);//启动消费者线程  
         threadPool.execute(c2);  
         threadPool.execute(c3);  
           
         Thread.sleep(10*1000);  
         //停止生产  
         p1.stop();  
         p2.stop();  
         p3.stop();//当消费者处理完缓冲区中所有数据，程序执行完毕  
           
         Thread.sleep(3000);  
         threadPool.shutdown();    
     }  
 }  