生产——消费者模式
2017-08-07 18:11
302 查看
public class Data {
private String id;
private String name;
public Data(String id, String name) {
super();
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() {
// TODO Auto-generated method stub
return "(id:"+id+",name:"+name+")";
}
}
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{
//生产数据执行时间
Thread.sleep(r.nextInt(1000));
//获取数据进行累计
int id = count.incrementAndGet();
Data data = new Data(Integer.toString(id), "数据"+id);
System.out.println("当前线程:"+Thread.currentThread().getName()+",获取了数据,id为"+id+",进行装载到公共缓冲区...");
if(!this.queue.offer(data,2,TimeUnit.SECONDS)){//add直接加元素 offer加元素带加入的时间 2秒加入返回true
System.out.println("提交缓冲区数据...");
}
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
public void stop(){
this.isRunning = false;
}
}
public class Consumer implements Runnable{
private BlockingQueue<Data> queue;
public Consumer(BlockingQueue queue) {
this.queue = queue;
}
//随机对象
public static Random r = new Random();
@Override
public void run() {
while(true){
try{
Data data = this.queue.take();
Thread.sleep(r.nextInt(1000));
System.out.println("当前消费线程为:"+Thread.currentThread().getName()+",消费成功,消费数据为id:"+data.getId());
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
}
public class Main {
public static void main(String[] args) {
//内存缓冲区域
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) {
// TODO Auto-generated catch block
e.printStackTrace();
}
p1.stop();
p2.stop();
p3.stop();
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
private String id;
private String name;
public Data(String id, String name) {
super();
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() {
// TODO Auto-generated method stub
return "(id:"+id+",name:"+name+")";
}
}
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{
//生产数据执行时间
Thread.sleep(r.nextInt(1000));
//获取数据进行累计
int id = count.incrementAndGet();
Data data = new Data(Integer.toString(id), "数据"+id);
System.out.println("当前线程:"+Thread.currentThread().getName()+",获取了数据,id为"+id+",进行装载到公共缓冲区...");
if(!this.queue.offer(data,2,TimeUnit.SECONDS)){//add直接加元素 offer加元素带加入的时间 2秒加入返回true
System.out.println("提交缓冲区数据...");
}
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
public void stop(){
this.isRunning = false;
}
}
public class Consumer implements Runnable{
private BlockingQueue<Data> queue;
public Consumer(BlockingQueue queue) {
this.queue = queue;
}
//随机对象
public static Random r = new Random();
@Override
public void run() {
while(true){
try{
Data data = this.queue.take();
Thread.sleep(r.nextInt(1000));
System.out.println("当前消费线程为:"+Thread.currentThread().getName()+",消费成功,消费数据为id:"+data.getId());
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
}
public class Main {
public static void main(String[] args) {
//内存缓冲区域
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) {
// TODO Auto-generated catch block
e.printStackTrace();
}
p1.stop();
p2.stop();
p3.stop();
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
相关文章推荐
- 详解Java设计模式中的生产-消费者模式及代码实现
- 在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
- 生产消费者模式
- 生产消费者模式
- 多线程等待唤醒机制之生产消费者模式
- 非阻塞队列实现生产消费者模式
- 深入理解生产消费者模式
- Java线程机制的浅析 生产/消费者模式 LinkedBlockingQueue
- 设计模式笔记之六:生产消费者模式
- 生产消费者模式
- 多线程等待唤醒机制之生产消费者模式
- RabbitMQ下的生产消费者模式与订阅发布模式
- 多线程等待唤醒机制之生产消费者模式
- 原版线程通信之生产消费者模式
- 多线程等待唤醒机制之生产消费者模式
- java生产消费者模式
- 一个简单的java生产着消费者模式的代码
- RabbitMQ下的生产消费者模式与订阅发布模式
- python 线程同步:生产/消费者模式
- 生产消费者模式