生产者消费者问题

生产者消费者问题

 

生产者消费者问题是研究多线程程序时绕不开的问题，它的描述是有一块生产者和消费者共享的有界缓冲区，生产者往缓冲区放入产品，消费者从缓冲区取走产品，这个过程可以无休止的执行，不能因缓冲区满生产者放不进产品而终止，也不能因缓冲区空消费者无产品可取而终止。

解决生产者消费者问题的方法有两种，一种是采用某种机制保持生产者和消费者之间的同步，一种是在生产者和消费者之间建立一个管道。前一种有较高的效率并且可控制性较好，比较常用，后一种由于管道缓冲区不易控制及被传输数据对象不易封装等原因，比较少用。

同步问题的核心在于，CPU是按时间片轮询的方式执行程序，我们无法知道某一个线程是否被执行、是否被抢占、是否结束等，因此生产者完全可能当缓冲区已满的时候还在放入产品，消费者也完全可能当缓冲区为空时还在取出产品。

现在同步问题的解决方法一般是采用信号或者加锁机制，即生产者线程当缓冲区已满时放弃自己的执行权，进入等待状态，并通知消费者线程执行。消费者线程当缓冲区已空时放弃自己的执行权，进入等待状态，并通知生产者线程执行。这样一来就保持了线程的同步，并避免了线程间互相等待而进入死锁状态。

JAVA语言提供了独立于平台的线程机制，保持了”write once, run anywhere”的特色。同时也提供了对同步机制的良好支持。在JAVA中，一共有四种方法支持同步，其中三个是同步方法，一个是管道方法。

1. 方法wait()/notify()。
2. 方法await()/signal()。
3. 阻塞队列方法BlockingQueue。

 

1. 方法wait()/notify()

wait()和notify()是根类Object的两个方法，也就意味着所有的JAVA类都会具有这个两个方法，为什么会被这样设计呢？我们可以认为所有的对象默认都具有一个锁，虽然我们看不到，也没有办法直接操作，但它是存在的。

• wait()方法表示：当缓冲区已满或空时，生产者或消费者线程停止自己的执行，放弃锁，使自己处于等待状态，让另一个线程开始执行；
• notify()方法表示：当生产者或消费者对缓冲区放入或取出一个产品时，向另一个线程发出可执行通知。

下面是一个例子代码：

 

import java.util.LinkedList;

public class ProducerConsumer {
	private LinkedList<Product> buffer;
	
	public ProducerConsumer() {
		buffer = new LinkedList<Product>();
	}
	
	public void start() {
		Producer p = new Producer(buffer);
		Consumer c = new Consumer(buffer);
		p.start();
		c.start();
	}
	
	public static void main(String[] args) {
		ProducerConsumer pc = new ProducerConsumer();
		pc.start();
	}
}

class Producer extends Thread {
	private LinkedList<Product> buffer;
	private final int MAX = 10;
	private int count = 0;
	
	public Producer(LinkedList<Product> buffer) {
		this.buffer = buffer;
	}
	
	public void run() {
		while (true) {
			try {
				synchronized (buffer) {
					while (buffer.size() == MAX) {
						buffer.wait();
					}
					Product p = new Product(++count);
					if (buffer.add(p)) {
						System.out.println("make product " + p.id);
						buffer.notify();
					}
				}
			} catch (InterruptedException ie) {
				System.out.println("producer is interrupted");
			}
		}
	}
}

class Consumer extends Thread {
	private LinkedList<Product> buffer;
	
	public Consumer(LinkedList<Product> buffer) {
		this.buffer = buffer;
	}
	
	public void run() {
		while (true) {
			try {
				synchronized (buffer) {
					while (buffer.size() == 0) {
						buffer.wait();
					}
					Product p = buffer.remove();
					System.out.println("consume product " + p.id);
					buffer.notify();
				}
			} catch (InterruptedException ie) {
				System.out.println("Consumer is interrupted");
			}
		}
	}
}

class Product {
	int id;
	
	public Product(int id) {
		this.id = id;
	}
}

 

 

2. 方法await()/signal()

在JDK5.0以后，JAVA提供了新的更加健壮的线程处理机制，包括了同步、锁定、线程池等等，它们可以实现更小粒度上的控制。await()和signal()就是其中用来做同步的两种方法，它们的功能基本上和 wait()/notify()相同，完全可以取代它们，但是它们和新引入的锁定机制Lock直接挂钩，具有更大的灵活性。

