【Main.class】
package com.jxb.thread12;
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) {
//内存缓冲区(阻塞队列 无界) 10个长度
BlockingQueue<Data> queue=new LinkedBlockingQueue<Data>(10);
//生产者(将生产出来的消息放到queue里面，所以要引用queue)
Provider p1=new Provider(queue);
Provider p2=new Provider(queue);
Provider p3=new Provider(queue);
//消费者(从queue里面取出消息，所以要引用queue)
Consumer c1=new Consumer(queue);
Consumer c2=new Consumer(queue);
Consumer c3=new Consumer(queue);
/**
* 创建线程池运行，这是一个缓存的线程池，可以创建无穷大的线程，没有任务的时候不创建线程，
* 空闲线程存活时间为60s(默认值)
*/
ExecutorService cachePool=Executors.newCachedThreadPool();
//execute() 方法传递的参数，需要实现Runnable接口;submit()这个方法也可以
cachePool.execute(p1);
cachePool.execute(p2);
cachePool.execute(p3);
cachePool.execute(c1);
cachePool.execute(c2);
cachePool.execute(c3);
try {
Thread.sleep(3000);
} catch (Exception e) {
e.printStackTrace();
}
p1.stop();
p2.stop();
p3.stop();
try {
Thread.sleep(2000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
【Provider.class】
package com.jxb.thread12;
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生成器(AtomicInteger跟int 差不多，只不过它是原子性的)
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(); // count++
//比如通过一个getData方法获取了
Data data = new Data(Integer.toString(id), "数据" + id);
System.out.println("当前线程:" + Thread.currentThread().getName() + ","
+ " 获取了数据，id为:" + id + ", 进行装载到公共缓冲区中...");
//往queue里面加 元素(加的值，数值，数值的单位) 两秒之内加入不了data就返回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.class】
package com.jxb.thread12;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
/**
* 消费者
*/
public class Consumer implements Runnable {
//接收共有的queue
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());
System.out.print("");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
【Data.class】
package com.jxb.thread12;
public 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 "Data [id=" + id + ", name=" + name + "]";
}
}
【运行效果】
当前线程:pool-1-thread-3, 获取了数据，id为:1, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-2, 获取了数据，id为:2, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-1, 获取了数据，id为:3, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-2, 获取了数据，id为:4, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-1, 获取了数据，id为:5, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-3, 获取了数据，id为:6, 进行装载到公共缓冲区中...
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 2
当前消费线程：pool-1-thread-4， 消费成功，消费数据为id: 1
当前线程:pool-1-thread-3, 获取了数据，id为:7, 进行装载到公共缓冲区中...
当前消费线程：pool-1-thread-6， 消费成功，消费数据为id: 3
当前线程:pool-1-thread-2, 获取了数据，id为:8, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-2, 获取了数据，id为:9, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-2, 获取了数据，id为:10, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-1, 获取了数据，id为:11, 进行装载到公共缓冲区中...
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 4
当前消费线程：pool-1-thread-4， 消费成功，消费数据为id: 5
当前线程:pool-1-thread-2, 获取了数据，id为:12, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-1, 获取了数据，id为:13, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-3, 获取了数据，id为:14, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-1, 获取了数据，id为:15, 进行装载到公共缓冲区中...
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 7
当前消费线程：pool-1-thread-6， 消费成功，消费数据为id: 6
当前消费线程：pool-1-thread-4， 消费成功，消费数据为id: 8
当前消费线程：pool-1-thread-4， 消费成功，消费数据为id: 11
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 9
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 13
当前线程:pool-1-thread-2, 获取了数据，id为:16, 进行装载到公共缓冲区中...
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 14
当前线程:pool-1-thread-2, 获取了数据，id为:17, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-3, 获取了数据，id为:18, 进行装载到公共缓冲区中...
当前线程:pool-1-thread-1, 获取了数据，id为:19, 进行装载到公共缓冲区中...
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 15
当前消费线程：pool-1-thread-4， 消费成功，消费数据为id: 12
当前线程:pool-1-thread-2, 获取了数据，id为:20, 进行装载到公共缓冲区中...
当前消费线程：pool-1-thread-6， 消费成功，消费数据为id: 10
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 16
当前消费线程：pool-1-thread-6， 消费成功，消费数据为id: 18
当前消费线程：pool-1-thread-4， 消费成功，消费数据为id: 17
当前消费线程：pool-1-thread-6， 消费成功，消费数据为id: 20
当前消费线程：pool-1-thread-5， 消费成功，消费数据为id: 19
