使用Lock和Condition实现生产者消费者模型

package Condition;

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 仓库类Storage实现缓冲区
 * @author Tao
 *
 */
class Storage {

//仓库最大存储量
private int MAX_SIZE;
//仓库存储的载体
private LinkedList<Object> list = new LinkedList<>();
//锁
private final Lock lock = new ReentrantLock();
//“仓库满”的条件变量
private final Condition produceCondition = lock.newCondition();
//“仓库空”的条件变量
private final Condition consumeCondition = lock.newCondition();

public Storage(int maxSize) {
this.MAX_SIZE = maxSize;
}

/**
 * 生产num个产品
 * @param num
 */
public void produce(int num) {

//获取锁
lock.lock();

//剩余空位置不足
while(list.size() + num >= MAX_SIZE) {

System.out.println("【要生产的产品数量】:" + num + "\t【库存量】:" + list.size()  
             + "\t【剩余空间】:" + (MAX_SIZE - list.size()) + "\t暂时不能执行生产任务!"); 

try {
//没法生产，先阻塞
produceCondition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

//剩余空间足够，生产num个产品
for(int i = 0; i < num; ++i) {
list.addLast(new Object());
}
System.out.println("【本次生产产品数】:" + num + "\t【当前库存为】:" + list.size()); 

//唤醒所有消费者线程
consumeCondition.signalAll();
//解锁
lock.unlock();
}

/**
 * 消费num个产品
 * @param num
 */
public void consume(int num) {

//获得锁
lock.lock();

//如果仓库存储量不够num个，本次消费阻塞
while(list.size() < num) {

System.out.println("【要消费的产品数量】:" + num + "\t【库存量】:" + list.size()  
            + "\t暂时不能执行消费任务!");  

try {
//没法消费，阻塞
consumeCondition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

//库存足够本次消费,消费num个产品
for(int i = 0; i < num; ++i) {
list.removeFirst();
}
System.out.println("【本次消费产品数】:" + num + "\t【当前库存为】:" + list.size());

//消费完了,唤醒所有生产者线程
produceCondition.signalAll();
//释放锁
lock.unlock();
}

}


/**
 * 生产者类
 * 一个生产者对应一个线程
 * @author Tao
 *
 */
class Producer extends Thread {

//每次生产的数量
private int num;
//持有的仓库的引用
private Storage storage;

public Producer(int num, Storage storage) {
this.num = num;
this.storage = storage;
}

@Override
public void run() {
try {
while(true) {
//生产产品
storage.produce(num);
Thread.sleep(3000);
}
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

}


/**
 * 消费者类
 * 一个消费者对应一个线程
 * @author Tao
 *
 */
class Consumer extends Thread {

//每次消费的数量
private int num;
//仓库的引用
private Storage storage;

public Consumer(int num, Storage storage) {
this.num = num;
this.storage = storage;
}

@Override
public void run() {
try {
while(true) {
storage.consume(num);
Thread.sleep(4000);
}
} catch (Exception e) {
e.printStackTrace();
}

}

}



public class Test {

public static void main(String[] args) {

Storage storage = new Storage(100);

//生产者 
Producer[] producers = new Producer[7];
for(int i = 0; i < producers.length; ++i) {
producers[i] = new Producer(40, storage);
}

//消费者
Consumer[] consumers = new Consumer[14];
for(int i = 0; i < consumers.length; ++i) {
consumers[i] = new Consumer(26, storage);
}

for (Producer producer : producers) {
producer.start();
}
for (Consumer consumer : consumers) {
consumer.start();
}

}
}