通过管程和信号量分别实现生产者消费者模式

朱雀 2021-06-10 20:38 388阅读 0赞

# 了解管程 #

> 管程是操作系统中的一个概念，在操作系统中，当多个进程或线程方法访问共享变量（俗称临界区）的时候，我们需要控制同一时刻只能有一个线程对共享变量进行访问。  
> 其中一种方案是通过互斥量实现，比如锁、信号量，该种方案需要程序员在访问临界区前手动加锁，结束访问后释放锁，增加了编码难度和错误率。  
> 管程将要访问的共享变量、访问变量的过程、数据结构 封装起来，程序员只要调用管程就可以实现线程同步，而同步的操作，管程已经给你做好了。

# 了解生产者消费者模式 #

> 在现实生活中，总是有生产消费的问题，比如我给你写一封邮件，我就是邮件的生产方，你就是邮件的消费方，而邮箱就是一个存放邮件的中介。假如没有邮箱，我给你写邮件的时候假如你不在线，你就无法收到邮件，邮件就不见了。再或者我给你在1分钟内发了10封邮件，而你在这1分钟只看了2封，然后另外8封就不见了。针对此般问题，一种典型解决方案： 生产者 – 缓冲区 – 消费者 就诞生了。  
> 在该模式中，生产者和消费者是解耦的，生产者将生产好的产品放入缓冲区，消费者从缓冲区取产品进行消费。缓冲区则可以缓和两者速度不匹配的问题。  
> 需要注意的一点是，在该模式中，缓冲区不一定是临界区的存在，要看缓冲区的实现方式。可以参看java.util.concurrent中的ArrayBlockingQueue和LinkedBlockingQueue。前者采用数组作为缓冲区，后者是链表。链表是一个有序的结构，消费每次从头取，生产插入到链表尾部，采用的是锁分离的思想，同一时刻可以有两个线程进入缓冲区。

#### 通过信号量实现生产者消费者模式 ####

import java.util.ArrayDeque;
    import java.util.Queue;
    import java.util.concurrent.Semaphore;
    
    public class ProducerComsumer { 
        // 只允许同时只有一个producer或consumer进入共享区
        private static Semaphore metux = new Semaphore(1);
        // 共享区空闲数
        private static Semaphore empty = new Semaphore(1000);
        // 共享区装满数
        private static Semaphore full = new Semaphore(0);
        // 缓冲区（缓冲区用什么数据结构，根据实际需求选择）
        private static Queue<String> container = new ArrayDeque<>();
    
        public static void main(String[] args) { 
            Thread producer1 = new Thread(() -> produce());
            Thread consumer1 = new Thread(() -> consume());
            producer1.start();
            consumer1.start();
        }
    
        /** * 生产方法 */
        public static void produce() { 
            while (true) { 
                try { 
                    empty.acquire(); // 空闲容量 - 1
                    metux.acquire(); // 获取互斥量，保证同时只有一个线程进入临界区
                    doProduce();    // 临界区操作
                    metux.release(); // 释放互斥量
                    full.release();  // 装满容量 + 1
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        /** * 消费方法 */
        public static void consume() { 
            while (true) { 
                try { 
                    full.acquire(); // 获取商品，容量的信号量 - 1
                    metux.acquire(); // 获取互斥量，保证同时只有一个线程进入临界区
                    doConsume();    // 临界区操作
                    metux.release(); // 释放互斥量
                    empty.release(); // 空闲容量 + 1
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        /** * 真实生产方法 */
        private static void doProduce() throws InterruptedException { 
            container.offer("product");
            Thread.sleep(1000);
            System.out.println("生产者生产product到缓冲区,当前大小：" + container.size() + " 当前时间：" + System.currentTimeMillis());
        }
    
        /** * 真实消费方法方法 */
        private static void doConsume() throws InterruptedException { 
            container.poll();
            Thread.sleep(1000);
            System.out.println("消费者从缓冲区消费product,当前大小：" + container.size() + " 当前时间：" + System.currentTimeMillis());
        }
    }

![运行输出：][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70]

#### 通过管程实现生产者消费者模式 ####

import java.util.ArrayDeque;
    import java.util.Queue;
    
    public class MonitorProducerConsumer { 
    
        /** * 主线程，启动生产者和消费者 * @param args */
        public static void main(String[] args) { 
            MyMonitor myMonitor = new MyMonitor();
    
            Thread produce = new Thread(() -> { 
                while (true) {   // 生产者线程
                    myMonitor.produce();
                }
            });
            Thread consume = new Thread(() -> { 
                while (true) {   // 消费者线程
                    myMonitor.consume();
                }
    
            });
            consume.start();
            produce.start();
        }
    
        /** * 管程（包含了共享变量、操作过程、数据结构） * 特点：操作方法都由synchronized标注，表明同一时刻只有一个线程可以执行对象方法，符合生产消费者的互斥访问 */
        static class MyMonitor { 
    
            private static final int MAX = 100;
    
            /** * 缓冲区 * 缓冲区用什么数据结构，根据实际需求选择 */
            private Queue<String> container = new ArrayDeque<>();
    
            /** * 生产方法 */
            public synchronized void produce() { 
                try { 
                    if (getSize() == MAX) wait();   //这里默认阻塞的对象就是MyMonitor类的对象，也就是当前对象
                    doProduce();
                    if (getSize() == 1) notify();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
    
            /** * 消费方法 */
            public synchronized void consume() { 
                try { 
                    if (getSize() == 0) wait();
                    doConsume();
                    if (getSize() == MAX - 1) notify();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
    
    
            private int getSize() { 
                return container.size();
            }
    
            /** * 真实生产方法 */
            private void doProduce() throws InterruptedException { 
                container.offer("product");
                Thread.sleep(1000);
                System.out.println("生产者生产product到缓冲区,当前大小：" + container.size() + " 当前时间：" + System.currentTimeMillis());
            }
    
            /** * 真实消费方法方法 */
            private void doConsume() throws InterruptedException { 
                container.poll();
                Thread.sleep(1000);
                System.out.println("消费者从缓冲区消费product,当前大小：" + container.size() + " 当前时间：" + System.currentTimeMillis());
            }
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70 1]  
可以看到，在管程的实现中，用户（main方法）无需关注访问临界区的同步操作，只需要启动生产者和消费者就可以了。

java中的生产消费模式可以参看java.util.concurrent中的ArrayBlockingQueue和LinkedBlockingQueue的实现。

如果对你有帮助，帮忙点个赞哦！是对我最大的鼓励。  
如果有什么问题，欢迎讨论。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70]: /images/20210518/f22e8a3bec154e8bace4adf03423431d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70 1]: /images/20210518/326d5d79925f412f84ccce10e44f53b8.png