本文主要是介绍《Java高并发程序设计》学习 --5.3 并行模式之生产者-消费者模式，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
生产者-消费者是一个经典的多线程设计模式，它为多线程间的协作提供了良好的解决方案。在生产者-消费者模式中，通常有两类线程，即若干个生产者线程和若干个消费者线程。生产者线程负责提交用户请求，消费者线程负责具体处理生产者提交的任务。生产者和消费者之间则通过共享内存缓冲进行通信。
生产者-消费者模式的核心组件是共享内存缓冲区，它作为生产者消费者间的通信桥梁，避免了两者直接通信，从而将生产者和消费者进行解耦。生产者不需要知道消费者存在，消费者也不需要知道生产者的存在。
同时，由于内存缓冲区的存在,允许生产者和消费者在执行速度上存在时间差，无论谁快谁慢,都可以通过共享缓冲区得到缓解,确保系统稳定允许。
生产者-消费者模式主要角色如下表所示。
| 角色 | 作用 |
| 生产者 | 用于提交用户请求,提取用户任务,并装入内存缓冲区 |
| 消费者 | 在内存缓冲区提取并处理任务 |
| 内存缓冲区 | 缓存生产者提交的任务或数据,供消费者使用 |
| 任务 | 生产者向内存缓冲区提交的数据结构 |
| Main | 使用生产者和消费者的客户端 |
实现一个基于生产者-消费者模式的求整数平方的并行程序。
首先，生产者线程的实现如下，它构建PCData对象，并放入BlockingQueue队列中。
public class Producer implements Runnable{private volatile boolean isRunning = true;private BlockingDeque<PCData> queue; //内存缓冲区,通过构造时外部引入,保证和消费者用的是同样的内存缓冲区.private static AtomicInteger count = new AtomicInteger(); //总数,原子操作.private static final int SLEEPTIME = 1000;public Producer(BlockingDeque<PCData> queue) {this.queue = queue;}@Overridepublic void run() {PCData data = null;Random random = new Random();System.out.println("start producter .."+Thread.currentThread().getId());try {while (isRunning){Thread.sleep(random.nextInt(SLEEPTIME)); //模拟执行过程data = new PCData(count.incrementAndGet()); //现获取当前值再+1System.out.println(data + " is put into Queue");//提交数据到缓冲队列中.设定等待的时间，如果在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败if (!queue.offer(data,2, TimeUnit.SECONDS)){System.out.println("failed to put data "+data);}}} catch (InterruptedException e) {e.printStackTrace();//因为BlockingQueue的offer操作上的锁是重入锁中的可以中断的锁,所以如果有异常,就中断,防止死锁.Thread.currentThread().interrupt();}}public void stop(){isRunning = false;}
}对应的消费者线程的实现如下。它从BlockingQueue队列中取出PCData对象，并进行相应的计算。
public class Consumer implements Runnable {private BlockingDeque<PCData> queue;private static final int SLEEPTIME = 1000;//同理,和Producter共用同一个BlockingQueue,保证存/取都在一个缓冲区public Consumer(BlockingDeque<PCData> queue) {this.queue = queue;}@Overridepublic void run() {System.out.println("start Consumer id : "+Thread.currentThread().getId());Random r = new Random();try {while (true){PCData data = queue.take();if (null != data){int re = data.getIntData() * data.getIntData();System.out.println(MessageFormat.format("{0} * {0} = {1}",data.getIntData(),re));Thread.sleep(r.nextInt(SLEEPTIME));}}}catch (InterruptedException e){e.printStackTrace();Thread.currentThread().interrupt();}}
}PCData作为生产者和消费者之间的共享数据模型，定义入下：
public class PCData {private final int intData;public PCData(int d) {intData = d;}public PCData(String d){intData = Integer.parseInt(d);}public int getIntData() {return intData;}@Overridepublic String toString() {return "PCData{" +"intData=" + intData +'}';}
}在主函数中，创建三个生产者和消费者，并让它们协作运行。在主函数的实现中，定义LinkedBlockingQueue作为BlockingQueue的实现类。
public class Main {public static void main(String[] a) throws InterruptedException {//建立共享缓冲区BlockingDeque<PCData> queue = new LinkedBlockingDeque<>(10);//建立生产者Producer producter1 = new Producer(queue);Producer producter2 = new Producer(queue);Producer producter3 = new Producer(queue);Producer producter4 = new Producer(queue);Producer producter5 = new Producer(queue);//建立消费者Consumer consumer1 = new Consumer(queue);Consumer consumer2 = new Consumer(queue);Consumer consumer3 = new Consumer(queue);//建立线程池ExecutorService es = Executors.newCachedThreadPool();//运行生产者es.execute(producter1);es.execute(producter2);es.execute(producter3);es.execute(producter4);es.execute(producter5);//运行消费者es.execute(consumer1);es.execute(consumer2);es.execute(consumer3);//运行时间Thread.sleep(1000 * 10);//停止生产者producter1.stop();producter2.stop();producter3.stop();producter4.stop();producter5.stop();//停止生产者后,预留时间给消费者执行Thread.sleep(1000 * 5);System.out.println("关闭线程池...");//关闭线程池es.shutdown();}
} 注：本篇博客内容摘自《 Java 高并发程序设计》
这篇关于《Java高并发程序设计》学习 --5.3 并行模式之生产者-消费者模式的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
