Java线程通信实现生产者-消费者

生产者消费者是经典的线程之间同步通信问题，生产者线程只有在产品仓库中没有产品的时候才生产产品，当它生成完一个产品以后唤醒消费者线程，消费者线程只有在产品仓库中有产品的时候才能取走产品，然后唤醒生产者线程。

Java可以有好几种方法解决这个问题。首先基础的当然是用Object的wait()、notify()和notifyAll()。

产品仓库类：

//产品仓库
public class ProductStore {
	private boolean flag =  false;
	public boolean hasProduct(){//是否有产品
		return flag;
	}
	/**
	 * 生产产品
	 * @throws Exception
	 */
	public synchronized void makeProduct() throws Exception{
		while(hasProduct()){//如果生产线程唤醒的还是生产线程，这个被唤醒的生产线程将继续wait
			this.wait();
		}
		Thread.sleep(300);
		flag = true;
		System.out.println(Thread.currentThread().getName()+":生产了一个产品");
		this.notifyAll();//唤醒所有线程
	}
	/**
	 * 取走产品
	 * @throws Exception
	 */
	public synchronized void getProduct() throws Exception{
		while(!hasProduct()){
			this.wait();
		}
		Thread.sleep(100);
		flag = false;
		System.out.println(Thread.currentThread().getName()+":取走一个产品");
		this.notifyAll();
	}
}

 生产者类：

public class Producer implements Runnable{
	
	ProductStore store;
	public Producer(ProductStore store){
		this.store = store;
	}

