【并发编程】生产者-消费者模式
2023-06-13 09:13:57 时间
生产者和消费者也是一个非常经典的多线程模式 在生产-消费模式中:通常由两类线程 即若干生产者的线程和若干消费者的线程。生产者线程负责提交用户请求 消费者线程负责具体处理生产者提交的人 在生产者和消费者之间通过共享内存缓存区进行通信。
生产者消费者例子: Data类:
public final class Data {
private String id;
private String name;
public Data(String id, String name){
this.id = id;
this.name = name;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString(){
return "{id: " + id + ", name: " + name + "}";
}
}
Provider 类:
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Provider implements Runnable{
//共享缓存区
private BlockingQueue<Data> queue;
//多线程间是否启动变量,有强制从主内存中刷新的功能。即时返回线程的状态
private volatile boolean isRunning = true;
//id生成器
private static AtomicInteger count = new AtomicInteger();
//随机对象
private static Random r = new Random();
public Provider(BlockingQueue queue){
this.queue = queue;
}
@Override
public void run() {
while(isRunning){
try {
//随机休眠0 - 1000 毫秒 表示获取数据(产生数据的耗时)
Thread.sleep(r.nextInt(1000));
//获取的数据进行累计...
int id = count.incrementAndGet();
//比如通过一个getData方法获取了
Data data = new Data(Integer.toString(id), "数据" + id);
System.out.println("当前线程:" + Thread.currentThread().getName() + ", 获取了数据,id为:" + id + ", 进行装载到公共缓冲区中...");
//添加数据 在2秒中添加不了返回false
if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){
System.out.println("提交缓冲区数据失败....");
//do something... 比如重新提交
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public void stop(){
this.isRunning = false;
}
}
Consumer 类:
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
public class Consumer implements Runnable{
private BlockingQueue<Data> queue;
public Consumer(BlockingQueue queue){
this.queue = queue;
}
//随机对象
private static Random r = new Random();
@Override
public void run() {
while(true){
try {
//获取数据
Data data = this.queue.take();
//进行数据处理。休眠0 - 1000毫秒模拟耗时
Thread.sleep(r.nextInt(1000));
System.out.println("当前消费线程:" + Thread.currentThread().getName() + ", 消费成功,消费数据为id: " + data.getId());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
Main 类:
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class Main {
public static void main(String[] args) throws Exception {
//内存缓冲区
BlockingQueue<Data> queue = new LinkedBlockingQueue<Data>(10);
//生产者
Provider p1 = new Provider(queue);
Provider p2 = new Provider(queue);
Provider p3 = new Provider(queue);
//消费者
Consumer c1 = new Consumer(queue);
Consumer c2 = new Consumer(queue);
Consumer c3 = new Consumer(queue);
//创建线程池运行,这是一个缓存的线程池,可以创建无穷大的线程,没有任务的时候不创建线程。空闲线程存活时间为60s(默认值)
ExecutorService cachePool = Executors.newCachedThreadPool();
cachePool.execute(p1);
cachePool.execute(p2);
cachePool.execute(p3);
cachePool.execute(c1);
cachePool.execute(c2);
cachePool.execute(c3);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
p1.stop();
p2.stop();
p3.stop();
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// cachePool.shutdown();
// cachePool.shutdownNow();
}
}
运行结果:
相关文章
- 高并发下线程安全的单例模式(最全最经典)
- java并发之SynchronousQueue实现原理[通俗易懂]
- 【并发编程】future模式
- 【Java 并发编程】线程池机制 ( 线程池阻塞队列 | 线程池拒绝策略 | 使用 ThreadPoolExecutor 自定义线程池参数 )
- 掌握这两本书:搞定高并发这块硬骨头|文末赠书
- Linux进程与线程:一种新的并发处理模式(linux的进程和线程)
- 掌握MSSQL存储过程并发处理,提高系统性能(mssql存储过程并发)
- MySQL Nowait:高效实现并发控制的方式(mysqlnowait)
- Redis并发写入的实现及应用(并发写入redis)
- 解决Redis高并发一招制胜(redis高并发怎么处理)
- 利用Redis实现高并发应用程序(redis高并发功能)
- Redis在高并发环境下的挑战(redis高并发下的问题)
- php与redis的超强大联合支持高并发量数据(redis 高并发php)
- Redis网络架构支持优化的并发模式(redis网络并发模型)
- Go语言并发技术详解