package book.thread.product;

public class Consumer extends Thread{
　　private Warehouse warehouse;//消费者获取产品的仓库
　　private boolean running = false;//是否需要结束线程的标志位
　　public Consumer(Warehouse warehouse,String name){
　　　　super(name);
　　　　this.warehouse = warehouse;
　　}
　　public void start(){
　　　　this.running = true;
　　　　super.start();
　　}
　　public void run(){
　　　　Product product;
　　　　try {
　　　　　　while(running){
　　　　　　　　//从仓库中获取产品
　　　　　　　　product = warehouse.getProduct();
　　　　　　　　sleep(500);
　　　　　　}
　　　　} catch (InterruptedException e) {
　　　　　　e.printStackTrace();
　　　　}
　　} 
　　//停止消费者线程
　　public void stopConsumer(){
　　　　synchronized(warehouse){
　　　　　　this.running = false;
　　　　　　warehouse.notifyAll();//通知等待仓库的线程
　　　　}
　　}
　　//消费者线程是否在运行
　　public boolean isRunning(){
　　　　return running;
　　}
}

 

package book.thread.product;

public class Producer extends Thread{
　  private Warehouse warehouse;//生产者存储产品的仓库
　　private static int produceName = 0;//产品的名字
　　private boolean running = false;//是否需要结束线程的标志位

　　public Producer(Warehouse warehouse,String name){
　　　　super(name);
　　　　this.warehouse = warehouse;
　　}
　　public void start(){
　　　　this.running = true;
　　　　super.start();
　　}
　　public void run(){
　　　　Product product;
　　　　//生产并存储产品
　　　　try {
　　　　while(running){
　　　　　　product = new Product((++produceName)+"");
　　　　　　this.warehouse.storageProduct(product);
　　　　　　sleep(300);
　　　　　　}
　　　　} catch (InterruptedException e) {
　　　　　　e.printStackTrace();
　　　　}
　　}
　　//停止生产者线程
　　public void stopProducer(){
　　　　synchronized(warehouse){
　　　　　　this.running = false;
　　　　　　//通知等待仓库的线程
　　　　　　warehouse.notifyAll();
　　　　}
　　}
　　//生产者线程是否在运行
　　public boolean isRunning(){
　　　　return running;
　　}
}

 

package book.thread.product;

public class Product {
　　private String name;//产品名
　　public Product(String name){
　　　　this.name = name;
　　}
　　public String toString(){
　　　　return "Product-"+name;
　　}
}

 

package book.thread.product;

//产品的仓库类，内部采用数组来表示循环队列，以存放产品
public class Warehouse {
　　private static int CAPACITY = 11;//仓库的容量
　　private Product[] products;//仓库里的产品
　　//[front,rear]区间的产品未被消费
　　private int front = 0;//当前仓库中第一个未被消费的产品的下标
　　private int rear = 0;//仓库中最后一个未被消费的产品下标加1
　　public Warehouse(){
　　　　this.products = new Product[CAPACITY];
　　}
　　public Warehouse(int capacity){
　　　　this();
　　　　if(capacity > 0){
　　　　　　CAPACITY = capacity +1;
　　　　　　this.products = new Product[CAPACITY];
　　　　}
　　}

　　//从仓库获取一个产品
　　public Product getProduct() throws InterruptedException{
　　　　synchronized(this){
　　　　　　boolean consumerRunning = true;//标志消费者线程是否还在运行
　　　　　　Thread currentThread = Thread.currentThread();//获取当前线程
　　　　　　if(currentThread instanceof Consumer){
　　　　　　　　consumerRunning = ((Consumer)currentThread).isRunning();
　　　　　　}else{
　　　　　　　　return null;//非消费者不能获取产品
　　　　　　}
　　　　　　//若消费者线程在运行中，但仓库中没有产品了，则消费者线程继续等待
　　　　　　while((front==rear) && consumerRunning){
　　　　　　　　wait();
　　　　　　　　consumerRunning = ((Consumer)currentThread).isRunning();
　　　　　　}
　　　　　　//如果消费者线程已经停止运行，则退出该方法，取消获取产品
　　　　　　if(!consumerRunning){
　　　　　　　　return null;
　　　　　　}
　　　　　　//获取当前未被消费的第一个产品
　　　　　　Product product = products[front];
　　　　　　System.out.println("Consumer[" + currentThread.getName()+"] getProduct:"+product);
　　　　　　//将当前未被消费产品的下标后移一位，如果到了数组末尾，则移动到首部
　　　　　　front = (front+1+CAPACITY)%CAPACITY;
　　　　　　System.out.println("仓库中还没有被消费的产品数量："+(rear+CAPACITY-front)%CAPACITY);
　　　　　　//通知其他等待线程
　　　　　　notify();
　　　　　　return product;
　　　　}
　　}
　　//向仓库存储一个产品
　　public void storageProduct(Product product) throws InterruptedException{
　　synchronized(this){
　　　　boolean producerRunning = true;//标志生产者线程是否在运行
　　　　Thread currentThread = Thread.currentThread();
　　　　if(currentThread instanceof Producer){
　　　　　　producerRunning = ((Producer)currentThread).isRunning();
　　　　}else{
　　　　　　return;
　　　　}
　　　　//如果最后一个未被消费的产品与第一个未被消费的产品的下标紧挨着，则说明没有存储空间了。
　　　　//如果没有存储空间了，而生产者线程还在运行，则生产者线程等待仓库释放产品
　　　　while(((rear+1)%CAPACITY == front) && producerRunning){
　　　　　　wait();
　　　　　　producerRunning = ((Producer)currentThread).isRunning();
　　　　}
　　　　//如果生产线程已经停止运行了，则停止产品的存储
　　　　if(!producerRunning){
　　　　　　return;
　　　　}
　　　　//保存产品到仓库
　　　　products[rear] = product;
　　　　System.out.println("Producer[" + Thread.currentThread().getName()+"] storageProduct:" + product);
　　　　//将rear下标循环后移一位
　　　　rear = (rear + 1)%CAPACITY;
　　　　System.out.println("仓库中还没有被消费的产品数量："+(rear + CAPACITY -front)%CAPACITY);
　　　　notify();
　　　　}
　　}
}

 

