java多线程设计模式之生产者与消费者

Java实现多线程生产者与消费者：

生产者线程负责生产产品，将产品保存到产品队列中，消费者负责从队列中取出产品消费。产品队列有一定的容量，当产品队列中没有产品时，消费者线程必须等待生产者线程生产产品，当产品队列中产品数量达到队列容量时，生产者等待消费者消费产品。
设计如下：
Product：产品类存储产品信息
ProductQueue：产品队列负责存储产品
ProducerThread：生产者线程负责生产产品
ConsumerThread：消费者线程负责消费产品
Main：程序入口

实现如下：
Product：产品类

package com.thread.product;

/**
 * 产品类
 * @author Administrator
 *
 */
public class Product {

	
	/**
	 * 产品名称
	 */
	private final String name;
	
	public Product(String name){
		this.name = name;
	}
	
	public String getName(){
		return name;
	}
	
	public String toString(){
		return "{product."+name+"}";
	}
	
}

ProductQueue：产品队列类

package com.thread.product;

import java.util.LinkedList;

/**
 * 产品队列类
 * @author Administrator
 *
 */
public class ProductQueue {

	/**
	 * 存储产品的队列
	 */
	private final LinkedList<Product> products = new LinkedList<Product>();
	/**
	 * 队列的容量大小
	 */
	private final int size;
	
	public ProductQueue(int size){
		this.size = size;
	}
	
