Java"设计模式"——生产消费者模式及三种实现方式

小咪咪 2021-12-02 00:12 365阅读 0赞

## Java"设计模式"——生产消费者模式及三种实现方式 ##

>    **虽然生产消费者模型不算是Java的23种设计模式之一，但还是将其归为半个"设计模式"（属于我的设计模式）**

### 一、生产消费者模式原理 ###

####   1、生产消费者模式主要是将生产者与消费者解耦，通过一个容器来解决生产者和消费者的强耦合问题，生产者消费者彼此之间不直接通讯，而是通过阻塞队列来进行通讯。 ####

####   2、生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者也不找生产者要数据，而是直接从阻塞队列中取得，阻塞队列相当于一个`缓冲区`，平衡了生产者与消费者的处理能力，使得生产者与消费者是两个完全独立的并发主体。 ####

####   3、生产者消费者的关系如下 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70]

#####     ①生产速度与消费速度的关系 #####

######      a)生产速度>消费速度：存货 ######

######      b)生产速度=消费速度：零库存（最理想） ######

######      c)生产速度<消费速度：缺货 ######

#####     ②容器的容量要有限制，否则会有存货积压 #####

######      a) 容器已满，停止生产，通知消费者消费（生产快于消费），生产者等待消费者消费。 ######

######      b) 容器已空，停止消费，通知生产者生产（消费快于生产），消费者等待生产者生产 ######

####   4、生产消费者模型应用 ####

#####     ①生产消费者模型应用 #####

######      a) 生产者与消费者速率不匹配问题（A产生数据，B消费数据，且速率不同） ######

######      b) 解耦，生产者与消费者相互之间没有关系（很多东西都能做容器） ######

#####     ②生产消费者实际开发问题 #####

######      调整生产速度（休眠时间）以及生产者与消费者的个数用来控制生产消费模式（休眠时间基本实际开发中不会使用，实际开发中需要进行测试，得到生产者与消费者的速度差异，根据快慢情况决定生产者消费者的线程数，实现生产速度约等于消费速度） ######

### 二、生产消费者模式的多种实现 ###

####   共3种实现 ####

>   **等待唤醒机制**  
>   **`等待唤醒+阻塞队列机制`**  
>   **Condition机制（内置等待唤醒+阻塞队列）**

####   1、等待唤醒机制 ####

#####     概述：在商品类中设置两个synchronized方法，一个是生产商品方法（由生产者调用），另一个是消费商品方法（由消费者调用），当商品数为最大值时，停止生产，让消费者消费，当商品数为0时，停止消费，让生产者生产。 #####

#####     特点：生产者与消费者耦合度还是太高（两者只是通过一个商品类来交互） #####

