Java并发编程与技术内幕:消费者-生产者模式研究

ゞ浴缸里的玫瑰 2021-09-11 07:22 333阅读 0赞

消费者=生产者模式是Java并发编程中一个很好的应用实例，一般要求会如下：

1、生产者仅仅在仓储未满时候生产，仓满则停止生产。  
2、消费者仅仅在仓储有产品时候才能消费，仓空则等待。  
3、当消费者发现仓储没产品可消费时候会通知生产者生产。  
4、生产者在生产出可消费产品时候，应该通知等待的消费者去消费。

光看要求的话，应该不复杂。但是真正编程起来时，还是要考虑不少东西，本文将主要以四种方法来实现生产者-消费者模式：

一、wait/notify实现

这是一种比较常规的实现方法，但是代码个人感觉有点复杂。并且缺点是不能完全实现仓储有产品马上就消费。代码如下：

package com.lin;
    
    import java.util.LinkedList;
    import java.util.List;
    
    class Storehouse {
    	// 仓库的容量
    	private int capacity;
    	// object当成是生产的商品
    	private List<Object> list = new LinkedList<Object>();
    
    	public Storehouse(int capacity) {
    		this.capacity = capacity;
    		System.out.println("当前仓库产品数量：" + list.size());
    	}
    
    	public int getCapacity() {
    		return capacity;
    	}
    
    	public void setCapacity(int capacity) {
    		this.capacity = capacity;
    	}
    
    	/**
    	 * 生产的方法
    	 * 
    	 * @throws InterruptedException
    	 */
    	public void produrce(int num) throws InterruptedException {
    		// 同步方法
    		synchronized (list) {
    			// 仓库还未满，且再生产num个产品不会超过仓库容量时可以生产产品
    			while (list.size() + num > this.capacity) {
    				// 仓库已满，或者放不下
    				System.out.println("【仓库已无法再生产：" + num + "个产品】" + "当前仓库产品数量：" + list.size());
    				list.wait();
    
    			}
    
    			System.out.println("【仓库还未满，生产：" + num + "个产品没有问题】" + "当前仓库产品数量：" + list.size());
    			for (int i = 0; i < num; i++) {
    				list.add(new Object());
    			}
    			list.notifyAll();
    		}
    	}
    
    	/**
    	 * 消费
    	 * 
    	 * @param num
    	 * @throws InterruptedException
    	 */
    	public void consumer(int num) throws InterruptedException {
    		// 同步方法
    		synchronized (list) {
    			// 仓库有没有num个产品可消费
    			while (list.size() < num) {
    				System.out.println("【仓库没有：" + num + "个产品可消费】" + "当前仓库产品数量：" + list.size());
    				list.wait();
    			}
    			System.out.println("【仓库有：" + num + "个产品可消费】" + "当前仓库产品数量：" + list.size());
    			for (int i = 0; i < num; i++) {
    				list.remove(0);
    			}
    			list.notifyAll();
    		}
    	}
    }
    
    	class ProducerThread extends Thread {
    		// 每次生产的产品数量
    		private int num;
    
    		// 所在放置的仓库
    		private Storehouse storehouse;
    
    		// 构造函数，设置仓库
    		public ProducerThread(Storehouse storehouse, int num) {
    			this.storehouse = storehouse;
    			this.num = num;
    		}
    
    		public void run() {
    			try {
    				storehouse.produrce(num);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    
    	class ConsumerThread extends Thread {
    		// 每次生产的产品数量
    		private int num;
    
    		// 所在放置的仓库
    		private Storehouse storehouse;
    
    		// 构造函数，设置仓库
    		public ConsumerThread(Storehouse storehouse, int num) {
    			this.storehouse = storehouse;
    			this.num = num;
    		}
    
    		public void run() {
    			try {
    				storehouse.consumer(num);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    
    	}
    
    public class Test1 {
    
    	public static void main(String[] args) {
    
    		// 仓库对象
    		Storehouse storage = new Storehouse(1000);
    
    		// 生产者对象
    		ProducerThread p1 = new ProducerThread(storage, 200);
    		ProducerThread p2 = new ProducerThread(storage, 200);
    		ProducerThread p3 = new ProducerThread(storage, 100);
    		ProducerThread p4 = new ProducerThread(storage, 300);
    		ProducerThread p5 = new ProducerThread(storage, 400);
    		ProducerThread p6 = new ProducerThread(storage, 200);
    		ProducerThread p7 = new ProducerThread(storage, 500);
    
    		// 消费者对象
    		ConsumerThread c1 = new ConsumerThread(storage, 500);
    		ConsumerThread c2 = new ConsumerThread(storage, 200);
    		ConsumerThread c3 = new ConsumerThread(storage, 800);
    
    		// 线程开始执行
    		c1.start();
    		c2.start();
    		c3.start();
    		p1.start();
    		p2.start();
    		p3.start();
    		p4.start();
    		p5.start();
    		p6.start();
    		p7.start();
    
    	}
    
    }

输出结果：

当前仓库产品数量：0
    【仓库没有：500个产品可消费】当前仓库产品数量：0
    【仓库没有：800个产品可消费】当前仓库产品数量：0
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：0
    【仓库没有：800个产品可消费】当前仓库产品数量：200
    【仓库没有：500个产品可消费】当前仓库产品数量：200
    【仓库还未满，生产：300个产品没有问题】当前仓库产品数量：200
    【仓库有：500个产品可消费】当前仓库产品数量：500
    【仓库没有：800个产品可消费】当前仓库产品数量：0
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：0
    【仓库没有：800个产品可消费】当前仓库产品数量：200
    【仓库有：200个产品可消费】当前仓库产品数量：200
    【仓库没有：800个产品可消费】当前仓库产品数量：0
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：0
    【仓库没有：800个产品可消费】当前仓库产品数量：200
    【仓库还未满，生产：100个产品没有问题】当前仓库产品数量：200
    【仓库没有：800个产品可消费】当前仓库产品数量：300
    【仓库还未满，生产：400个产品没有问题】当前仓库产品数量：300
    【仓库没有：800个产品可消费】当前仓库产品数量：700
    【仓库已无法再生产：500个产品】当前仓库产品数量：700

结果不一定会和我一样

二、lock实现

lock的实现其实和上面差不多，只不过还引入了lock,newCondition的条件，这是一个类似wait/notify的东西，用法也差不多。代码如下：

package com.lin;
    
    import java.util.LinkedList;
    import java.util.List;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    class Storehouse {
    	// 仓库的容量
    	private int capacity;
    	// object当成是生产的商品
    	private List<Object> list = new LinkedList<Object>();
    
    	 // 锁  
        private final Lock lock = new ReentrantLock();  
      
        // 仓库满的条件变量  
        private final Condition full = lock.newCondition();  
      
        // 仓库空的条件变量  
        private final Condition empty = lock.newCondition();  
    	
    	public Storehouse(int capacity) {
    		this.capacity = capacity;
    		System.out.println("当前仓库产品数量：" + list.size());
    	}
    
    	public int getCapacity() {
    		return capacity;
    	}
    
    	public void setCapacity(int capacity) {
    		this.capacity = capacity;
    	}
    
    	/**
    	 * 生产的方法
    	 * 
    	 * @throws InterruptedException
    	 */
    	public void produrce(int num) throws InterruptedException {
    		try {
    			lock.lock();
    			// 仓库还未满，且再生产num个产品不会超过仓库容量时可以生产产品
    			while (list.size() + num > this.capacity) {
    				// 仓库已满，或者放不下
    			   System.out.println("【仓库已无法再生产：" + num + "个产品】" + "当前仓库产品数量：" + list.size());
    			   empty.await(); 
    			}
    			System.out.println("【仓库还未满，生产：" + num + "个产品没有问题】" + "当前仓库产品数量：" + list.size());
    			for (int i = 0; i < num; i++) {
    				list.add(new Object());
    			}
    			full.signalAll();
    			empty.signalAll();
    		} finally {
    			lock.unlock();
    		}
    	}
    
    	/**
    	 * 消费
    	 * 
    	 * @param num
    	 * @throws InterruptedException
    	 */
    	public void consumer(int num) throws InterruptedException {
    		try {
    			lock.lock();
    			// 仓库有没有num个产品可消费
    			while (list.size() < num) {
    				System.out.println("【仓库没有：" + num + "个产品可消费】" + "当前仓库产品数量：" + list.size());
    				full.await();
    			}
    			System.out.println("【仓库有：" + num + "个产品可消费】" + "当前仓库产品数量：" + list.size());
    			for (int i = 0; i < num; i++) {
    				list.remove(0);
    			}
    			empty.signalAll();
    			full.signalAll();
    		} finally {
    			lock.unlock();
    		}
    	}
    }
    
    	class ProducerThread extends Thread {
    		// 每次生产的产品数量
    		private int num;
    
    		// 所在放置的仓库
    		private Storehouse storehouse;
    
    		// 构造函数，设置仓库
    		public ProducerThread(Storehouse storehouse, int num) {
    			this.storehouse = storehouse;
    			this.num = num;
    		}
    
    		public void run() {
    			try {
    				storehouse.produrce(num);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    
    	class ConsumerThread extends Thread {
    		// 每次生产的产品数量
    		private int num;
    
    		// 所在放置的仓库
    		private Storehouse storehouse;
    
    		// 构造函数，设置仓库
    		public ConsumerThread(Storehouse storehouse, int num) {
    			this.storehouse = storehouse;
    			this.num = num;
    		}
    
    		public void run() {
    			try {
    				storehouse.consumer(num);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    
    	}
    
    public class Test2 {
    
    	public static void main(String[] args) {
    
    		// 仓库对象
    		Storehouse storage = new Storehouse(1000);
    
    		// 生产者对象
    		ProducerThread p1 = new ProducerThread(storage, 200);
    		ProducerThread p2 = new ProducerThread(storage, 200);
    		ProducerThread p3 = new ProducerThread(storage, 100);
    		ProducerThread p4 = new ProducerThread(storage, 300);
    		ProducerThread p5 = new ProducerThread(storage, 400);
    		ProducerThread p6 = new ProducerThread(storage, 200);
    		ProducerThread p7 = new ProducerThread(storage, 500);
    
    		// 消费者对象
    		ConsumerThread c1 = new ConsumerThread(storage, 500);
    		ConsumerThread c2 = new ConsumerThread(storage, 200);
    		ConsumerThread c3 = new ConsumerThread(storage, 800);
    
    		// 线程开始执行
    		c1.start();
    		c2.start();
    		c3.start();
    		p1.start();
    		p2.start();
    		p3.start();
    		p4.start();
    		p5.start();
    		p6.start();
    		p7.start();
    
    	}
    
    }

输出结果：

当前仓库产品数量：0
    【仓库没有：500个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：0
    【仓库还未满，生产：100个产品没有问题】当前仓库产品数量：200
    【仓库还未满，生产：500个产品没有问题】当前仓库产品数量：300
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：800
    【仓库已无法再生产：300个产品】当前仓库产品数量：1000
    【仓库已无法再生产：200个产品】当前仓库产品数量：1000
    【仓库有：500个产品可消费】当前仓库产品数量：1000
    【仓库有：200个产品可消费】当前仓库产品数量：500
    【仓库还未满，生产：400个产品没有问题】当前仓库产品数量：300
    【仓库没有：800个产品可消费】当前仓库产品数量：700
    【仓库还未满，生产：300个产品没有问题】当前仓库产品数量：700
    【仓库已无法再生产：200个产品】当前仓库产品数量：1000
    【仓库有：800个产品可消费】当前仓库产品数量：1000
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：200

三、BlockingQueue实现

BlockingQueue是阻塞队列，可以实现入队阻塞、取数阻塞。对于生产者-消费者很适用，而且可以实现一有产品就可以马上就进行消费。不用像上面的两种方法都等到仓库生产足够的产品后才进行消费。而且代码量也不多，容易理解，代码如下：

package com.lin;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.atomic.AtomicInteger;
    
    class Storehouse {
    	// 仓库的容量
    	private int capacity;
    	// 仓库存储的载体
    	private BlockingQueue<Object> blockingQueue;
    	// 当前个数
    	private AtomicInteger curNum = new AtomicInteger(0);
    
    	public Storehouse(int capacity) {
    		this.capacity = capacity;
    		this.blockingQueue = new ArrayBlockingQueue<Object>(capacity);
    	}
    
    	/**
    	 * 生产的方法
    	 * 
    	 * @throws InterruptedException
    	 */
    	public void produrce(int num) {
    		while (num + curNum.get() > capacity) {
    			System.out.println("【仓库已无法再生产：" + num + "个产品】" + "当前仓库产品数量：" + curNum.get());
    		}
    
    		System.out.println("【仓库还未满，生产：" + num + "个产品没有问题】" + "当前仓库产品数量：" + blockingQueue.size());
    		for (int i = 0; i < num; i++) {
    			blockingQueue.add(new Object());
    			curNum.incrementAndGet();
    		}
    
    	}
    
    	/**
    	 * 消费
    	 * 
    	 * @param num
    	 * @throws InterruptedException
    	 */
    	public void consumer(int num) {
    		while (num > curNum.get()) {
    			System.out.println("【仓库没有：" + num + "个产品可消费】" + "当前仓库产品数量：" + blockingQueue.size());
    		}
    
    		System.out.println("【仓库有：" + num + "个产品可消费】" + "当前仓库产品数量：" + blockingQueue.size());
    		for (int i = 0; i < num; i++) {
    			try {
    				blockingQueue.take();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    			curNum.decrementAndGet();
    		}
    
    	}
    }
    
    class ProducerThread extends Thread {
    	// 每次生产的产品数量
    	private int num;
    
    	// 所在放置的仓库
    	private Storehouse storehouse;
    
    	// 构造函数，设置仓库
    	public ProducerThread(Storehouse storehouse, int num) {
    		this.storehouse = storehouse;
    		this.num = num;
    	}
    
    	public void run() {
    			storehouse.produrce(num);
    	}
    }
    
    class ConsumerThread extends Thread {
    	// 每次生产的产品数量
    	private int num;
    
    	// 所在放置的仓库
    	private Storehouse storehouse;
    
    	// 构造函数，设置仓库
    	public ConsumerThread(Storehouse storehouse, int num) {
    		this.storehouse = storehouse;
    		this.num = num;
    	}
    
    	public void run() {
    			storehouse.consumer(num);
    	}
    
    }
    
    public class Test3 {
    
    	public static void main(String[] args) {
    		// 仓库对象
    		Storehouse storage = new Storehouse(1000);
    
    		// 生产者对象
    		ProducerThread p1 = new ProducerThread(storage, 200);
    		ProducerThread p2 = new ProducerThread(storage, 200);
    		ProducerThread p3 = new ProducerThread(storage, 100);
    		ProducerThread p4 = new ProducerThread(storage, 300);
    		ProducerThread p5 = new ProducerThread(storage, 400);
    		ProducerThread p6 = new ProducerThread(storage, 200);
    		ProducerThread p7 = new ProducerThread(storage, 500);
    
    		// 消费者对象
    		ConsumerThread c1 = new ConsumerThread(storage, 500);
    		ConsumerThread c2 = new ConsumerThread(storage, 200);
    		ConsumerThread c3 = new ConsumerThread(storage, 800);
    
    		// 线程开始执行
    		c1.start();
    		c2.start();
    		c3.start();
    		p1.start();
    		p2.start();
    		p3.start();
    		p4.start();
    		p5.start();
    		p6.start();
    		p7.start();
    
    	}
    
    }

输出结果：

【仓库没有：500个产品可消费】当前仓库产品数量：0
    【仓库没有：500个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：0
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：1
    【仓库没有：500个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：0
    【仓库没有：200个产品可消费】当前仓库产品数量：39
    【仓库还未满，生产：300个产品没有问题】当前仓库产品数量：39
    【仓库还未满，生产：100个产品没有问题】当前仓库产品数量：113
    【仓库没有：800个产品可消费】当前仓库产品数量：0
    【仓库还未满，生产：200个产品没有问题】当前仓库产品数量：165
    【仓库没有：200个产品可消费】当前仓库产品数量：39
    【仓库有：200个产品可消费】当前仓库产品数量：273
    【仓库没有：500个产品可消费】当前仓库产品数量：39
    【仓库没有：500个产品可消费】当前仓库产品数量：327
    【仓库没有：500个产品可消费】当前仓库产品数量：327
    【仓库没有：500个产品可消费】当前仓库产品数量：327
    【仓库还未满，生产：400个产品没有问题】当前仓库产品数量：327
    【仓库没有：800个产品可消费】当前仓库产品数量：327
    【仓库没有：500个产品可消费】当前仓库产品数量：327
    【仓库没有：800个产品可消费】当前仓库产品数量：479
    【仓库没有：800个产品可消费】当前仓库产品数量：497
    【仓库没有：500个产品可消费】当前仓库产品数量：516
    【仓库有：500个产品可消费】当前仓库产品数量：516
    【仓库还未满，生产：500个产品没有问题】当前仓库产品数量：506
    【仓库没有：800个产品可消费】当前仓库产品数量：558
    【仓库没有：800个产品可消费】当前仓库产品数量：558
    【仓库没有：800个产品可消费】当前仓库产品数量：558
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：568
    【仓库没有：800个产品可消费】当前仓库产品数量：567
    【仓库没有：800个产品可消费】当前仓库产品数量：557
    【仓库没有：800个产品可消费】当前仓库产品数量：607
    【仓库没有：800个产品可消费】当前仓库产品数量：710
    【仓库没有：800个产品可消费】当前仓库产品数量：727
    【仓库没有：800个产品可消费】当前仓库产品数量：719
    【仓库没有：800个产品可消费】当前仓库产品数量：719
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：760
    【仓库没有：800个产品可消费】当前仓库产品数量：769
    【仓库有：800个产品可消费】当前仓库产品数量：835

四、管道实现

待编程。。