下面是一个例子代码：

 

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

public class ProducerConsumer2 {
	private LinkedList<Product> buffer;
	
	public ProducerConsumer2() {
		buffer = new LinkedList<Product>();
	}
	
	public void start() {
		Producer2 p = new Producer2(buffer);
		Consumer2 c = new Consumer2(buffer);
		p.start();
		c.start();
	}
	
	public static void main(String[] args) {
		ProducerConsumer2 pc = new ProducerConsumer2();
		pc.start();
	}
}

class Producer2 extends Thread {
	private LinkedList<Product> buffer;
	private final int MAX = 10;
	private int count = 0;
	private final Lock lock = new ReentrantLock();
	private final Condition full = lock.newCondition();
	private final Condition empty = lock.newCondition();
	
	public Producer2(LinkedList<Product> buffer) {
		this.buffer = buffer;
	}
	
	public void run() {
		while (true) {
			lock.lock();
			try {
				while (buffer.size() == MAX) {
					full.await();
				}
				Product p = new Product(++count);
				if (buffer.add(p)) {
					System.out.println("make product " + p.id);
					empty.signal();
				}
			} catch (InterruptedException ie) {
				System.out.println("producer is interrupted");
			} finally {
				lock.unlock();
			}
		}
	}
}

class Consumer2 extends Thread {
	private LinkedList<Product> buffer;
	private final Lock lock = new ReentrantLock();
	private final Condition full = lock.newCondition();
	private final Condition empty = lock.newCondition();
	
	public Consumer2(LinkedList<Product> buffer) {
		this.buffer = buffer;
	}
	
	public void run() {
		while (true) {
			lock.lock();
			try {
				while (buffer.size() == 0) {
					empty.await();
				}
				Product p = buffer.remove();
				System.out.println("consume product " + p.id);
				full.signal();
			} catch (InterruptedException ie) {
				System.out.println("Consumer is interrupted");
			} finally {
				lock.unlock();
			}
		}
	}
}

 

3. 阻塞队列方法BlockingQueue

BlockingQueue也是JDK5.0的一部分，它是一个已经在内部实现了同步的队列，实现方式采用的是我们的第2种await()/signal()方法。它可以在生成对象时指定容量大小。

它用于阻塞操作的是put()和take()方法。

• put()方法类似于我们上面的生产者线程，容量最大时，自动阻塞。
• take()方法类似于我们上面的消费者线程，容量为0时，自动阻塞。

 

下面是一个例子代码：

 

import java.util.concurrent.LinkedBlockingQueue;

public class ProducerConsumer3 {
	private LinkedBlockingQueue<Product> buffer;
	
	public ProducerConsumer3() {
		buffer = new LinkedBlockingQueue<Product>(10);
	}
	
	public void start() {
		Producer3 p = new Producer3(buffer);
		Consumer3 c = new Consumer3(buffer);
		p.start();
		c.start();
	}
	
	public static void main(String[] args) {
		ProducerConsumer3 pc = new ProducerConsumer3();
		pc.start();
	}
}

class Producer3 extends Thread {
	private LinkedBlockingQueue<Product> buffer;
	private final int MAX = 10;
	private int count = 0;
	
	public Producer3(LinkedBlockingQueue<Product> buffer) {
		this.buffer = buffer;
	}
	
	public void run() {
		while (true) {
			try {
				if (buffer.size() == MAX) {
					System.out.println("is full");
				}
				Product p = new Product(++count);
				buffer.put(p);
				System.out.println("make product " + p.id);
			} catch (InterruptedException ie) {
				System.out.println("producer is interrupted");
			}
		}
	}
}

class Consumer3 extends Thread {
	private LinkedBlockingQueue<Product> buffer;
	
	public Consumer3(LinkedBlockingQueue<Product> buffer) {
		this.buffer = buffer;
	}
	
	public void run() {
		while (true) {
			try {
				if (buffer.size() == 0) {
					System.out.println("is empty");
				}
				Product p = buffer.take();
				System.out.println("consume product " + p.id);
			} catch (InterruptedException ie) {
				System.out.println("producer is interrupted");
			}
		}
	}
}