	/**
	 * 向队列中添加产品，判断队列中产品大小是否达到了容量大小，
	 * 如果达到了当前生产线程等待，唤醒其他线程。
	 * 如果没有达到，当前生产线程向队列中添加产品。
	 * @param product
	 */
	public synchronized void putProduct(Product product){
		
		while(products.size() > size){//判断队列中产品是否达到队列容量
			try {
				System.out.println(Thread.currentThread().getName()+" {putProduct wait}");
				wait();//生产线程等待
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		products.addLast(product);//添加产品
		System.out.println(Thread.currentThread().getName()+" putProduct "+product);
		notifyAll();//唤醒其他等待线程
	};
	
	
	/**
	 * 获取队列中的产品，判断队列中是否存在产品，
	 * 如果没有则当前消费线程等待。
	 * 如果有则当前消费线程取出产品，唤醒其他等待线程。
	 * @return
	 */
	public synchronized Product getProduct(){
		
		while(products.size() <= 0){//判断是否队列中是否存在产品
			try {
				System.out.println(Thread.currentThread().getName()+" {getProduct wait}");
				wait();//消费线程等待
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		Product product = products.removeFirst();//取出产品
		System.out.println(Thread.currentThread().getName()+" getProduct "+product);
		notifyAll();//唤醒其他线程
		return product;
	};
	
}

ProducerThread：生产者类

package com.thread.product;

import java.util.Random;

/**
 * 生产产品线程类
 * @author Administrator
 *
 */
public class ProducerThread extends Thread{

	
	private ProductQueue productQueue;
	private Random random = new Random();
	
	public ProducerThread(String name, ProductQueue queue){
		super(name);
		this.productQueue = queue;
	}
	
	/**
	 * 循环生产产品
	 */
	public void run(){
		int count = 0;
		while(true){
			Product product = new Product("product."+count+"."+this.getName());
			productQueue.putProduct(product);
			count++;
			try {
				Thread.sleep(random.nextInt(1000));//随机休息
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
}

ConsumerThread：消费者类

package com.thread.product;

import java.util.Random;


/**
 * 消费者线程类
 * @author Administrator
 *
 */
public class ConsumerThread extends Thread{

	
	private ProductQueue productQueue;
	private Random random = new Random();
	
	public ConsumerThread(String name, ProductQueue queue){
		super(name);
		this.productQueue = queue;
	}
	
	/**
	 * 循环消费产品
	 */
	public void run(){
		while(true){
			productQueue.getProduct();
			try {
				Thread.sleep(random.nextInt(1400));//随机休息
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}

Main：main类

package com.thread.product;

/**
 * main类
 * @author Administrator
 *
 */
public class Main {

	/**
	 * 
	 * @param args
	 */
	public static void main(String[] args) {
		//初始化产品队列容量为5
		ProductQueue queue = new ProductQueue(5);
		//启动两个生产产品线程实例，a和b
		new ProducerThread("ProducerThread.a", queue).start();
		new ProducerThread("ProducerThread.b", queue).start();
		//启动两个消费者线程实例，a和b
		new ConsumerThread("ConsumerThread.a", queue).start();
		new ConsumerThread("ConsumerThread.b", queue).start();
	}
	
	
}

执行结果：

ProducerThread.b putProduct {product.product.0.ProducerThread.b}
ProducerThread.a putProduct {product.product.0.ProducerThread.a}
ConsumerThread.a getProduct {product.product.0.ProducerThread.b}
ConsumerThread.b getProduct {product.product.0.ProducerThread.a}
ConsumerThread.b {getProduct wait}
ProducerThread.a putProduct {product.product.1.ProducerThread.a}
ConsumerThread.b getProduct {product.product.1.ProducerThread.a}
ConsumerThread.b {getProduct wait}
ConsumerThread.a {getProduct wait}
ProducerThread.a putProduct {product.product.2.ProducerThread.a}
ConsumerThread.a getProduct {product.product.2.ProducerThread.a}
ConsumerThread.b {getProduct wait}
ConsumerThread.a {getProduct wait}
ProducerThread.b putProduct {product.product.1.ProducerThread.b}
ConsumerThread.a getProduct {product.product.1.ProducerThread.b}
ConsumerThread.b {getProduct wait}
ProducerThread.b putProduct {product.product.2.ProducerThread.b}
ConsumerThread.b getProduct {product.product.2.ProducerThread.b}
ConsumerThread.b {getProduct wait}
ProducerThread.b putProduct {product.product.3.ProducerThread.b}
ConsumerThread.b getProduct {product.product.3.ProducerThread.b}
ProducerThread.a putProduct {product.product.3.ProducerThread.a}
ProducerThread.b putProduct {product.product.4.ProducerThread.b}
ProducerThread.a putProduct {product.product.4.ProducerThread.a}
ProducerThread.a putProduct {product.product.5.ProducerThread.a}
ConsumerThread.a getProduct {product.product.3.ProducerThread.a}
ProducerThread.b putProduct {product.product.5.ProducerThread.b}
ConsumerThread.b getProduct {product.product.4.ProducerThread.b}
ConsumerThread.b getProduct {product.product.4.ProducerThread.a}
ProducerThread.a putProduct {product.product.6.ProducerThread.a}
ConsumerThread.a getProduct {product.product.5.ProducerThread.a}
ProducerThread.a putProduct {product.product.7.ProducerThread.a}
ProducerThread.b putProduct {product.product.6.ProducerThread.b}
ProducerThread.a putProduct {product.product.8.ProducerThread.a}
ProducerThread.b putProduct {product.product.7.ProducerThread.b}
ConsumerThread.b getProduct {product.product.5.ProducerThread.b}
ConsumerThread.a getProduct {product.product.6.ProducerThread.a}
ProducerThread.a putProduct {product.product.9.ProducerThread.a}
ProducerThread.b putProduct {product.product.8.ProducerThread.b}
ConsumerThread.b getProduct {product.product.7.ProducerThread.a}
ProducerThread.a putProduct {product.product.10.ProducerThread.a}
ConsumerThread.a getProduct {product.product.6.ProducerThread.b}
ProducerThread.a putProduct {product.product.11.ProducerThread.a}
ConsumerThread.b getProduct {product.product.8.ProducerThread.a}
ConsumerThread.b getProduct {product.product.7.ProducerThread.b}
ProducerThread.b putProduct {product.product.9.ProducerThread.b}
ProducerThread.a putProduct {product.product.12.ProducerThread.a}
ConsumerThread.a getProduct {product.product.9.ProducerThread.a}
ConsumerThread.b getProduct {product.product.8.ProducerThread.b}
ProducerThread.b putProduct {product.product.10.ProducerThread.b}
ProducerThread.a putProduct {product.product.13.ProducerThread.a}
ProducerThread.b {putProduct wait}
ConsumerThread.a getProduct {product.product.10.ProducerThread.a}
ProducerThread.b putProduct {product.product.11.ProducerThread.b}
ProducerThread.a {putProduct wait}
ConsumerThread.b getProduct {product.product.11.ProducerThread.a}
ProducerThread.a putProduct {product.product.14.ProducerThread.a}
ProducerThread.b {putProduct wait}
ConsumerThread.a getProduct {product.product.9.ProducerThread.b}
ProducerThread.b putProduct {product.product.12.ProducerThread.b}
ConsumerThread.b getProduct {product.product.12.ProducerThread.a}
ProducerThread.a putProduct {product.product.15.ProducerThread.a}
ProducerThread.a {putProduct wait}
。。。。。。