生产者消费者模型

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

生产者消费者实现

• 生产者消费者示例
• 线程池中可以实现生产者消费者模型
• 这里通过简单的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() 方法唤醒同步锁上的所有线程,包括生产者线程,这样就可以通知生产者进行生产

链接：https://juejin.cn/post/7034728443480637453