python多线程程序设计 之二

线程同步机制

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 = 0

lock_full = Lock()
cv_full = Condition(lock_full)

def q_avail():
    global in_ndx, out_ndx
    if out_ndx == in_ndx:
        return False
    return True

def q_full():
    global in_ndx, out_ndx
    
    if (out_ndx + 1) % Q_SIZE == in_ndx:
        return True

    return False

def q_get():
    global in_ndx, out_ndx
    with 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_SIZE

    with cv_full:
        cv_full.notify()
   
    return v
    
def q_put(v):
    global in_ndx, out_ndx
    
    with 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_SIZE
        q[out_ndx] = v
        cv.notify()

下面代码展示生产者/消费者程序模型，它调用上述的循环队列。

import signal
import sys
import time
import random

from 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_s
        print("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.1
        producer_delay = 0.1
    elif (count == 2):
        producer_delay = float(sys.argv[1])
        consumer_delay = 0.1
    elif (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()