并发编程实现模型之（三）Producer-Consumer模式_并发 producer/consumer-程序员宅基地

技术标签：并发 Java并发编程 java 编程

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

一、架构模式图：

类图：

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

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

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

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

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

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

（1）Producer生产者线程：

[java]view plain copy print?

package ProducerConsumer;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Producer implements Runnable{
//Volatile修饰的成员变量在每次被线程访问时，都强迫从共享内存中重读该成员变量的值。
//而且，当成员变量发生变化时，强迫线程将变化值回写到共享内存。
//这样在任何时刻，两个不同的线程总是看到某个成员变量的同一个值。
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("start producer id = "+ Thread .currentThread().getId());
try{
while(isRunning){
Thread.sleep(r.nextInt(SLEEPTIME));
//构造任务数据
data= new PCData(count.incrementAndGet());
System.out.println("data is put into queue ");
//提交数据到缓冲区
if(!queue.offer(data,2,TimeUnit.SECONDS)){
System.out.println("faile to put data: "+ data);
}
}
}catch (InterruptedException e){
e.printStackTrace();
Thread.currentThread().interrupt();
}
}
public void stop(){
isRunning=false;
}
}

（2）Consumer消费者线程：

[java]view plain copy print?

package ProducerConsumer;
import java.text.MessageFormat;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
public class Consumer implements Runnable {
//缓冲区
private BlockingQueue<PCData> queue;
private static final int SLEEPTIME=1000;
public Consumer(BlockingQueue<PCData> queue) {
this.queue = queue;
}
@Override
public 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.getData()*data.getData();
System.out.println(MessageFormat.format("{0}*{1}={2}",
data.getData(),data.getData(),re
));
Thread.sleep(r.nextInt(SLEEPTIME));
}
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
}
}
}

（3）PCData共享数据模型：

[java]view plain copy print?

package ProducerConsumer;
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函数：

[java]view plain copy print?

package ProducerConsumer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
public class Main {
/**
* @param args
*/
public static void main(String[] args) throws InterruptedException{
//建立缓冲区
BlockingQueue<PCData> queue = new LinkedBlockingDeque<PCData>(10);
//建立生产者
Producer producer1 = new Producer(queue);
Producer producer2 = new Producer(queue);
Producer producer3 = new Producer(queue);
//建立消费者
Consumer consumer1 = new Consumer(queue);
Consumer consumer2 = new Consumer(queue);
Consumer consumer3 = new Consumer(queue);
//建立线程池
ExecutorService service = Executors.newCachedThreadPool();
//运行生产者
service.execute(producer1);
service.execute(producer2);
service.execute(producer3);
//运行消费者
service.execute(consumer1);
service.execute(consumer2);
service.execute(consumer3);
Thread.sleep(10*1000);
//停止生产者
producer1.stop();
producer2.stop();
producer3.stop();
Thread.sleep(3000);
service.shutdown();
}
}

三、注意：

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

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

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

本文链接：https://blog.csdn.net/zmx729618/article/details/78517675

原作者删帖不实内容删帖广告或垃圾文章投诉

智能推荐

ARP协议图文详解，包含完整测试代码-程序员宅基地

文章浏览阅读4.1k次，点赞3次，收藏56次。图文详解ARP协议，花5分钟时间彻底弄懂ARP协议。_arp协议

oracle查看锁表语句、解锁方法_oracle查询锁表语句-程序员宅基地

文章浏览阅读2.2w次，点赞8次，收藏57次。oracle死锁原因查看以及解决办法_oracle查询锁表语句

文章浏览阅读4.5k次，点赞4次，收藏29次。Seata是一款开源的分布式事务解决方案，可以用于解决分布式系统中的数据一致性问题。它是由阿里巴巴集团发起的开源项目，目前得到了广泛的应用和支持。在 Seata 开源之前，其内部版本在阿里系内部一直扮演着应用架构层数据一致性的中间件角色，帮助经济体平稳的度过历年的双11在分布式系统中，由于数据存储在不同的节点上，因此需要保证分布式事务的原子性（即要么全部成功，要么全部失败），否则将会出现数据不一致的情况。Seata提供了一种可靠的分布式事务解决方案，可以对全局事务进行管理和协调，从而实现分布式事务的原子_seata配置

Golang | Leetcode Golang题解之第36题有效的数独-程序员宅基地

文章浏览阅读520次。Golang | Leetcode Golang题解之第36题有效的数独

稍微深入分析Ubuntu环境下安装NVIDIA驱动导致黑屏的原因_prime-select nvidia 黑屏-程序员宅基地

文章浏览阅读8.2k次，点赞5次，收藏48次。本文承接之前写的有关如何用正确姿势安装NVIDIA驱动的博文（https://blog.csdn.net/Edward_ed_liu/article/details/109552761）。首先之所以要更新Linux内核，是因为不更新内核就无法使用笔记本自带的无线网卡。其次，目前NVIDIA官方不建议把Linux内核更新到最新版（5.9），而且这条消息只在英文的官网才有，中文的则是广告。TWICE如果强行更新到5.9版本，之后的Cuda安装表面上会显示成功，但在实际使用Cuda的过程中._prime-select nvidia 黑屏

java中间件有哪些_金九银十期间成功斩获58万架构师Offer！六面字节跳动面经和面试题分享 - 小梦爱Java...-程序员宅基地

文章浏览阅读167次。金九银十期间成功斩获58万Offer！六面字节跳动面经(成功关键：吃透九大核心知识+狂刷大厂面试真题)第一轮：团队面试第一轮基本上是你的团队成员面试你，是和你同级或者高你一个P的师兄来面你，我的话基本没问什么特别的，主要还是讲自己简历上的做的项目，这里需要你很熟悉自己的项目才行，我个人觉得这里你要把项目里你的角色做了什么没做什么讲清楚，然后最好能把自己做的那部分重点展开来讲，然后面试官会从你讲的内..._java架构师中间件简历