多线程模式(5):生产者消费者模式

1. 定义共享数据封装
   

package com.xqi.p_c;/*** 请求数据封装&＃xff08;以不变模式定义&＃xff09;* * &＃64;author mike 
*         2015年7月24日*/
public final class PCData {private final int intData;public PCData(int d) {intData &＃61; d;}public PCData(String d) {intData &＃61; Integer.valueOf(d);}public int getData(){return this.intData;}&＃64;Overridepublic String toString() {return "data:" &＃43; intData;}}

 2. 定义生产者

package com.xqi.p_c;import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;/*** 定义生产者* * &＃64;author mike 
*         2015年7月24日*/
public class Produer implements Runnable {/* 用volatile来修饰的变量&＃xff0c;表示随时都可能被其他的线程来改变&＃xff0c;并使其他的线程直接访问此变量&＃xff0c;而不保存其他的备份&＃xff01; */private volatile boolean isRunning &＃61; true;/* 定义内存缓冲区&＃xff0c;来存放生产者提交的请求与数据&＃xff0c; PCData就是数据的封装 */private BlockingQueue queue;/* 总数&＃xff0c;原子操作 &＃xff0c;AtomicInteger提供线程安全的加减操作接口 */private static AtomicInteger count &＃61; new AtomicInteger();/* 休眠时间 */private static final int SLEEPTIME &＃61; 1000;/*** 构造方法&＃xff0c;注入缓冲队列* * &＃64;param queue*/public Produer(BlockingQueue queue) {this.queue &＃61; queue;}public void run() {PCData data &＃61; null;Random r &＃61; new Random();System.out.println("start produer id &＃61; " &＃43; Thread.currentThread().getId());try {while (isRunning) {Thread.sleep(r.nextInt(SLEEPTIME)); // 使用随机产生时间差&＃xff08;消费者中一样&＃xff09;data &＃61; new PCData(count.incrementAndGet());// incrementAndGet表示&＃43;1System.out.println(data &＃43; " is put into queue!");if (!queue.offer(data, 2, TimeUnit.SECONDS)) {// 设定等待的时间为2秒&＃xff0c;如果在指定的时间内&＃xff0c;还不能往队列中加入&＃xff0c;则返回失败。System.err.println("failed to put data : " &＃43; data);}}} catch (InterruptedException e) {// 如果发生了异常就中断这个线程e.printStackTrace();Thread.currentThread().interrupt();}}//public void stop() {isRunning &＃61; false;}
}

 3. 定义消费者

package com.xqi.p_c;import java.text.MessageFormat;
import java.util.Random;
import java.util.concurrent.BlockingQueue;/*** 消费者* * &＃64;author mike 
*         2015年7月24日*/
public class Consumer implements Runnable {private BlockingQueue queue;private static final int SLEEPTIME &＃61; 1000;public Consumer(BlockingQueue queue) {this.queue &＃61; queue;}public void run() {System.out.println("start consumer id &＃61; " &＃43; Thread.currentThread().getId());Random r &＃61; new Random();try {while (true) {PCData data &＃61; queue.take();// 取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止if (null !&＃61; data) {int re &＃61; data.getData() * data.getData();System.out.println(MessageFormat.format("{0}*{1}&＃61;{2}", data.getData(), data.getData(), re));Thread.sleep(r.nextInt(SLEEPTIME));}}} catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();}}}

 4. 测试主类

package com.xqi.p_c;import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;/*** 测试主类* * &＃64;author mike 
*         2015年7月24日*/
public class PCTest {public static void main(String[] args) throws InterruptedException {BlockingQueue queue &＃61; new LinkedBlockingQueue(10);// 创建生产者Produer p1 &＃61; new Produer(queue);Produer p2 &＃61; new Produer(queue);Produer p3 &＃61; new Produer(queue);// 创建消费者Consumer c1 &＃61; new Consumer(queue);Consumer c2 &＃61; new Consumer(queue);Consumer c3 &＃61; new Consumer(queue);// 创建线程池&＃xff0c;newCachedThreadPool 定义了短期可重用的线程池ExecutorService service &＃61; Executors.newCachedThreadPool();// 运行生产者和消费者service.execute(p1);service.execute(p2);service.execute(p3);service.execute(c1);service.execute(c2);service.execute(c3);Thread.sleep(10 * 1000);// 终止线程中的whilep1.stop();p2.stop();p3.stop();Thread.sleep(3000);// shutdown() This method does not wait for previously submitted tasks to complete execution.// shutdown() 这个方法不会等待任务执行完成。(有说&＃xff1a;不再接受新的任务&＃xff0c;如果有等待的任务&＃xff0c;就执行完成)service.shutdown();// Attempts to stop all actively executing tasks, halts the// processing of waiting tasks, and returns a list of the tasks// that were awaiting execution. These tasks are drained (removed)// from the task queue upon return from this method.// 立即变为shutdown状态&＃xff0c;如果有正在执行的任务&＃xff0c;尝试停止&＃xff0c;并返回未完成的任务列表,然后移除// List taskList &＃61; service.shutdownNow();System.out.println("程序结束&＃xff01;");}}

Ps&＃xff1a;感觉shutdown了以后&＃xff0c;线程还是没有终止&＃xff0c;也进行不了操作&＃xff1f;&＃xff1f;&＃xff1f;&＃xff1f;&＃xff1f;&＃xff1f;不知道为什么&＃xff01;