python线程学习-Condition
状态
这是一种同步机制：一个线程在等待一个特殊的状态量，其他的线程可以发出这个状态量。当一个状态量出现，
一个线程将获得独占资源的锁去访问共享资源
通过生产者/消费者可以更好的描述这种方式，生产者添加一个随机数字到公共列表，而消费者将这个数字在公共
列表中清除。源码可以在threads/condition.py中找到。
看一下生产者类，生产者获得一个锁，添加一个数字，然后通知消费者线程有一些东西可以来清除，最后释放锁定。
在各项动作不段切换的时候，将会触发不定期的随机暂停。
class Producer(threading.Thread):
"""
Produces random integers to a list
"""
def __init__(self, integers, condition):
"""
Constructor.
@param integers list of integers
@param condition condition synchronization object
"""
threading.Thread.__init__(self)
self.integers = integers
self.condition = condition
def run(self):
"""
Thread run method. Append random integers to the integers list
at random time.
"""
while True:
integer = random.randint(0, 256)
self.condition.acquire()
print 'condition acquired by %s' % self.name
self.integers.append(integer)
print '%d appended to list by %s' % (integer, self.name)
print 'condition notified by %s' % self.name
self.condition.notify()
print 'condition released by %s' % self.name
self.condition.release()
time.sleep(1)
下面是消费者类。它获得锁以后，首先检查列表中是否有整数，然后等待来自生产者的通知。当一个元素在列表中被
清除的时候，消费者将释放锁定。
class Consumer(threading.Thread):
"""
Consumes random integers from a list
"""
def __init__(self, integers, condition):
"""
Constructor.
@param integers list of integers
@param condition condition synchronization object
"""
threading.Thread.__init__(self)
self.integers = integers
self.condition = condition
def run(self):
"""
Thread run method. Consumes integers from list
"""
while True:
self.condition.acquire()
print 'condition acquired by %s' % self.name
while True:
if self.integers:
integer = self.integers.pop()
print '%d popped from list by %s' % (integer, self.name)
break
print 'condition wait by %s' % self.name
self.condition.wait()
print 'condition released by %s' % self.name
self.condition.release()
我们在主函数main中创建2个线程然后启动它们：
def main():
integers = []
condition = threading.Condition()
t1 = Producer(integers, condition)
t2 = Consumer(integers, condition)
t1.start()
t2.start()
t1.join()
t2.join()
if __name__ == '__main__':
main()
运行结果可能如下：
$ python condition.py
condition acquired by Thread-1
159 appended to list by Thread-1
condition notified by Thread-1
condition released by Thread-1
condition acquired by Thread-2
159 popped from list by Thread-2
condition released by Thread-2
condition acquired by Thread-2
condition wait by Thread-2
condition acquired by Thread-1
116 appended to list by Thread-1
condition notified by Thread-1
condition released by Thread-1
116 popped from list by Thread-2
condition released by Thread-2
condition acquired by Thread-2
condition wait by Thread-2
线程1将159 添加到公共列表然后通知消费者并释放锁定。线程2获得锁定，清除了159然后释放锁定。生产者在此时仍然在
等待，因为time.sleep(1) 延迟了1秒，所以消费者将再次锁定，然后等待来自生产者的通知。当wait()被调用时，消费者将
解锁资源，生产者就可以重新锁定资源并添加一个新的数字到列表，然后通知消费者。
让我们看一下python实现状态量同步机制的内部原理。如果在condition构造函数中没有定义锁，那么将自动创建一个rlock，
利用这个锁来实现锁定和解锁：
class _Condition(_Verbose):
def __init__(self, lock=None, verbose=None):
_Verbose.__init__(self, verbose)
if lock is None:
lock = RLock()
self.__lock = lock
下面看一下wait()方法。我们假设我们将在没有延迟的情况下调用wait()来简单的说明一下wait()的代码。当状态被置为
锁定的时候，一个名为waiter的锁将被建立。waiter锁被用于生产者释放锁定时通知消费者的联系过程。这个锁对象将会加入
到waiters列表中，并阻塞其他waiter获取锁定。注意状态量锁状态在等待开始时被保存，在wait()返回值时恢复。
def wait(self, timeout=None):
...
waiter = _allocate_lock()
waiter.acquire()
self.__waiters.append(waiter)
saved_state = self._release_save()
try: # restore state no matter what (e.g., KeyboardInterrupt)
if timeout is None:
waiter.acquire()
...
...
finally:
self._acquire_restore(saved_state)
notify()方法被用来解除waiter的锁。生产者调用notify()方法来解除消费者在wait()方法上的阻塞：
def notify(self, n=1):
...
__waiters = self.__waiters
waiters = __waiters[:n]
...
for waiter in waiters:
waiter.release()
try:
__waiters.remove(waiter)
except ValueError:
pass
你也可以使用“with”来模拟锁定和解锁的行为。下面是使用with关键字重写的生产者与消费者类：
class Producer(threading.Thread):
...
def run(self):
while True:
integer = random.randint(0, 256)
with self.condition:
print 'condition acquired by %s' % self.name
self.integers.append(integer)
print '%d appended to list by %s' % (integer, self.name)
print 'condition notified by %s' % self.name
self.condition.notify()
print 'condition released by %s' % self.name
time.sleep(1)
class Consumer(threading.Thread):
...
def run(self):
while True:
with self.condition:
print 'condition acquired by %s' % self.name
while True:
if self.integers:
integer = self.integers.pop()
print '%d popped from list by %s' % (integer, self.name)
break
print 'condition wait by %s' % self.name
self.condition.wait()
print 'condition released by %s' % self.name
这是一种同步机制：一个线程在等待一个特殊的状态量，其他的线程可以发出这个状态量。当一个状态量出现，
一个线程将获得独占资源的锁去访问共享资源
通过生产者/消费者可以更好的描述这种方式，生产者添加一个随机数字到公共列表，而消费者将这个数字在公共
列表中清除。源码可以在threads/condition.py中找到。
看一下生产者类，生产者获得一个锁，添加一个数字，然后通知消费者线程有一些东西可以来清除，最后释放锁定。
在各项动作不段切换的时候，将会触发不定期的随机暂停。
class Producer(threading.Thread):
"""
Produces random integers to a list
"""
def __init__(self, integers, condition):
"""
Constructor.
@param integers list of integers
@param condition condition synchronization object
"""
threading.Thread.__init__(self)
self.integers = integers
self.condition = condition
def run(self):
"""
Thread run method. Append random integers to the integers list
at random time.
"""
while True:
integer = random.randint(0, 256)
self.condition.acquire()
print 'condition acquired by %s' % self.name
self.integers.append(integer)
print '%d appended to list by %s' % (integer, self.name)
print 'condition notified by %s' % self.name
self.condition.notify()
print 'condition released by %s' % self.name
self.condition.release()
time.sleep(1)
下面是消费者类。它获得锁以后，首先检查列表中是否有整数，然后等待来自生产者的通知。当一个元素在列表中被
清除的时候，消费者将释放锁定。
class Consumer(threading.Thread):
"""
Consumes random integers from a list
"""
def __init__(self, integers, condition):
"""
Constructor.
@param integers list of integers
@param condition condition synchronization object
"""
threading.Thread.__init__(self)
self.integers = integers
self.condition = condition
def run(self):
"""
Thread run method. Consumes integers from list
"""
while True:
self.condition.acquire()
print 'condition acquired by %s' % self.name
while True:
if self.integers:
integer = self.integers.pop()
print '%d popped from list by %s' % (integer, self.name)
break
print 'condition wait by %s' % self.name
self.condition.wait()
print 'condition released by %s' % self.name
self.condition.release()
我们在主函数main中创建2个线程然后启动它们：
def main():
integers = []
condition = threading.Condition()
t1 = Producer(integers, condition)
t2 = Consumer(integers, condition)
t1.start()
t2.start()
t1.join()
t2.join()
if __name__ == '__main__':
main()
运行结果可能如下：
$ python condition.py
condition acquired by Thread-1
159 appended to list by Thread-1
condition notified by Thread-1
condition released by Thread-1
condition acquired by Thread-2
159 popped from list by Thread-2
condition released by Thread-2
condition acquired by Thread-2
condition wait by Thread-2
condition acquired by Thread-1
116 appended to list by Thread-1
condition notified by Thread-1
condition released by Thread-1
116 popped from list by Thread-2
condition released by Thread-2
condition acquired by Thread-2
condition wait by Thread-2
线程1将159 添加到公共列表然后通知消费者并释放锁定。线程2获得锁定，清除了159然后释放锁定。生产者在此时仍然在
等待，因为time.sleep(1) 延迟了1秒，所以消费者将再次锁定，然后等待来自生产者的通知。当wait()被调用时，消费者将
解锁资源，生产者就可以重新锁定资源并添加一个新的数字到列表，然后通知消费者。
让我们看一下python实现状态量同步机制的内部原理。如果在condition构造函数中没有定义锁，那么将自动创建一个rlock，
利用这个锁来实现锁定和解锁：
class _Condition(_Verbose):
def __init__(self, lock=None, verbose=None):
_Verbose.__init__(self, verbose)
if lock is None:
lock = RLock()
self.__lock = lock
下面看一下wait()方法。我们假设我们将在没有延迟的情况下调用wait()来简单的说明一下wait()的代码。当状态被置为
锁定的时候，一个名为waiter的锁将被建立。waiter锁被用于生产者释放锁定时通知消费者的联系过程。这个锁对象将会加入
到waiters列表中，并阻塞其他waiter获取锁定。注意状态量锁状态在等待开始时被保存，在wait()返回值时恢复。
def wait(self, timeout=None):
...
waiter = _allocate_lock()
waiter.acquire()
self.__waiters.append(waiter)
saved_state = self._release_save()
try: # restore state no matter what (e.g., KeyboardInterrupt)
if timeout is None:
waiter.acquire()
...
...
finally:
self._acquire_restore(saved_state)
notify()方法被用来解除waiter的锁。生产者调用notify()方法来解除消费者在wait()方法上的阻塞：
def notify(self, n=1):
...
__waiters = self.__waiters
waiters = __waiters[:n]
...
for waiter in waiters:
waiter.release()
try:
__waiters.remove(waiter)
except ValueError:
pass
你也可以使用“with”来模拟锁定和解锁的行为。下面是使用with关键字重写的生产者与消费者类：
class Producer(threading.Thread):
...
def run(self):
while True:
integer = random.randint(0, 256)
with self.condition:
print 'condition acquired by %s' % self.name
self.integers.append(integer)
print '%d appended to list by %s' % (integer, self.name)
print 'condition notified by %s' % self.name
self.condition.notify()
print 'condition released by %s' % self.name
time.sleep(1)
class Consumer(threading.Thread):
...
def run(self):
while True:
with self.condition:
print 'condition acquired by %s' % self.name
while True:
if self.integers:
integer = self.integers.pop()
print '%d popped from list by %s' % (integer, self.name)
break
print 'condition wait by %s' % self.name
self.condition.wait()
print 'condition released by %s' % self.name