python线程同步原语--源码阅读

前面两篇文章，写了python线程同步原语的基本应用。下面这篇文章主要是通过阅读源码来了解这几个类的内部原理和是怎么协同一起工作来实现Python多线程的。

相关文章链接：python同步原语--线程锁 

                      python--线程同步原语                            

 

一、关于Condition类

Condition的用法：

用来记录线程的状态变量

查看Condition的源码，会看到作者给开发者提供的文档说明。‘Class that implemets a condition variable’写得很明白，这是一个用来记录线程状态的类。

 

1. Condition对象初始化

从这段代码可以看出，Condition使用了threading模块的Rlock类，关于Rlock的用法可以看我之前写的一篇文章python同步原语--线程锁 。在对象初始化的同时，将Rlock的请求锁和释放锁方法赋给了内部的self.acquire和self.release对象方法。当初始化对象同时初始化这两个方法，也就是说，每个对象在实例化的时候都会实例一个新的可重入锁（RLock）。这样可以避免不同对象（condition实例对象）间对类中共享方法的争夺，避免出现死锁的问题。

 

 

这段代码非常的重要。如果熟悉python的上下文管理的朋友应该一看就明白，这是上下文管理中的进入和退出操作。当在调用with时，程序会自动调用_ _enter_ _方法，在程序执行完毕，退出此上下文环境时，自动调用_ _exit_ _方法。那么在这里的_ _enter_ _和_ _exit_ _方法分别有什么用呢？通过阅读源码发现，前者是调用了Rlock的acquier方法（获取锁），而后者调用了Rlock的release方法（释放锁）。在下面我会继续讲这两个方法在类中的作用。

 

2. wait()方法

源码中对wait()方法的定义是‘Wait  untified  or  until  a  timeout  occurs’。意思是阻塞等待知道有提示（notify）或者超时时间（timeout）的到达。

 

再看看wait()函数的内部逻辑

_is_owned()方法是判断此Condition对象是否有获取到锁，如果没有获取到锁（可能是可重入锁的获取次数已经达到预定值，不过这种情况很少发生），就会报出错误。接下来是对需要等待的程序进行一些列的处理。先是给这个程序分配锁，对它的程序空间和内部变量进行封锁。同时把这个加锁后的程序放进双端队列（deque）‘等待者们’中。

好像wait()方法的功能到此就结束了。但是注意到下面还有try函数块，旁边一行注释写着‘restore  state  no matter  what’然后又举了一个KeyboardInterrupt的异常情况。意思是当出现了例如键盘输入ctrl+C这类操作的时候，程序如何退出阻塞。如果在调用wait方法的时候没有传入timeout参数，那么，等待者程序就会重新获取锁。如果有timeout参数，就会根据参数来确定退出阻塞的时间。这就是为什么我们有时在输入ctrl+C强行退出阻塞的时候，程序会等待一会儿才给出退出程序的提示的原因。

 

 3. notify()方法

接下来这个notify()方法在Condition类中也是非常的重要（queue模块内部也调用了这个函数）

notify()方法内部实现：

notify直接翻译过来就是‘提示’的意思。那么为什么Condition对象需要‘提示’呢？阅读源码下来，其真正的功能不是提示，而是锁的释放，并且在释放了指定数量的waiters之后，顺便将他们从‘等待者们’队列中删除。如果直接理解为提示，就会很难理解了。但这是老外在定义函数时的写法，本人的理解是，有点像给阻塞的程序发出信号（提示），停止阻塞（释放锁），这么理解应该也算勉强解释得过去吧。

 

Condition内部另外还有一个notify_all（）方法，这个方法对‘等待者们’队列中的所有的程序都发出‘提示’，释放锁，而没有像notify中那样有数量n的限制。

源码：

 

那么总结上面的Condititon内部的方法实现，可以看出，Condition类是为了实现一种状态的‘保存’，即在多线程编程的情况下，由于线程间共享空间而容易引发错误，往往需要让一些线程先执行，而后面的线程等待（阻塞）。那么如果这些程序需要阻塞等待，就会调用Condition类实例对象的wait方法，当结束等待的信号发出时，就会调用Condition的notify方法对队列中的程序进行释放锁操作。

 

二、关于Segmaphore和BoundedSegmaphore

如果在主机执行IO密集型任务的时候再执行这种短时间内完成大量任务（多线程）的程序时，计算机就有很大可能会宕机。

这时候就可以为这段程序添加一个计数器（counter）功能，来限制一个时间点内的线程数量。当每次进行IO操作时，都需要向segmaphore请求资源（锁），如果没有请求到，就阻塞等待，请求成功才就像执行任务。

 

那么segmaphore的内部实现是怎样的呢？实质上segmaphore也是锁，其内部也是通过Lock和Condition实现的。Lock是单锁，而segmaphore是可以自己定义的多锁。在初始化segmaphore时，需要传入参数counter。当线程向segmaphore请求资源（锁）时，内部的counter会自动减1。当释放资源（锁）的时，counter就会自动加1。

segmaphore主要有两个方法，acquire()和release()方法。

1. acquire()方法

官方的定义：

def acquire(self, blocking=True, timeout=None):

当内部的counter（源码实际上是用value变量保存）等于0的时候，其他线程acquire会阻塞。这个时候，之前向segmaphore发出请求并获得锁的线程，它们如果同时执行完任务并希望释放锁时，那么锁的释放是随机的。任何一个完成任务的线程都会释放锁，这个顺序跟线程向请求的时间和任务完成的时间是没有任何关系的。

参数的解析：

1）blocking：默认为True，当线程请求不到资源的时候，会阻塞等待。如果设置为False，则线程请求不到资源时不会阻塞。

2）timeout：如果设置blocking = True，即默认值时，经过timeout时间会退出阻塞。

 

2. release()方法

这个方法与Lock的release方法很像，具体可以看看我之前写的关于锁的一篇文章。

链接：python同步原语--线程锁 

源码：

解析：

当一个实例请求释放锁的时候，segmaphore内部的_value会自动加1，同时调用notify方法，将被锁住的线程‘唤醒’。

 

 

三、关于Event类

阅读源码知道，Event是也基于Condition和Lock实现的

 

1. set()方法

在 python--线程同步原语 这篇文章我曾经写过一个案例，在进程中调用一次event.set()函数就可以一次性通知（释放）所有阻塞的等待的锁。其内部实现的原理在这里，最关键的一个方法是notify_all()。

在调用set()方法的时候，方法内部会将_flags设置为True，即等待的事件会退出阻塞。

 

2. clear()方法

clear()方法不同作太多解析了，就是内部的_flags重新设置为False

 

3. wait()方法

wait()方法的定义：

def wait(self, timeout=None):

内部实现：

原理还是很简单的，实际上Event的wait()方法正是调用了Condition中的实例方法wait()。调用wait()方法的时候可以传入参数timeout（超时时间），作为Event事件自动退出阻塞的时间界限。