	@Override
	public void run() {
		try {
			store.makeProduct();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

 消费者类：

public class Consumer implements Runnable{
	
	ProductStore store;
	public Consumer(ProductStore store){
		this.store = store;
	}
	
	@Override
	public void run() {
		try {
			store.getProduct();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

 主测试类：

public class Test3 {	
	public static void main(String[] args) {
		ProductStore store = new ProductStore();
		for (int i = 1; i <= 5; i++) {
			new Thread(new Consumer(store), "消费者"+i).start();
		}
		
		for (int i = 1; i <= 5; i++) {
			new Thread(new Producer(store), "生产者"+i).start();
		}
	}
}

运行结果：

生产者1:生产了一个产品
消费者4:取走一个产品
生产者4:生产了一个产品
消费者5:取走一个产品
生产者2:生产了一个产品
消费者1:取走一个产品
生产者5:生产了一个产品
消费者2:取走一个产品
生产者3:生产了一个产品
消费者3:取走一个产品

 第二种方法就是利用java.util.concurrent包下的Lock得到Conditon，利用Condition的await()、signal()、signalAll()实现线程的通信：

产品仓库类：

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

//产品仓库
public class ProductStore {
	private boolean flag =  false;	
	private Lock lock = new ReentrantLock();  
	private Condition condition = lock.newCondition(); //得到condition   

	public boolean hasProduct(){//是否有产品
		return flag;
	}
	/**
	 * 生产产品
	 * @throws Exception
	 */
	public void makeProduct() throws Exception{
		lock.lock();  
		try {
			while(hasProduct()){//如果生产线程唤醒的还是生产线程，这个被唤醒的生产线程将继续wait
				condition.await();
			}
			Thread.sleep(300);
			flag = true;
			System.out.println(Thread.currentThread().getName()+":生产了一个产品");
			condition.signalAll();//唤醒所有线程
		} finally{
			lock.unlock();  
		}
	}
	/**
	 * 取走产品
	 * @throws Exception
	 */
	public void getProduct() throws Exception{
		lock.lock();  
		try {
			while(!hasProduct()){
				condition.await();
			}
			Thread.sleep(100);
			flag = false;
			System.out.println(Thread.currentThread().getName()+":取走一个产品");
			condition.signalAll();
		} finally {
			lock.unlock();
		}
	}
}

 makeProduct和getProduct方法不再使用synchronized修饰，所以使用Lock来控制同步，conditon的await()、singal()、singalAll()分别替换了Object的wait()、notify()和notifyAll()。

 

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////华丽分割线 

 

好了，发现上面代码中用的都是notifyAll()和singalAll()，如果把notifyAll改成notify，singalAll改成singal会有什么问题呢。因为notify只为在阻塞队列里面随机选一个线程唤醒，那么如果某个消费者拿到一个产品后，notify的任然是一个消费者的线程，那么完了，整个程序就锁住了。看下面的例子：

//产品仓库
public class ProductStore {
	private boolean flag = false;

	public boolean hasProduct() {// 是否有产品
		return flag;
	}

	/**
	 * 生产产品
	 * 
	 * @throws Exception
	 */
	public synchronized void makeProduct() throws Exception {
		while (hasProduct()) {// 如果生产线程唤醒的还是生产线程，这个被唤醒的生产线程将继续wait
			this.wait();
			System.out.println(Thread.currentThread().getName() + "被唤醒了");
		}
		Thread.sleep(300);
		flag = true;
		System.out.println(Thread.currentThread().getName() + ":生产了一个产品");
		this.notify();// 只唤醒一个线程

	}

	/**
	 * 取走产品
	 * 
	 * @throws Exception
	 */
	public synchronized void getProduct() throws Exception {
		while (!hasProduct()) {
			this.wait();
			System.out.println(Thread.currentThread().getName() + "被唤醒了");
		}
		Thread.sleep(100);
		flag = false;
		System.out.println(Thread.currentThread().getName() + ":取走一个产品");
		this.notify();// 只唤醒一个线程
	}
}

 执行结果：

生产者1:生产了一个产品
消费者2被唤醒了
消费者2:取走一个产品
消费者4被唤醒了
生产者5:生产了一个产品
消费者1被唤醒了
消费者1:取走一个产品
消费者3被唤醒了

 消费者1取完一个产品后，唤醒的是消费者3，所以整个程序死住了。如果用notifyAll的话当然就不会出现这种问题，因为所有的生产者线程被唤醒，并且最终肯定有一个能获得对象锁，从而生产一个产品使得程序顺利执行。

 

有没有更好的办法解决这个问题呢，如果消费者线程在拿走一个产品后只唤醒生产者线程岂不是更完美，用两个Condition就OK了。代码如下：

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

//产品仓库
public class ProductStore {
	private boolean flag =  false;	
	private Lock lock = new ReentrantLock();  
	private Condition producerCond = lock.newCondition(); //控制生产者的condition  
	private Condition consumerCond = lock.newCondition(); //控制消费者的condition

	public boolean hasProduct(){//是否有产品
		return flag;
	}
	/**
	 * 生产产品
	 * @throws Exception
	 */
	public void makeProduct() throws Exception{
		lock.lock();  
		try {
			while(hasProduct()){//还有产品，阻塞生产者
				producerCond.await();
			}
			Thread.sleep(300);
			flag = true;
			System.out.println(Thread.currentThread().getName()+":生产了一个产品");
			consumerCond.signal();//唤醒一个消费者
		} finally{
			lock.unlock();  
		}
	}
	/**
	 * 取走产品
	 * @throws Exception
	 */
	public void getProduct() throws Exception{
		lock.lock();  
		try {
			while(!hasProduct()){//没有产品，阻塞消费者
				consumerCond.await();
			}
			Thread.sleep(100);
			flag = false;
			System.out.println(Thread.currentThread().getName()+":取走一个产品");
			producerCond.signal();//唤醒一个生产者
		} finally {
			lock.unlock();
		}
	}
}

 

 执行结果如下

生产者2:生产了一个产品
消费者1:取走一个产品
生产者4:生产了一个产品
消费者3:取走一个产品
生产者1:生产了一个产品
消费者2:取走一个产品
生产者5:生产了一个产品
消费者5:取走一个产品
生产者3:生产了一个产品
消费者4:取走一个产品

当然，实现生产者消费者还可以用BlockingQueue，见另一篇博客。