多线程协作-生产者和消费者

public class Product{
    private String name; 
    
    public Product(String name){
        this.name = name;
        }
    
    //重写了Product类的toString方法，自定义Product类的Feild展示为字符串的格式
    public String toString(){
        return "Product-"+this.name;
        }    
    }

import java.lang.Thread;

//采用默认的default访问控制符，仅限当前类的同一个包中类访问
class Producter extends Thread{
    //生产者生产产品，需要放入仓库，还需要区分生产的产品
    private WareHouse wareHouse ;
    private static int productName ;
    //线程是否运行标记
    private boolean running = false;
    
    public Producter(WareHouse wareHouse, String producterNmae){
        //继承Thread的构造函数，也就继承了Thread对象的功能
        super(producterNmae);
        this.wareHouse = wareHouse;
        }
    
    public void start(){
        this.running = true;
        super.start();
        }
        
    public void run(){
        Product product;
        try{
            while(this.running){
            //(int类型的productName)+("")，合并的结果为String类型
            product = new Product((++productName)+"");
            this.wareHouse.storageProduct(product);
            //等待wait状态的线程执行完synchronized代码块，当前线程再释放对象锁
            //主要是为了暂停当前线程，把cpu片段让出给其他线程，减缓当前线程的执行
            Thread.sleep(10);
            }
        }catch(InterruptedException e){
            e.printStackTrace();
            }
        }
    
    //将当前生产者线程停止
    public void stopProduct(){
        //线程同步的资源对象是仓库
        synchronized (wareHouse){
            this.running = false;
            System.out.println("Producter["+Thread.currentThread().getName()+"]停止了！");
            //将所有等待WareHouse对象的线程唤醒，告知当前线程停止
            this.wareHouse.notifyAll();
//            this.wareHouse.notify();
            }
        }
        
    public boolean isRunning(){
        return this.running;
        }    
    }

import java.lang.Thread;

//采用默认的default访问控制符，仅限当前类的同一个包中类访问
class Customer extends Thread{
    private WareHouse wareHouse;
    private boolean running = false ;
    
    public Customer(WareHouse wareHouse, String customerName){
        super(customerName);
        this.wareHouse = wareHouse ;
        }
        
    public void start(){
        this.running = true ;
        super.start();
        }
        
    public void run(){
        try{
            while(this.running){
            this.wareHouse.getProduct();
            Thread.sleep(1000);
            }
        }catch(InterruptedException e){
            e.printStackTrace();
            }
        }
        
    public void stopCustomer(){
        synchronized(this.wareHouse){
            this.running = false ;
            System.out.println("Customer["+Thread.currentThread().getName()+"]停止了！");
            //将所有等待WareHouse对象的线程唤醒，告知当前线程停止
            this.wareHouse.notifyAll();
//            this.wareHouse.notify();
            }
        }
        
    public boolean isRunning(){
        return this.running;
        }
    }

import java.lang.Thread;


//采用默认的default访问控制符，仅限当前类的同一个包中类访问
class WareHouse{
    //默认仓库容量
    private static int CAPACITY = 11 ;
    //存放产品的循环数组，生产的新产品紧跟现有产品后存放，消费的产品从现有产品首个开始获取
    private Product[] products;
    
    private int front = 0 ;        //仓库中第一个未消费产品下标
    private int rear = 0 ;        //仓库中最后一个未消费产品下标
    
    public WareHouse(){
        this.products = new Product[CAPACITY];
        }
        
    public WareHouse(int capacity){
        //初始还是默认仓库容量
        this();
        if(capacity > 0){
            CAPACITY = capacity + 1 ;
            this.products = new Product[CAPACITY];
            }
        }
        
