【设计模式】生产者消费者模式

野性酷女 2022-05-16 03:40 287阅读 0赞

# 生产消费者模型 #

生产者消费者模型具体来讲，就是在一个系统中，存在生产者和消费者两种角色，他们通过内存缓冲区进行通信，生产者生产消费者需要的资料，消费者把资料做成产品。生产消费者模式如下图。  
![这里写图片描述][70]  
　　在日益发展的服务类型中，譬如注册用户这种服务，它可能解耦成好几种独立的服务（账号验证，邮箱验证码，手机短信码等）。它们作为消费者，等待用户输入数据，在前台数据提交之后会经过分解并发送到各个服务所在的url，分发的那个角色就相当于生产者。消费者在获取数据时候有可能一次不能处理完，那么它们各自有一个请求队列，那就是内存缓冲区了。做这项工作的框架叫做消息队列。  
\#生产者消费者模型的实现  
　　生产者是一堆线程，消费者是另一堆线程，内存缓冲区可以使用List数组队列，数据类型只需要定义一个简单的类就好。关键是如何处理多线程之间的协作。这其实也是多线程通信的一个范例。

在这个模型中，最关键就是内存缓冲区为空的时候消费者必须等待，而内存缓冲区满的时候，生产者必须等待。其他时候可以是个动态平衡。值得注意的是多线程对临界区资源的操作时候必须保证在读写中只能存在一个线程，所以需要设计锁的策略。

# 第一种：使用wait-notify #

public class ProducerAndConsumer {
    
    	// 作为synchronized的对象监视器
    	private static Object lock = new Object();
    	// 随机数
    	private static Random random = new Random();
    	
    	static class Producer implements Runnable {
    
    		private Queue queue;
    		
    		public Producer(Queue queue) {
    			this.queue = queue;
    		}
    		
    		@Override
    		public void run() {
    			while(true) {
    				synchronized(lock) {
    					try {
    						Thread.sleep(500);
    						
    						// 超过规定的队列长度，停止生产
    						if(queue.size() >= 10) {
    							lock.wait();
    						}
    						
    						// 生产
    						int i = random.nextInt(100);
    						queue.add(i);
    						System.out.println(Thread.currentThread().getName() + "已生产：" + i);
    						
    						// 生产完后队列不为空，唤醒消费者
    						lock.notify();
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}
    				}
    			}
    		}
    	}
    	
    	static class Consumer implements Runnable {
    
    		private Queue queue;
    		
    		public Consumer(Queue queue) {
    			this.queue = queue;
    		}
    		
    		@Override
    		public void run() {
    			while(true) {
    				synchronized(lock) {
    					try {
    						Thread.sleep(600);
    						
    						// 如果队列为空，则让消费者进入等待状态
    						if(queue.isEmpty()) {
    							lock.wait();
    						}
    						
    						// 消费
    						System.out.println(Thread.currentThread().getName() + "已消费：" + queue.poll());
    						
    						// 消费完后，唤醒生产者
    						lock.notify();
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}
    				}
    			}
    		}
    	}
    	
    	public static void main(String[] args) {
    		// 队列
    		Queue queue = new LinkedList();
    		
    		new Thread(new Producer(queue), "生产者1号").start();
    		new Thread(new Producer(queue), "生产者2号").start();
    		new Thread(new Consumer(queue), "消费者1号").start();
    		new Thread(new Consumer(queue), "消费者2号").start();
    	}
    }

# 第二种：使用阻塞队列 #

阻塞队列的特点：

*  当队列元素已满的时候，阻塞插入操作
 *  当队列元素为空的时候，阻塞获取操作

`ArrayBlockingQueue`与`LinkedBlockingQueue`都是支持FIFO(先进先出)

ArrayBlockingQueue 是有界的，而 LinkedBlockingQueue 是无界的

public class ProducerAndConsumer02 {
    
    	// 随机数
    	private static Random random = new Random();
    	
    	static class Producer implements Runnable {
    
    		private BlockingQueue queue;
    		
    		public Producer(BlockingQueue queue) {
    			this.queue = queue;
    		}
    		
    		@Override
    		public void run() {
    			while(true) {
    				try {
    					Thread.sleep(500);
    					
    					// 生产
    					int i = random.nextInt(100);
    					queue.put(i);
    					System.out.println(Thread.currentThread().getName() + "已生产：" + i);
    					
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	static class Consumer implements Runnable {
    
    		private BlockingQueue queue;
    		
    		public Consumer(BlockingQueue queue) {
    			this.queue = queue;
    		}
    		
    		@Override
    		public void run() {
    			while(true) {
    				try {
    					Thread.sleep(600);
    					
    					// 消费
    					System.out.println(Thread.currentThread().getName() + "已消费：" + queue.take());
    				
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	public static void main(String[] args) {
    		// 队列
    		BlockingQueue queue = new ArrayBlockingQueue(10);
    		
    		new Thread(new Producer(queue), "生产者1号").start();
    		new Thread(new Producer(queue), "生产者2号").start();
    		new Thread(new Consumer(queue), "消费者1号").start();
    		new Thread(new Consumer(queue), "消费者2号").start();
    	}
    }

[70]: /images/20220516/a756dedf8c464f3c944d922c5da1602a.png