package com.xiaoaxiao.test.thread_test.production_consumer_model;
    		
    		import java.util.ArrayList;
    		import java.util.List;
    		
    		/**
    		 * Created by xiaoaxiao on 2019/7/15
    		 * Description: 多线程的生产消费模式
    		 */
    		
    		class Goods1{
    		    private String goodsName;
    		    private int count;
    		    private int maxCount;
    		
    		    public Goods1(int maxCount) {
    		        this.maxCount = maxCount;
    		    }
    		
    		    // 生产商品方法
    		    public synchronized void set(String goodsName){
    		        while (this.count==maxCount){
    		            try {
    		                // 等待消费者消费
    		                wait();
    		            } catch (InterruptedException e) {
    		                e.printStackTrace();
    		            }
    		        }
    		        try {
    		            Thread.sleep(20);
    		        } catch (InterruptedException e) {
    		            e.printStackTrace();
    		        }
    		        this.goodsName = goodsName;
    		        this.count++;
    		        System.out.println(Thread.currentThread().getName()
    		                +"生产"+goodsName+toString());
    		        // 唤醒等待的线程
    		        notifyAll();
    		    }
    		
    		    // 消费商品方法
    		    public synchronized void get(){
    		  // 这里必须要使用while语句，保证每一次进入某一线程都要判断一下当前this.count是否为0
    		  // 错误实例：消费线程1与消费线程2都处于等待状态（都卡在了wait()语句处），当这两个线程被一个生产线程唤醒时，
    		  //  消费线程2先进入将count--，
    		  // 如果不用while语句，消费线程1后进入时（接着wait()继续执行）也会将count--，导致错误（数量产生负数）
    		        while (this.count == 0){
    		            System.out.println("卖完了");
    		            try {
    		                // 等待生产者生产
    		                wait();
    		            } catch (InterruptedException e) {
    		                e.printStackTrace();
    		            }
    		        }
    		        this.count--;
    		        System.out.println(Thread.currentThread().getName()
    		                +"消费"+goodsName+toString());
    		        // 唤醒等待中的生产者线程
    		        notifyAll();
    		    }
    		
    		    @Override
    		    public String toString() {
    		        return "Goods{" +
    		                "goodsName='" + goodsName + '\'' +
    		                ", count=" + count +
    		                '}';
    		    }
    		}
    		
    		class Producer1 implements Runnable{
    		
    		    private Goods1 goods;
    		
    		    public Producer1(Goods1 goods) {
    		        this.goods = goods;
    		    }
    		
    		    @Override
    		    public void run() {
    		        while (true) {
    		            this.goods.set("西瓜");
    		        }
    		    }
    		}
    		
    		class Customer1 implements Runnable{
    		
    		    private Goods1 goods;
    		
    		    public Customer1(Goods1 goods) {
    		        this.goods = goods;
    		
    		    }
    		
    		    @Override
    		    public void run() {
    		        // 不断消费
    		        while (true) {
    		            this.goods.get();
    		        }
    		    }
    		}
    		
    		public class MultipleThreadProductionConsumerModel {
    		
    		    public static void main(String[] args) {
    		
    		        Goods1 goods1 = new Goods1(10);
    		
    		        List<Thread> threadList = new ArrayList<>();
    		
    		        Producer1 producer1 = new Producer1(goods1);
    		        Customer1 customer1 = new Customer1(goods1);
    		
    		        for(int i=0;i<5;i++){
    		            Thread producerThread = new Thread(producer1,"生产者"+i);
    		            threadList.add(producerThread);
    		        }
    		
    		        for (int i=0;i<10;i++){
    		            Thread customerThread = new Thread(customer1,"消费者"+i);
    		            threadList.add(customerThread);
    		        }
    		
    		        for (Thread thread:threadList){
    		            thread.start();
    		        }
    		    }
    		}

#####     特别需要注意的是：无论是消费者方法还是生产者方法，在`判断当前商品数量时，一定要使用while()循环`。保证`每一次进入某一线程时必须先判断一下当前的count是否满足要求`。 #####

>  错误实例（使用if）：  
>   **消费线程1与消费线程2不同时进入该同步方法，由于没有生产，线程1和线程2都会处于等待状态，假设这两个线程被生产线程唤醒时，线程2先进入该同步方法将count-1，当线程2从该方法出来而线程1再次进入该方法时，会直接`从等待语句(wait())的下一句开始继续执行`，因此此时count会再次减1，若线程2将count-1后count已经等于0了，这样做就会导致错误（数量产生负数）。因此必须使用while()再次判断当前count的值。(PS：`同一个线程再次进入同步方法时会从上次等待的地方开始继续执行`)**

####   2、等待唤醒机制+阻塞队列 ####

#####     概述：将生产者消费者完全解耦，两者之间使用一个阻塞队列实现通讯，将`该阻塞队列作为锁`，在生产者与消费者中设置同步代码块，实现生产者与消费者之间的同步。 #####

#####     特点：生产者消费者耦合度很低，可读性及可维护性比较好。 #####

#####     Goods #####

package com.xiaoaxiao.test.thread_test.production_consumer_model.PcByQueue;
    		
    		/**
    		 * Created by xiaoaxiao on 2019/7/20
    		 * Description: 商品类
    		 */
    		public class Goods {
    		
    		    private final String id;
    		    private final String goodName;
    		
    		    public Goods(String id, String goodName) {
    		        this.id = id;
    		        this.goodName = goodName;
    		    }
    		
    		    public String getId() {
    		        return id;
    		    }
    		
    		    public String getGoodName() {
    		        return goodName;
    		    }
    		
    		    @Override
    		    public String toString() {
    		        return "Goods{" +
    		                "id='" + id + '\'' +
    		                ", goodName='" + goodName + '\'' +
    		                '}';
    		    }
    		}

#####     Producer #####