    public void storageProduct(Product product){
        synchronized(this){
            boolean producerRunning = true ;
            Thread currentThread = Thread.currentThread();
            if(currentThread instanceof Producter){
                producerRunning = ((Producter)currentThread).isRunning();
            }else{return;}
                
            //保留最后一个可以存储的位置并等待销售唤醒，唤醒后如果还需要生产，但仅剩最后一个可以存储的位置则循环等待
            while(((rear+1) % CAPACITY) == front && producerRunning){
                System.out.println("Producter["+currentThread.getName()+"]满仓了！");
                try{
                    this.wait();
                }catch(InterruptedException e){
                    e.printStackTrace();
                    }
                //唤醒后该生产者是否还需要生产
                producerRunning = ((Producter)currentThread).isRunning();
                System.out.println("Producter["+Thread.currentThread().getName()+"]"+ (producerRunning==true ? "正在运行":"停止了"));
                }
            //该生产者不需要生产，则结束
            if(!producerRunning){return;}
            this.products[rear] = product ;
            System.out.println("Producter["+currentThread.getName()+"] storageProduct :"+product.toString());
            //下一待存储下标循环下移
            rear = (rear + 1) %  CAPACITY;
            System.out.println("仓库中剩余未销售的产品数量："+(rear+CAPACITY-front) % CAPACITY);
            //随机唤醒等待仓库资源对象的一个线程，包括生产者线程和消费者线程
            this.notify();
            }
        }
        
    public Product getProduct(){
        synchronized(this){
            boolean customerRunning = false ;
            Thread customerThread = Thread.currentThread();
            if(customerThread instanceof Customer){
                customerRunning = ((Customer)customerThread).isRunning();
            }else{return null;}
            while(front==rear && customerRunning){
                System.out.println("Customer["+customerThread.getName()+"]仓库空了！");
                try{
                    this.wait();
                }catch(InterruptedException e){
                    e.printStackTrace();
                    }
                customerRunning = ((Customer)customerThread).isRunning();
                System.out.println("Customer["+Thread.currentThread().getName()+"]"+ (customerRunning==true ? "正在运行":"停止了"));
                }
            if(!customerRunning){return null;}
            Product product = this.products[front];
            System.out.println("Customer["+customerThread.getName()+"] getProduct : "+product.toString());
            //下移待销售下标循环下移
            front = (front + 1 + CAPACITY) % CAPACITY;
            System.out.println("仓库中剩余未销售的产品数量："+(rear+CAPACITY-front) % CAPACITY);
            //随机唤醒等待仓库资源对象的一个线程，包括生产者线程和消费者线程
            this.notify();
            return product;
            }
        }    
    }

import java.lang.Thread;

public class ProductTest{
    public static void main(String[] args){
        WareHouse wareHouse = new WareHouse(2);
        Producter producter1 = new Producter(wareHouse, "producter-1");
        Producter producter2 = new Producter(wareHouse, "producter-2");
        Producter producter3 = new Producter(wareHouse, "producter-3");
        Customer customer1 = new Customer(wareHouse, "customer-1");
        Customer customer2 = new Customer(wareHouse, "customer-2");
        Customer customer3 = new Customer(wareHouse, "customer-3");
        Customer customer4 = new Customer(wareHouse, "customer-4");
//        producter1.start();
//        producter2.start();
//        producter3.start();
//        customer1.start();
//        customer2.start();
//        customer3.start();
//        customer4.start();
        
        producter1.start();
        customer1.start();
        
        producter2.start();
//        customer2.start();
        
        producter3.start();
//        customer3.start();
//        customer4.start();
        try{
            Thread.sleep(1600);
        }catch(InterruptedException e){
            e.printStackTrace();
            }
//        producter1.stopProduct();
//        producter2.stopProduct();
//        producter3.stopProduct();
//        customer1.stopCustomer();
//        customer2.stopCustomer();
//        customer3.stopCustomer();
//        customer4.stopCustomer();
        
        producter1.stopProduct();
        customer1.stopCustomer();
        
        producter2.stopProduct();
//        customer2.stopCustomer();
        
        producter3.stopProduct();
//        customer3.stopCustomer();
//        customer4.stopCustomer();
        }
    }

运行结果：