Java多线程Day11-多线程之生产者消费者

女爷i 2022-12-07 11:44 341阅读 0赞

### 生产者消费者问题 ###

*  生产者消费者模型
 *  生产者消费者实现

# 生产者消费者模型 #

*  生产者消费者模型包括**生产者,消费者,仓库,产品:**
    
     *  生产者在仓库未满时才能生产,仓库满时停止生产
     *  消费者在仓库有产品时才能消费,仓库空时等待
     *  当消费者发现仓库为空,没有产品可以消费时,通知生产者生产
     *  生产者在生产出可以消费的产品时,通知等待的消费者消费

# 生产者消费者实现 #

*  [生产者消费者示例][Link 1]
 *  线程池中可以实现生产者消费者模型
 *  这里通过简单的**wait()** 和**notify()** 方式实现生产者消费者模型

class Depot { 
    	/** * 仓库容量 */
    	private int capacity;
    	/** * 仓库的实际容量 */
    	private int size = 0;
    
    	public Depot(int capacity) { 
    		this.capacity = capacity;
    		this.size = 0;
    	}
    
    	/** * 生产方法 * * @param val 需要生产的数量 */
    	public synchronized void produce(int val) { 
    		try { 
    			// want表示想要生产的数量.因为可能想要生产的数量太多,需要进行多次生产
    			int want = val;
    			while (want > 0) { 
    				// 如果库存已满时,等待消费者消费产品
    				while (size >= capacity) { 
    					wait();
    				}
    				/* * 获取实际生产的数量,即新增的产品数量 * - 如果库存和想要生产的数量之和大于仓库容量时,则实际生产数量等于总的容量减去当前容量 * - 如果库存和想要生产的数量之和小于等于仓库容量,则实际生产数量等于想要生产的数量 */
    				 int increment = (size + want) > capacity ? (capacity - size) : want;
    				 size += increment;
    				 want -= increment;
    				 System.out.printf("%sproduce(%3d) --> want=%3d, increment=%3d, size=%3d\n", Thread.currentThread().getName(), val, want, increment, size);
    			 	// 生产完成通知消费者消费
    				 notifyAll();
    			}		
    		} catch (InterruptedException e) { 
    			// 不需要进行处理
    		}
    	}
    	
    	/** * 消费方法 * * @param val 消费的数量 */
    	public synchronized void consume(int val) { 
    		try { 
    			// want表示客户消费的数量.因为想要消费的数量太多,需要进行多次消费
    			int want = val;
    			while (want > 0) { 
    				// 如果库存为空时,等待生产者生产产品
    				while (size <= 0) { 
    					wait();
    				}
    				/* * 获取实际消费的数量,即减少的数量 * - 如果库存的数量小于想要消费的数量,则实际消费数量等于库存量 * - 如果库存的数量等于或者大于想要消费的数量,则实际消费数量等于想要消费的数量 */
    				 int decrease = (size < want) ? size : want;
    				 size -= decrease;
    				 want -= decrease;
    				 System.out.printf("%s consume(%3d) <-- want=%3d, decrease=%3d, size=%3d\n");
    				 // 消费完成通知生产者生产
    				 notifyAll();
    			}
    		} catch (InterruptedException e) { 
    			// 不需要进行处理
    		}
    	}	
    }
    
    class Producer { 
    	private Depot depot;
    
    	public Producer(Depot depot) { 
    		this.depot = depot;
    	}
    
    	/** * 新建一个生产者线程,用于生产产品 * * @param val 需要生产的数量 */
    	public void produce(final int val) { 
    		new Thread () { 
    			public void run() { 
    				depot.produce(val);
    			}
    		}.start();
    	}
    }
    
    class Consumer { 
    	private Depot depot;
    
    	public Consumer(Depot depot) { 
    		this.depot = depot;
    	}
    
    	/** * 新建一个消费者,用于消费产品 * * @param val 需要消费的数量 */
    	public void consume(final int val) { 
    		new Thread() { 
    			public void run() { 
    				depot.consume(val);
    			}
    		}.start();
    	}
    }
    
    class ProducerConsumerTest { 
    	public static void main(String[] args) { 
    		Depot depot = new Depot(100);
    		Producer producer = new Producer(depot);
    		Consumer consumer = new Consumer(depot);
    
    		producer.produce(60);
    		producer.produce(120);
    		consumer.consume(90);
    		consumer.consume(150);
    		producer.produce(110);
    	}
    }

Thread-0 produce( 60) --> want=  0, increment= 60, size= 60
    Thread-4 produce(110) --> want= 70, increment= 40, size=100
    Thread-2 consume( 90) <-- want=  0, decrease= 90, size= 10
    Thread-3 consume(150) <-- want=140, decrease= 10, size=  0
    Thread-1 produce(120) --> want= 20, increment=100, size=100
    Thread-3 consume(150) <-- want= 40, decrease=100, size=  0
    Thread-4 produce(110) --> want=  0, increment= 70, size= 70
    Thread-3 consume(150) <-- want=  0, decrease= 40, size= 30
    Thread-1 produce(120) --> want=  0, increment= 20, size= 50

*  **Producer**是生产者类,与仓库**depot**关联. 当调用生产者的**produce()** 方法时,会新建一个线程并向仓库**depot**中生产产品
 *  **Consumer**是消费者类,与仓库**depot**关联. 当调用消费者的**consume()** 方法时,会新建一个线程并消费仓库**depot**中的商品
 *  **Depot**是仓库类,仓库类**Depot**中记录仓库的容量**capacity,** 以及仓库中当前产品的数目**size :**
    
     *  仓库类**Depot**的生产方法**produce()** 和消费方法**consume()** 方法都是**synchronized**方法,进入**synchronized**方法体,意味着这个线程获取了该仓库对象的同步锁
     *  意味着,同一时间生产者和消费者线程只能有一个能运行. 通过同步锁,实现了对仓库的互斥访问
     *  **produce()生产方法:**
        
         *  当仓库满时,生产者线程等待,需要等待消费者消费产品后,生产者线程才能生产
         *  生产者线程生产完产品之后,会通过**notifyAll()** 方法唤醒同步锁上的所有线程,包括消费者线程,这样就可以通知消费者进行消费
     *  **consume()消费方法:**
        
         *  当仓库为空时,消费者线程等待,需要等待生产者生产产品之后,消费者线程才能消费
         *  消费者线程消费完产品之后,会通过**notifyAll()** 方法唤醒同步锁上的所有线程,包括生产者线程,这样就可以通知生产者进行生产

[Link 1]: https://github.com/siege-projects/chova-siege