package com.xiaoaxiao.test.thread_test.production_consumer_model.PcByQueue;
    		
    		import java.util.Queue;
    		import java.util.concurrent.atomic.AtomicInteger;
    		
    		/**
    		 * Created by xiaoaxiao on 2019/7/20
    		 * Description: 生产者类
    		 *          1、生产商品
    		 *          2、将生产的商品添加到容器中
    		 *          3、如果容器满了，生产等待，通知消费者消费
    		 */
    		public class Producer implements Runnable{
    		
    		    private final Queue<Goods> goods;
    		
    		    private final Integer maxCapacity = 10;
    		
    		    // 创建一个原子变量，多个线程共享且线程安全
    		    private final AtomicInteger id = new AtomicInteger(0);
    		
    		    public Producer(Queue<Goods> goods) {
    		        this.goods = goods;
    		    }
    		
    		    @Override
    		    public void run() {
    		        while (true){
    		            try {
    		                Thread.sleep(1000);
    		            } catch (InterruptedException e) {
    		                e.printStackTrace();
    		            }
    		            // 将这个容器锁住
    		            synchronized (this.goods){
    		                if(this.goods.size()==maxCapacity){
    		                    System.out.println(Thread.currentThread().getName()
    		                            +" 容器满了 等待消费");
    		                    try {
    		                        this.goods.wait();
    		                    } catch (InterruptedException e) {
    		                        e.printStackTrace();
    		                    }
    		                }else {
    		                    Goods newGood = new Goods(
    		                            // 获取id并+1
    		                            String.valueOf(id.getAndIncrement()),
    		                            "商品"
    		                    );
    		
    		                    this.goods.add(newGood);
    		
    		                    System.out.println(Thread.currentThread().getName()
    		                                +" 生产商品 "+ newGood);
     						    // 生产了商品通知消费者消费
    		                    this.goods.notifyAll();
    		                }
    		
    		            }
    		        }
    		    }
    		
    		}

#####     Customer #####

package com.xiaoaxiao.test.thread_test.production_consumer_model.PcByQueue;
    		
    		import java.util.Queue;
    		
    		/**
    		 * Created by xiaoaxiao on 2019/7/20
    		 * Description: 消费者类
    		 *          1、消费商品
    		 *          2、从容器中取出商品
    		 *          3、如果容器为空，消费等待，通知生产者生产
    		 */
    		public class Customer implements Runnable{
    		
    		    private final Queue<Goods> goods;
    		
    		    public Customer(Queue<Goods> goods) {
    		        this.goods = goods;
    		    }
    		
    		
    		    @Override
    		    public void run() {
    		        while (true){
    		            try {
    		                Thread.sleep(1000);
    		            } catch (InterruptedException e) {
    		                e.printStackTrace();
    		            }
    		            synchronized (this.goods){
    		
    		                if(this.goods.isEmpty()){
    		                    System.out.println(Thread.currentThread().getName()
    		                            +" 容器已空，通知生产者生产");
    		                    try {
    		                        this.goods.wait();
    		                    } catch (InterruptedException e) {
    		                        e.printStackTrace();
    		                    }
    		                    
    		
    		                }else {
    		                    Goods newGood = this.goods.poll();
    		                    if(newGood!=null){
    		                        System.out.println(Thread.currentThread().getName()
    		                                        +" 消费商品 "+newGood);
    		                    }
    		                    // 消费了商品通知生产者生产
    		                    this.goods.notifyAll();
    		                }
    		
    		            }
    		        }
    		    }
    		}

#####     Test #####

package com.xiaoaxiao.test.thread_test.production_consumer_model.PcByQueue;
    		
    		import java.util.ArrayList;
    		import java.util.LinkedList;
    		import java.util.List;
    		import java.util.Queue;
    		
    		
    		/**
    		 * Created by xiaoaxiao on 2019/7/20
    		 * Description: 生产消费模型测试类
    		 */
    		public class PCTest {
    		
    		    public static void main(String[] args) {
    		
    		        final Queue<Goods> goods = new LinkedList<>();
    		
    		        List<Thread> threadList = new ArrayList<>();
    		
    		        // 创建一个Producer和Customer的对象和创建线程Thread时
    		        // 和每次都new一个新的Producer效果一样，但节省内存空间
    		        final Producer producer = new Producer(goods);
    		        final Customer customer = new Customer(goods);
    		
    		        // 创建生产者线程
    		        for (int i=0;i<5;i++){
    		            Thread thread = new Thread(producer,"生产者"+i);
    		            threadList.add(thread);
    		        }
    		
    		        // 创建消费者线程
    		        for (int i=0;i<10;i++){
    		            Thread thread = new Thread(customer,"消费者"+i);
    		            threadList.add(thread);
    		        }
    		
    		        // 通过threadList统一启动所有线程
    		        for (Thread thread : threadList){
    		            thread.start();
    		        }
    		    }
    		}

