管程 Monitor

管程的概念：管程将共享变量以及对这些共享变量的操作封装起来，形成一个具有一定接口的功能模块，这样只能通过管程提供的某个过程才能访问管程中的资源。进程只能互斥地使用管程，使用完之后必须释放管程并唤醒入口等待队列中的进程。

生产者-消费者问题

问题描述：使用一个缓冲区来存取数据。生产者线程否则生产数据；消费者线程负责消费数据。只有缓冲区没有满，生产者才可以写入数据；只有缓冲区不为空，消费者才可以读出数据。

缓冲区通常使用一个阻塞队列来实现。
也就是说，当消费者线程尝试从队列获取数据时，如果队列是空的，那么消费者线程会自动挂起等待（挂起时间也可以进行设置），直到它获得了数据为止。

代码实现

在python中，使用Queue模块实现一个阻塞队列。

put()方法
调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数，默认为1。如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0，put方法将引发Full异常。

get()方法
调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。

  1. from queue import Queue
  3. que = Queue()
  4. que.put(10) # 将一个值放入队列
  5. que.get()

我们创建一个特殊的信号量，约定好当consumer接受到这个特殊值的时候就停止程序。这样当我们要结束程序的时候，我们只需要把这个信号量加入队列即可。

image.png
这里有一个细节是我们在consumer当中，当读取到singal的时候，在跳出循环之前我们又把singal放回了队列。原因也很简单，因为有时候consumer线程不止一个，这个singal上游只放置了一个，只会被一个线程读取进来，其他线程并不会知道已经获得了singal的消息，所以还是会继续执行。

而当consumer关闭之前放入singal就可以保证每一个consumer在关闭的之前都会再传递一个结束的信号给其他未关闭的consumer读取。这样一个一个的传递，就可以保证所有consumer都关闭。

，虽然利用队列可以解决生产者和消费者通信的问题，但是上游的生产者并不知道下游的消费者是否已经执行完成了。假如我们想要知道，应该怎么办？

Python的设计者们也考虑到了这个问题，所以他们在Queue这个类当中加入了task_done和join方法。利用task_done，消费者可以通知queue这一个任务已经执行完成了。而通过调用join，可以等待所有的consumer完成。

image.png

q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作

我们之前在介绍一些分布式调度系统的时候曾经说到过，在调度系统当中，调度者会用一个优先队列来管理所有的任务。当有机器空闲的时候，会有限调度那些优先级高的任务。

其实这个调度系统也是基于我们刚才介绍的生产消费者模型开发的，只不过将调度队列从普通队列换成了优先队列而已。所以如果我们也希望我们的consumer能够根据任务的优先级来改变执行顺序的话，也可以使用优先队列来进行管理任务。

关于优先队列的实现我们已经很熟悉了，但是有一个问题是我们需要实现挂起等待的阻塞功能。这个我们自己实现是比较麻烦的，但好在我们可以通过调用相关的库来实现。比如threading中的Condition，Condition是一个条件变量可以通知其他线程，也可以实现挂起等待。
image.png
最后介绍一下Queue的其他设置，比如我们可以通过size参数设置队列的大小，由于这是一个阻塞式队列，所以如果我们设置了队列的大小，那么当队列被装满的时候，往其中插入数据的操作也会被阻塞。此时producer线程会被挂起，一直到队列不再满为止。

当然我们也可以通过block参数将队列的操作设置成非阻塞。比如que.get(block=False)，那么当队列为空的时候，将会抛出一个队列为空的异常。同样，que.put(data, block=False)时也一样会得到一个队列已满的异常。