package book.thread.product;

public class TestProduct {
　　public static void main(String[] args) {
　　　　Warehouse warehouse = new Warehouse(10);//建立一个仓库，容量为10
　　　　//建立生产者线程和消费者
　　　　Producer producers1 = new Producer(warehouse,"producer-1");
　　　　Producer producers2 = new Producer(warehouse,"producer-2");
　　　　Producer producers3 = new Producer(warehouse,"producer-3");
　　　　Consumer consumer1 = new Consumer(warehouse,"consumer-1");
　　　　Consumer consumer2 = new Consumer(warehouse,"consumer-2");
　　　　Consumer consumer3 = new Consumer(warehouse,"consumer-3");
　　　　Consumer consumer4 = new Consumer(warehouse,"consumer-4");
　　　　//启动生产者线程和消费者线程
　　　　producers1.start();
　　　　producers2.start();
　　　　consumer1.start();
　　　　producers3.start();
　　　　consumer2.start();
　　　　consumer3.start();
　　　　consumer4.start();
　　　　//让生产者/消费者程序运行1600ms
　　　　try {
　　　　　　Thread.sleep(1600);
　　　　} catch (InterruptedException e) {
　　　　　　e.printStackTrace();
　　　　}
　　　　//停止消费者线程
　　　　producers1.stopProducer();
　　　　consumer1.stopConsumer();
　　　　producers2.stopProducer();
　　　　consumer2.stopConsumer();
　　　　producers3.stopProducer();
　　　　consumer3.stopConsumer();
　　　　consumer4.stopConsumer();
　　}
}

Producer[producer-1] storageProduct:Product-1
仓库中还没有被消费的产品数量：1
Consumer[consumer-2] getProduct:Product-1
仓库中还没有被消费的产品数量：0
Producer[producer-3] storageProduct:Product-3
仓库中还没有被消费的产品数量：1
Producer[producer-2] storageProduct:Product-2
仓库中还没有被消费的产品数量：2
Consumer[consumer-3] getProduct:Product-3
仓库中还没有被消费的产品数量：1
Consumer[consumer-1] getProduct:Product-2
仓库中还没有被消费的产品数量：0
Producer[producer-1] storageProduct:Product-4
仓库中还没有被消费的产品数量：1
Consumer[consumer-4] getProduct:Product-4
仓库中还没有被消费的产品数量：0
Producer[producer-3] storageProduct:Product-6
仓库中还没有被消费的产品数量：1
Producer[producer-2] storageProduct:Product-5
仓库中还没有被消费的产品数量：2
Consumer[consumer-1] getProduct:Product-6
仓库中还没有被消费的产品数量：1
Consumer[consumer-2] getProduct:Product-5
仓库中还没有被消费的产品数量：0
Producer[producer-1] storageProduct:Product-7
仓库中还没有被消费的产品数量：1
Consumer[consumer-3] getProduct:Product-7
仓库中还没有被消费的产品数量：0
Producer[producer-3] storageProduct:Product-8
仓库中还没有被消费的产品数量：1
Producer[producer-2] storageProduct:Product-9
仓库中还没有被消费的产品数量：2
Consumer[consumer-4] getProduct:Product-8
仓库中还没有被消费的产品数量：1
Producer[producer-1] storageProduct:Product-10
仓库中还没有被消费的产品数量：2
Producer[producer-3] storageProduct:Product-11
仓库中还没有被消费的产品数量：3
Producer[producer-2] storageProduct:Product-12
仓库中还没有被消费的产品数量：4
Consumer[consumer-1] getProduct:Product-9
仓库中还没有被消费的产品数量：3
Consumer[consumer-2] getProduct:Product-10
仓库中还没有被消费的产品数量：2
Consumer[consumer-3] getProduct:Product-11
仓库中还没有被消费的产品数量：1
Producer[producer-3] storageProduct:Product-13
仓库中还没有被消费的产品数量：2
Producer[producer-1] storageProduct:Product-14
仓库中还没有被消费的产品数量：3
Producer[producer-2] storageProduct:Product-15
仓库中还没有被消费的产品数量：4
Consumer[consumer-4] getProduct:Product-12
仓库中还没有被消费的产品数量：3
Consumer[consumer-1] getProduct:Product-13
仓库中还没有被消费的产品数量：2
Consumer[consumer-2] getProduct:Product-14
仓库中还没有被消费的产品数量：1
Producer[producer-1] storageProduct:Product-16
仓库中还没有被消费的产品数量：2
Producer[producer-3] storageProduct:Product-17
仓库中还没有被消费的产品数量：3
Producer[producer-2] storageProduct:Product-18
仓库中还没有被消费的产品数量：4