python多线程程序设计 之二
- 线程同步机制
- lock对象
- acquire
- release
- locked
- RLock对象
- 条件变量
- 条件变量应用实列
- 实列代码
线程同步机制
lock对象
原语锁是一种同步原语,锁定时不属于特定线程。在Python中,它是目前可用的最低级别的同步原语,由_thread扩展模块直接实现。
原始锁处于两种状态之一
- “锁定”
- “解锁”
lock对象有两个基本方法:acquire() 和release()。
当状态解锁时,acquire()将状态更改为锁定,并立即返回。当状态被锁定时,acquire()会阻塞,直到另一个线程中对release()的调用将其更改为解锁状态,然后acquire()调用将其重置为锁定状态并返回。
release()方法只能在锁定状态下调用;它将状态更改为解锁,并立即返回。如果尝试释放未锁定的锁,则会引发运行时错误。
当多个线程阻塞在 acquire() 中等待状态转为解锁状态时,当调用 release() 将状态重置为解锁时,只有一个线程继续执行;哪一个等待线程继续进行是未定义的,并且可能因实现而异。
所有方法均以原子方式执行。
acquire
acquire(blocking=True, timeout=-1)
获取锁,阻塞或非阻塞。
当参数blocking为True(默认值)时,acquire将阻塞,直到锁解锁,然后将其设置为锁定,并返回True。
当参数blocking为 False时,不阻塞,返回 False;否则,将锁设置为锁定,并返回 True。
当浮点参数timeout设置为正值时,如果一直无法获取锁,则,最多会阻塞timeout指定的秒数。如果参数timeout是-1,则 指定无限等待。当blocking为False时,禁止指定参数timeout。
如果成功获取锁,则返回值为 True,否则返回值为 False。
release
release()
释放锁。这可以从任何线程调用,而不仅仅是已获取锁的线程。
当锁被锁定时,将其重置为解锁,然后返回。如果任何其他线程在等待锁解锁时,被阻塞,则只允许其中一个线程继续进行。
当在未锁定的锁上调用时,会引发 RuntimeError。
locked
locked()
如果已经获取了锁,则返回True。
RLock对象
可重入锁是一种同步原语,同一线程可以多次获取它。在内部,除了原始锁使用的锁定/解锁状态之外,它还使用“所属线程”和“递归级别”的概念。在锁定状态下,某个线程拥有锁;在解锁状态下,没有线程拥有它。
线程调用锁的 acquire() 方法加锁,并调用其 release() 方法解锁。
可重入锁支持上下文管理协议,因此建议使用with而不是手动调用acquire()和release()来处理代码块的锁获取和释放。
RLock 的 acquire()/release() 调用对可以嵌套,这与 Lock 的 acquire()/release() 不同。只有最后的release()(最外层对的release())将锁重置为解锁状态,并允许在acquire()中阻塞的另一个线程继续进行。
acquire()/release() 必须成对使用:每次获取都必须在已获取锁的线程中释放一次。未能多次调用释放来获取锁,可能会导致死锁。
条件变量
条件变量总是与某种类型的锁相关联;锁可以作为参数传入,也可以默认创建一个。当多个条件变量必须共享同一锁时,传入一个很有用。锁是条件对象的一部分,所以不必单独跟踪它。
条件变量遵循上下文管理协议:使用 with 语句在封闭块的持续时间内获取关联的锁。 acquire() 和release() 方法也会调用关联锁的相应方法。
该类的其他方法必须与相关联的锁一起调用。
- wait()方法释放锁,然后阻塞,直到另一个线程通过调用notify()或notify_all()唤醒它。一旦被唤醒,wait()重新获取锁并返回。还可以指定超时。
- notify() 方法会唤醒等待条件变量的线程之一。 notify_all() 方法唤醒所有等待条件变量的线程。
notify()和notify_all()方法不会释放锁;这意味着被唤醒的一个或多个线程不会立即从其 wait() 调用中返回,而是仅在调用 notify() 或 notify_all() 的线程最终释放锁所有权时,唤醒的线程才能返回。
使用条件变量的典型编程风格使用锁来同步对某些共享状态的访问;对特定状态更改感兴趣的线程会重复调用 wait() ,直到看到所需的状态,而修改状态的线程在以可能的方式更改状态时,调用 notification() 或 notify_all(),而这个状态正是某个等待的线程期望的状态。
threading.Condition(lock=None)
此类实现条件变量对象。条件变量允许一个或多个线程等待,直到收到另一线程的通知。
如果给出了锁参数而不是 None,则它必须是 Lock 或 RLock 对象,并且它被用作底层锁。否则,将创建一个新的 RLock 对象并将其用作基础锁。
条件变量应用实列
这个实列演示,生产者/消费者程序模型如何使用条件变量同步多线程程序运行。
通过下列的命令行
python multi_thread_app.py 0.5 0.1
你将看到下列的显示
get wait in_ndx: 0 out_ndx: 0
说明消费者处于饥饿状态,通过等待,实现与生产者同步。
通过下列的命令行
python multi_thread_app.py 0.1 0.5
你将看到大量下列的显示
put wait in_ndx: 1 out_ndx: 0
说明生产者处于等待状态,通过等待,实现与消费者同步。
实列代码
下列代码使用条件变量,实现一个循环数组队列。
- 当队列空时,取数据线程等待
- 当队列满时,存数据线程等待
from threading import *
from queue import *lock = Lock()
cv = Condition(lock)Q_SIZE = 6
q = [x for x in range(Q_SIZE)]
in_ndx = 0
out_ndx = 0lock_full = Lock()
cv_full = Condition(lock_full)def q_avail():global in_ndx, out_ndxif out_ndx == in_ndx:return Falsereturn Truedef q_full():global in_ndx, out_ndxif (out_ndx + 1) % Q_SIZE == in_ndx:return Truereturn Falsedef q_get():global in_ndx, out_ndxwith cv:while not q_avail():print("get wait in_ndx: {0} out_ndx: {1}".format(in_ndx, out_ndx))cv.wait()v = q[in_ndx]in_ndx = (in_ndx + 1) % Q_SIZEwith cv_full:cv_full.notify()return vdef q_put(v):global in_ndx, out_ndxwith cv_full:while q_full():print("put wait in_ndx: {0} out_ndx: {1}".format(in_ndx, out_ndx))cv_full.wait()with cv:out_ndx = (out_ndx + 1) % Q_SIZEq[out_ndx] = vcv.notify()
下面代码展示生产者/消费者程序模型,它调用上述的循环队列。
import signal
import sys
import time
import randomfrom threading import *
from cond_vars import *def signal_handler(sig, frame):print('You pressed Ctrl+C!')sys.exit(0)signal.signal(signal.SIGINT, signal_handler)class Consumer(Thread):def __init__(self, delay_s):super(Consumer,self).__init__()self.delay_s = delay_sprint("Consumer")def run(self):while True:v = q_get()time.sleep(self.delay_s)class Producer(Thread):def __init__(self, delay_s):super(Producer, self).__init__()self.delay_s = delay_s;print("Producer") def run(self):while True: v_list = random.sample(range(1, 5000), 10)for v in range(len(v_list)):time.sleep(self.delay_s)q_put(v_list[v])if __name__ == "__main__":print("Here")count = len(sys.argv)if ( count == 1):consumer_delay = 0.1producer_delay = 0.1elif (count == 2):producer_delay = float(sys.argv[1])consumer_delay = 0.1elif (count >= 3):producer_delay = float(sys.argv[1])consumer_delay = float(sys.argv[2])print("consumer delay: {0} seconds".format(consumer_delay))print("producer delay: {0} seconds".format(producer_delay))t1 = Producer(producer_delay);t2 = Consumer(consumer_delay)t1.start()t2.start()t1.join()t2.join()