主函數：

1 package com.ietree.basicskill.mutilthread.designpattern.ProducerConsumer; 2 3 import java.util.concurrent.BlockingQueue; 4 import java.util.concurrent.ExecutorService; 5 import java.util.concurrent.Executors; 6 import java.util.concurrent.LinkedBlockingQueue; 7 8 /** 9 * Created by Administrator on 2017/5/17. 10 */ 11 public class Main { 12 public static void main(String[] args) throws Exception { 13 //內存緩沖區 14 BlockingQueue<Data> queue = new LinkedBlockingQueue<Data>(10); 15 //生產者 16 Producer p1 = new Producer(queue); 17 Producer p2 = new Producer(queue); 18 Producer p3 = new Producer(queue); 19 //消費者 20 Consumer c1 = new Consumer(queue); 21 Consumer c2 = new Consumer(queue); 22 Consumer c3 = new Consumer(queue); 23 24 //創建線程池運行,這是一個緩存的線程池，可以創建無窮大的線程，沒有任務的時候不創建線程。空閑線程存活時間為60s（默認值） 25 ExecutorService cachePool = Executors.newCachedThreadPool(); 26 cachePool.execute(p1); 27 cachePool.execute(p2); 28 cachePool.execute(p3); 29 cachePool.execute(c1); 30 cachePool.execute(c2); 31 cachePool.execute(c3); 32 33 try { 34 Thread.sleep(3000); 35 } catch (InterruptedException e) { 36 e.printStackTrace(); 37 } 38 p1.stop(); 39 p2.stop(); 40 p3.stop(); 41 try { 42 Thread.sleep(2000); 43 } catch (InterruptedException e) { 44 e.printStackTrace(); 45 } 46 // cachePool.shutdown(); 47 // cachePool.shutdownNow(); 48 49 } 50 }

Producer類：

1 package com.ietree.basicskill.mutilthread.designpattern.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 /** 9 * Created by Administrator on 2017/5/17. 10 */ 11 public class Producer implements Runnable { 12 13 //共享緩存區 14 private BlockingQueue<Data> queue; 15 //多線程間是否啟動變量，有強制從主內存中刷新的功能。即時返回線程的狀態 16 private volatile boolean isRunning = true; 17 //id生成器 18 private static AtomicInteger count = new AtomicInteger(); 19 //隨機對象 20 private static Random r = new Random(); 21 22 public Producer(BlockingQueue queue){ 23 this.queue = queue; 24 } 25 26 @Override 27 public void run() { 28 while(isRunning){ 29 try { 30 //隨機休眠0 - 1000 毫秒 表示獲取數據(產生數據的耗時) 31 Thread.sleep(r.nextInt(1000)); 32 //獲取的數據進行累計... 33 int id = count.incrementAndGet(); 34 //比如通過一個getData方法獲取了 35 Data data = new Data(Integer.toString(id), "數據" + id); 36 System.out.println("當前線程:" + Thread.currentThread().getName() + ", 獲取了數據，id為:" + id + ", 進行裝載到公共緩沖區中..."); 37 if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){ 38 System.out.println("提交緩沖區數據失敗...."); 39 //do something... 比如重新提交 40 } 41 } catch (InterruptedException e) { 42 e.printStackTrace(); 43 } 44 } 45 } 46 47 public void stop(){ 48 this.isRunning = false; 49 } 50 }

Consumer類：

1 package com.ietree.basicskill.mutilthread.designpattern.ProducerConsumer; 2 3 import java.util.Random; 4 import java.util.concurrent.BlockingQueue; 5 6 /** 7 * Created by Administrator on 2017/5/17. 8 */ 9 public class Consumer implements Runnable { 10 private BlockingQueue<Data> queue; 11 12 public Consumer(BlockingQueue queue){ 13 this.queue = queue; 14 } 15 16 //隨機對象 17 private static Random r = new Random(); 18 19 @Override 20 public void run() { 21 while(true){ 22 try { 23 //獲取數據 24 Data data = this.queue.take(); 25 //進行數據處理。休眠0 - 1000毫秒模擬耗時 26 Thread.sleep(r.nextInt(1000)); 27 System.out.println("當前消費線程：" + Thread.currentThread().getName() + "， 消費成功，消費數據為id: " + data.getId()); 28 } catch (InterruptedException e) { 29 e.printStackTrace(); 30 } 31 } 32 } 33 }

Data類

1 package com.ietree.basicskill.mutilthread.designpattern.ProducerConsumer; 2 3 /** 4 * Created by Administrator on 2017/5/17. 5 */ 6 public class Data { 7 private String id; 8 private String name; 9 10 public Data(String id, String name){ 11 this.id = id; 12 this.name = name; 13 } 14 15 public String getId() { 16 return id; 17 } 18 19 public void setId(String id) { 20 this.id = id; 21 } 22 23 public String getName() { 24 return name; 25 } 26 27 public void setName(String name) { 28 this.name = name; 29 } 30 31 @Override 32 public String toString(){ 33 return "{id: " + id + ", name: " + name + "}"; 34 } 35 }

?

轉載于:https://www.cnblogs.com/Dylansuns/p/6869869.html

總結

以上是生活随笔為你收集整理的并发设计模式之生产者消费者设计模式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。