#####     需要注意的是：在这种实现机制中，创建一个Producer和Customer的对象和创建线程Thread和每次都new一个新的Producer效果一样，但节省了很多内存空间。另外，注意原子变量（AtomicInteger），保证了多个线程共享且线程安全。 #####

####   3、`Condition机制` ####

#####     概述：与等待唤醒+阻塞队列比较类似，只不过Condition中自带一个同步队列和多个等待队列。都不需要手动创建阻塞队列，而且使用Condition机制更能降低耦合度。 #####

package com.xiaoaxiao.test.thread_test.production_consumer_model.realize_by_condition;
    		
    		import org.omg.PortableServer.THREAD_POLICY_ID;
    		
    		import java.util.ArrayList;
    		import java.util.List;
    		import java.util.concurrent.locks.Condition;
    		import java.util.concurrent.locks.Lock;
    		import java.util.concurrent.locks.ReentrantLock;
    		
    		/**
    		 * Created by xiaoaxiao on 2019/7/21
    		 * Description: Condition机制实现生产消费者模型
    		 */
    		
    		class Goods{
    		
    		    private String name;
    		    private int count;
    		    private int maxCount;
    		
    		    public Goods(int maxCount){
    		        this.maxCount = maxCount;
    		    }
    		
    		    private Lock lock = new ReentrantLock();
    		    // 消费者队列
    		    private Condition customer = lock.newCondition();
    		    // 生产者队列
    		    private Condition producer = lock.newCondition();
    		
    		    // 生产商品
    		    public void setGoods(String name){
    		        lock.lock();
    		        try{
    		            // 此处同样需要使用while，因为会从await()下一句开始继续执行（被唤醒时）
    		            while (this.count == this.maxCount){
    		                System.out.println("商品数量已最大，停止生产");
    		                producer.await();
    		
    		            }
    		//            Thread.sleep(200);
    		            // 生产商品
    		            this.name = name;
    		            this.count++;
    		            System.out.println(Thread.currentThread().getName()+"生产了"+toString());
    		            customer.signalAll();
    		        } catch (InterruptedException e) {
    		            e.printStackTrace();
    		        } finally {
    		            lock.unlock();
    		        }
    		    }
    		
    		    // 消费商品
    		    public void getGoods(){
    		        lock.lock();
    		        try {
    		            while (this.count==0){
    		                System.out.println("商品已空，停止消费");
    		                customer.await();
    		
    		            }
    		            this.count--;
    		            System.out.println(Thread.currentThread().getName()+"消费了"+toString());
    		            producer.signalAll();
    		        } catch (InterruptedException e) {
    		            e.printStackTrace();
    		        } finally {
    		            lock.unlock();
    		        }
    		    }
    		
    		    @Override
    		    public String toString() {
    		        return "Goods{" +
    		                "name='" + name + '\'' +
    		                ", count=" + count +
    		                '}';
    		    }
    		}
    		
    		class Producer implements Runnable{
    		
    		    private Goods goods;
    		
    		    public Producer(Goods goods){
    		        this.goods = goods;
    		    }
    		
    		    @Override
    		    public void run() {
    		        while (true){
    		            this.goods.setGoods("西瓜");
    		        }
    		    }
    		}
    		
    		class Customer implements Runnable{
    		
    		    private Goods goods;
    		
    		    public Customer(Goods goods){
    		        this.goods = goods;
    		    }
    		
    		    @Override
    		    public void run() {
    		        while (true){
    		            this.goods.getGoods();
    		        }
    		    }
    		}
    		
    		public class MultipleThreadPcByCondition {
    		
    		    public static void main(String[] args) {
    		        Goods goods = new Goods(10);
    		
    		        Producer producer = new Producer(goods);
    		        Customer customer = new Customer(goods);
    		
    		        List<Thread> threadList = new ArrayList<>();
    		
    		        for (int i=0;i<5;i++){
    		            Thread thread = new Thread(producer,"生产者"+i);
    		            threadList.add(thread);
    		        }
    		
    		        for (int i=0;i<9;i++){
    		            Thread thread = new Thread(customer,"消费者"+i);
    		            threadList.add(thread);
    		        }
    		
    		        for (Thread thread : threadList){
    		            thread.start();
    		        }
    		    }
    		}

#####     需要注意的是：此时在判断this.count的值时同样需要使用while，因为会从await()下一句开始继续执行（被唤醒时） #####

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70]: /images/20211201/4557ed1f442d4390a36b63cdd0757ce5.png