去过机场的人应该都会看到放行李的传送带，工作人员将托运的行李不停的放入传送带，乘客下飞机之后会不停的拿走行李。
这个例子和生产者/消费者设计模式相似。
这种设计模式需要满足以下三点要求：
（1）生产者生产数据到缓冲区中，消费者从缓冲区中取数据。
（2）如果缓冲区已经满了，则生产者线程阻塞；
（3）如果缓冲区为空，那么消费者线程阻塞。
编写之前分析：
（1）定义一个缓存队列，选择一个集合当做缓存，给予缓存上限，缓存队列只有两种行为（生产数据和消费数据）；
（2）定义一个生产者线程，调用缓存队列中的生产行为；
（3）定义一个消费者线程，调用缓存队列中的消费行为；
开始编写代码：
定义一个缓存队列
/**
* 公共缓存队列
* 只做两件事：（1）生产；（2）消费
*/
public class PublicQueue<T> {
private int putIndex = 0;//数据插入的角标
private int maxCount = 50;//缓存区最大长度
private LinkedHashMap<Integer, T> linkedHashMap = new LinkedHashMap<>();//缓冲区
public synchronized void add(T msg){
if(linkedHashMap.size() == maxCount){
//如果缓存区达到最大数量，则阻塞生产者线程
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}else{
notifyAll();//唤醒所有线程
}
linkedHashMap.put(putIndex, msg);
System.out.println("生产一个产品，当前商品角标为："+putIndex+"===文本为："+msg+"===缓存长度为："+linkedHashMap.size());
putIndex = (putIndex + 1 >= maxCount) ? (putIndex + 1) % maxCount : putIndex + 1;
}
public synchronized T remove(){
if(linkedHashMap.size() == 0){
//如果缓存区没有数据，则阻塞消费线程
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}else{
notifyAll();//唤醒所有线程
}
Iterator it = linkedHashMap.entrySet().iterator();
T t = null;
if(it.hasNext()){
Map.Entry<Integer, T> entry = (Map.Entry<Integer, T>) it.next();
t = entry.getValue();
int index = entry.getKey();
linkedHashMap.remove(index);
System.out.println("消费一个产品，当前商品角标为："+index+"===文本为："+ t +"===缓存长度为："+linkedHashMap.size());
}
return t;
}
}
定义一个生产者线程
/**
* 生产者线程
*/
public class ProducerThread extends Thread {
private PublicQueue publicQueue;
public ProducerThread(PublicQueue publicQueue){
this.publicQueue = publicQueue;
}
@Override
public void run() {
for(int i=0;i<60;i++){
publicQueue.add(String.valueOf(i));
}
}
}
定义一个消费者
/**
* 消费者线程
*/
public class ConsumerThread extends Thread {
private PublicQueue publicQueue;
public ConsumerThread(PublicQueue publicQueue){
this.publicQueue = publicQueue;
}
@Override
public void run() {
for(;;){
publicQueue.remove();
}
}
}
启动：
public class ProducerConsumerTest {
public static void main(String[] args){
PublicQueue publicQueue = new PublicQueue();
ProducerThread producerThread = new ProducerThread(publicQueue);
ConsumerThread consumerThread = new ConsumerThread(publicQueue);
producerThread.start();//启动生产者线程
consumerThread.start();//启动消费者线程
}
}
代码分析：
（1）生产者/消费者设计模式顾名思义就是两个互斥线程，一个负责生产，一个负责消费，两者是线程不安全的；
（2）这里选择使用LinkedHashMap作为缓存队列，LinkedHashMap是一个双向链表，用来处理线程不安全的数据，可以保证取出第一个数据，it.next()就是取出第一个数据。（LinkedHashMap可以保证遍历的顺序）
（3）为了保证互斥线程的安全性，需要做对应的处理，以上代码使用了synchronized 、wait()、notifyAll()来保证。
那么本例中保证线程安全还有什么方案呢？
（1）lock和condition的await、signalAll
/**
* 公共缓存队列
* 只做两件事：（1）生产；（2）消费
*/
public class PublicQueue<T> {
private int putIndex = 0;//数据插入的角标
private int maxCount = 50;//缓存区最大长度
private Lock lock;
private Condition addCondition;
private Condition removeCondition;
public PublicQueue(){
lock = new ReentrantLock();
addCondition = lock.newCondition();
removeCondition =lock.newCondition();
}
private LinkedHashMap<Integer, T> linkedHashMap = new LinkedHashMap<>();//缓冲区
public void add(T msg){
try {
lock.lock();
if (linkedHashMap.size() == maxCount){
//如果缓存区达到最大数量，则阻塞生产者线程
addCondition.await();//等待
}
linkedHashMap.put(putIndex, msg);
System.out.println("生产一个产品，当前商品角标为："+putIndex+"===文本为："+msg+"===缓存长度为："+linkedHashMap.size());
putIndex = (putIndex + 1 >= maxCount) ? (putIndex + 1) % maxCount : putIndex + 1;
removeCondition.signalAll();//唤醒所有线程
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public T remove(){
T t = null;
try {
lock.lock();
if (linkedHashMap.size() == 0){
//如果缓存区没有数据，则阻塞消费线程
removeCondition.await();//等待
}
Iterator it = linkedHashMap.entrySet().iterator();
if(it.hasNext()){
Map.Entry<Integer, T> entry = (Map.Entry<Integer, T>) it.next();
t = entry.getValue();
int index = entry.getKey();
linkedHashMap.remove(index);
System.out.println("消费一个产品，当前商品角标为："+index+"===文本为："+ t +"===缓存长度为："+linkedHashMap.size());
}
addCondition.signalAll();//唤醒所有线程
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
return t;
}
}
（2）利用阻塞队列BlockingQueue （最简单的）
/**
* 公共缓存队列
* 只做两件事：（1）生产；（2）消费
*/
public class PublicQueue<T> {
private BlockingDeque<T> blockingDeque = new LinkedBlockingDeque<>(50);//缓冲区
public void add(T msg){
try {
blockingDeque.put(msg);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("生产一个产品，当前商品角标为："+"===文本为："+msg);
}
public T remove(){
T t = null;
try {
t = blockingDeque.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("消费一个产品，当前商品角标为："+"===文本为："+t);
return t;
}
}
总结：
有关缓存队列的处理有三种方法：
（1）双向链表LinkedHashMap和synchronized结合；
（2）双向链表LinkedHashMap和lock结合；
（3）直接使用阻塞队列BlockingQueue。
[本章完...]
网友评论