生产者和消费者问题是操作系统并发性经典问题之一，描述的是有一块缓冲区为仓库，生产者可以将商品放入仓库，消费者可以从仓库拿走商品，解决生产者和消费者问题的方法可以分为两类：1.采用某种机制保护生产者和消费者之间的同步。2.在生产者和消费者之间建立一个管道。
第2种方法不常用，暂不做讨论。同步的核心问题是如何保证同一资源在多线程并发访问时的完整性。常用的同步的方法是采用信号或加锁机制，保证资源在任意时刻之多被一个线程访问。在Java中一共有四种方法支持同步：
- wait/notify
- await/signal
- BlockingQueue
- PipedInputStream/PipedOutStream
前三种是同步方法，最后一种是管道方法，本文介绍前三种。
1.wait/notify(notifyAll)
wait,notify,notifyAll是基类Object的三个方法，它们必须是由同步监视器(synchronized包围的那个对象)来调用，用于协调多个线程对共享数据的存储。
wait:使当前线程暂停执行并释放对象锁标志，让其它线程可以进入synchronized数据快，当前线程被放入对象等待池中。
notify:从对象等待池中取走任意一个线程放入锁标志等待池中，如果锁标志等待池中没有线程，则notify不起作用。
notifyAll:从对象等待池中取走所有的线程放入对象等待池中。
注意：notify从对象等待线程池中取出的线程有可能不是期望的线程，而notifyAll有可能取出不必要的线程，影响性能。
下面贴上实现代码。
首先定义一个资源：
public class PublicResource {
private int number = 0;
private int size = 10;
/**
* 生产
*/
public synchronized int increase() {
while (number >= size) {
System.out.println(Thread.currentThread().getName() + "资源已满,生产者等待...");
try {
//对象的wait方法被调用后,线程进入对象的等待队列中,并释放对象锁,其它线程可以竞争使用此对象锁,
//sleep方法使得一个线程进入睡眠状态,但是线程所占有的资源并没有释放。
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
++number;
System.out.println(Thread.currentThread().getName() + "---生产者---生产了1个,总共有" + number);
notifyAll();
return number;
}
/**
* 消费
*/
public synchronized int decrease() {
while (number <= 0) {
System.out.println(Thread.currentThread().getName() + "---资源已空,消费者等待...");
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
--number;
System.out.println(Thread.currentThread().getName() + "---消费者---消费了1个,总共有" + number);
notifyAll();
return number;
}
/**
* 获取资源大小
*
* @return 资源大小
*/
public int getSize() {
return size;
}
}
贴上生产者代码：
public class IncreaseRunnable implements Runnable {
private PublicResource mPublicResource;
public IncreaseRunnable(PublicResource publicResource) {
mPublicResource = publicResource;
}
@Override
public void run() {
//生产一个资源
mPublicResource.increase();
}
}
贴上消费者代码：
public class DecreaseRunnable implements Runnable {
private PublicResource mPublicResource;
public DecreaseRunnable(PublicResource publicResource) {
mPublicResource = publicResource;
}
@Override
public void run() {
mPublicResource.decrease();
}
}
最后贴上main方法代码：
public class ProcuderConsumerDemo {
public static void main(String[] arg0) {
PublicResource publicResource = new PublicResource();
new Thread(new IncreaseRunnable(publicResource)).start();//生产一个资源线程
new Thread(new DecreaseRunnable(publicResource)).start();//消费一个资源线程
new Thread(new IncreaseRunnable(publicResource)).start();//生产一个资源线程
new Thread(new DecreaseRunnable(publicResource)).start();//消费一个资源线程
new Thread(new IncreaseRunnable(publicResource)).start();//生产一个资源线程
new Thread(new DecreaseRunnable(publicResource)).start();//消费一个资源线程
new Thread(new DecreaseRunnable(publicResource)).start();//消费一个资源线程
new Thread(new IncreaseRunnable(publicResource)).start();//生产一个资源线程
}
}
执行程序，打印日志如下：
Thread-0---生产了1个,总共有1
Thread-1---消费了1个,总共有0
Thread-1---资源已空,消费者等待...
Thread-2---生产了1个,总共有1
Thread-3---消费了1个,总共有0
Thread-3---资源已空,消费者等待...
Thread-4---生产了1个,总共有1
Thread-5---消费了1个,总共有0
Thread-5---资源已空,消费者等待...
await/signal(signalAll)
JDK5.0以后，Java提供了更加健壮的线程处理机制，包括同步，锁定，线程池等，它们可以实现更细粒度的线程控制。await／signal(signalAll)功能和wait／notify(notifyAll)基本相同，甚至可以完全取代。它们和新引入的锁机制Lock挂钩，具有更大的灵活性，同过在Lock对象上调用newCondition方法，将条件变量和锁对象进行绑定，进而控制并发程序访问竞争资源的安全。
贴上代码，这里只贴上PublicResource类，其它类同上面一样，就不重复贴出：
public class PublicResource {
private int number = 0;
private int size = 10;
private Lock mLock;
private Condition mIncreaseCondition;
private Condition mDecreaseCondition;
public PublicResource() {
mLock = new ReentrantLock();
mIncreaseCondition = mLock.newCondition();
mDecreaseCondition = mLock.newCondition();
}
/**
* 生产
*/
public int increase() {
mLock.lock();
while (number >= size) {
System.out.println(Thread.currentThread().getName() + "资源已满,生产者等待...");
try {
//对象的wait方法被调用后,线程进入对象的等待队列中,并释放对象锁,其它线程可以竞争使用此对象锁,
//sleep方法使得一个线程进入睡眠状态,但是线程所占有的资源并没有释放。
mIncreaseCondition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
++number;
System.out.println(Thread.currentThread().getName() + "---生产者---生产了1个,总共有" + number);
mIncreaseCondition.signalAll();
mDecreaseCondition.signalAll();
mLock.unlock();
return number;
}
/**
* 消费
*/
public int decrease() {
mLock.lock();
while (number <= 0) {
System.out.println(Thread.currentThread().getName() + "---资源已空,消费者等待...");
try {
mDecreaseCondition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
--number;
System.out.println(Thread.currentThread().getName() + "---消费者---消费了1个,总共有" + number);
mDecreaseCondition.signalAll();
mIncreaseCondition.signalAll();
mLock.unlock();
return number;
}
/**
* 获取资源大小
*
* @return 资源大小
*/
public int getSize() {
return size;
}
}
运行结果如下：
Thread-0---生产者---生产了1个,总共有1
Thread-1---消费者---消费了1个,总共有0
Thread-2---生产者---生产了1个,总共有1
Thread-3---消费者---消费了1个,总共有0
Thread-4---生产者---生产了1个,总共有1
Thread-5---消费者---消费了1个,总共有0
Thread-6---资源已空,消费者等待...
Thread-7---生产者---生产了1个,总共有1
Thread-6---消费者---消费了1个,总共有0
BlockingQueue##
BlockingQueue也是JDK5.0新增内容，在它内部实现了同步的队列，实现方式采用的是await/signal, 在生成对象时指定容量大小，用于阻塞操作是put方法和take方法。
put:类似生产者，容量达到最大时自动阻塞。
take:类似消费者，容量为0时自动阻塞。
贴上PublicResource代码，其它不变，同上。
public class PublicResource {
private int size = 10;
private BlockingDeque<Object> mBlockingDeque = new LinkedBlockingDeque<>(10);
/**
* 生产
*/
public int increase() {
while (mBlockingDeque.size() == size) {
System.out.println(Thread.currentThread().getName() + "资源已满,生产者等待...");
}
try {
mBlockingDeque.put(new Object());
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "---生产者---生产了1个,总共有" + mBlockingDeque.size());
return mBlockingDeque.size();
}
/**
* 消费
*/
public int decrease() {
while (mBlockingDeque.size() <= 0) {
System.out.println(Thread.currentThread().getName() + "---资源已空,消费者等待...");
}
try {
mBlockingDeque.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "---消费者---消费了1个,总共有" + mBlockingDeque.size());
return mBlockingDeque.size();
}
/**
* 获取资源大小
*
* @return 资源大小
*/
public int getSize() {
return size;
}
}
运行结果如下：
Thread-0---生产者---生产了1个,总共有1
Thread-1---消费者---消费了1个,总共有0
Thread-2---生产者---生产了1个,总共有1
Thread-3---消费者---消费了1个,总共有0
Thread-4---生产者---生产了1个,总共有1
Thread-5---消费者---消费了1个,总共有0
Thread-7---生产者---生产了1个,总共有1
Thread-6---资源已空,消费者等待...
Thread-6---消费者---消费了1个,总共有0
