python多任务处理 --- threading(二)

Event

Event事件，实现简单的线程间的通信机制。内部使用一个flag标记，通过flag标记标记True或者False来判断是否达到了满足的条件，从而选择不同的操作。

import threading

event = threading.Event()  # 创建一个event对象
# do something

print(event.wait())
print("===end====")

当执行到event.wait()时，程序会阻塞住，一直等待event的值被设置为True，阻塞才会终止。使用多线程模拟

import time, threading

event = threading.Event()  # 创建一个event对象

def run():
    time.sleep(3)
    print("event,将被设置")
    event.set()   #执行set设置event状态

threading.Thread(target).start()

print(event.wait())  # 等待，三秒后event的状态被设置，阻塞解除，打印返回值True并继续执行
print("==end==")

===执行结果===
event,将被设置
True
==end==

程序开启子线程并在3秒后将event结果进行设置，将event.wait()的阻塞状态解除，主线程继续执行。

event.wait()方法可以设置超时时间timeout，最多等待timeout时间，在此期间如果event被设置，立即结束阻塞并返回True，否则timeout时间结束，同样结束等待，但是返回False。

常用的方法有

方法	含义
set()	设置该event标记为True
clear()	设置该event标记为False
is_set()	判断标记是否为True
wait(timeout=None)	等待timeout秒，在此期间被设置为True，立即结束阻塞并返回True，否则时间结束阻塞解除返回False，timeout=None一直阻塞

Timer(延迟执行)

该类是Thread的子类，重写了run方法并新增cancal方法。当创建一个线程对象时，可以设置一个延迟时间interval。当线程start后，会创建一个新的线程，在延迟interval秒时间后，才调用targat函数执行。在target函数执行前，可以调用Timer对象的cancal方法取消执行该函数。

LOCK

全局锁对于各个线程来说是公共的资源，当一条线程获取了锁(通过.acquier()方法)，其他的线程在执行acqiure()方法时，将会阻塞等待锁释放。只有获取了锁的线程才能继续执行，获取锁的线程执行完操作后，需要将该锁释放（.release()），释放后锁后，其他等待acquire()获取锁的线程对锁进行争抢，最后其中一个线程会获得该锁。其余线程继续阻塞等待再一次释放。

基本使用

通过实例化threading.Lock类，可以获得一个锁对象，每次实例化获得的锁对象是不同的。程序中可以有多个锁

import threading
# lock1 和 lock2 是两个不同的锁，相互之间不会影响。加锁和解锁阻塞针对同一个锁对象
lock1 = threading.Lock()
lock2 = threading.Lock()

lock1.acquier()

lock1().acquire(timeout=4)   # 已经加锁，4秒后超时继续执行，但是本次返回值为False.

lock1.release()    # 将锁释放

lock2.acquier()    # lock2获取并加锁
lock2.acquier()    # 尝试获取锁，但是不会有release操作，程序永远阻塞，死锁

名称	含义
acquier(blocking=True, timeout=-1)	获取一个锁，默认阻塞等待直到获取到锁；block是否等待，timeout设置等待时间。在设置的时间内获得该锁，返回True，超时返回False
release()	释放这个锁对象，使其可以再次被获取，连续释放将会报错
locked()	判断该锁是否上锁

解决线程不安全问题

加锁就可以解决多条线程操作同一个资源时争抢资源而导致错误发生的问题。

import threading, time

lock = threading.Lock()
x = 0
def add():
    global x
    for i in range(1000000):
        lock.acquier()  # 加锁，加锁线程才能操作
        x += 1
        lock.release()  # 释放锁，释放后才能再次acquier
for i in range(5):
    threading.Thread(target=add).start()

while threading.active_count() != 1:
    time.sleep(1)
print(x)

======输出结果=======
5000000

上面的程序在执行x += 1操作时始终只有一条线程。这是一个原子操作，这样可以保证在读取x值后完成x+1，再将x的值写入x，将不会造成之前线程不安全的结果。

死锁

当一个线程获取该锁后，未释放锁，又对该锁进行acquier()操作，尝试获取，此将无法打开。

import threading

lock = threading.Lock(

lock2.acquier()   
lock2.acquier()

当有两个锁时，A等待B解锁才能解锁，同时B等待A解锁才能解锁，就产生了死锁

import threading

lock1 = threading.Lock()
lock2 = threading.Lock()

def func1():
    lock1.acquier()
    
    if lock2.acquier():  # 等待lock2释放后才能进入释放lock1
        lock1.release()

def func2():
    lock2.acquier()
    
    if lock1.acquier():  # 等待lock1释放后才能释放lock2
        lock2.release()

t1 = threading.Thread(target=func1)
t2 = threading.Thread(target=func2)
t1.start()
t2.start()

 当一个线程获得锁后，该线程由于产生了异常造成了线程的终止，该锁无法被释放

import threading

lock = threading.Lock()

def func1():
    try:
        lock.acquier()
        1/0              
    finally:
        lock.release()

    # 或者直接使用锁的上下文管理
    with lock:  # enter中加锁，exit中解锁
        1/0
threading.Thread(target=func1).start()
threading.Thread(target=func1).start()

非阻塞锁的使用

使用lock看互斥锁后，当一个线程想要获取锁时，如果该锁已被别的线程获取，该线程将会发生阻塞，该线程进入休眠状态。使用非阻塞的方式，可以让该线程在获取不到锁的情况下进行一些其他的操作。而不是进入休眠

import threading, time

def run():
    if lock.acquire(False):  # 获取到该锁
        # 执行正常的流程
        time.sleep(0.1)
        lock
    else:
        # 执行其他的流程
        time.sleep(0.1)
for i in range(5):
    threading.Thread(target=run).start()

在一条线程抢到锁后，其他四条线程并不会阻塞，而是执行其他事情，但是在做其他事情时，不要对加锁线程处理的数据进行操作，否则可能引起线程安全问题，将失去加锁的意义。

RLOCK

RLock是一个线程相关的锁，该锁一旦被一个线程获取后，其他线程不能获取该锁。

RLOCK在同一个线程可以多次获取并对应的多次释放，该线程获取该锁后，在释放前，该线程就是该锁的属主。当其他线程访问该RLOCK时，将不能使用这个已经拥有属主的锁。在属主线程中，释放和加锁时相同的次数才能将这个锁全部释放，这个锁的属主才能解除，解除后，其他线程可以获取该锁并成为这个锁的新属主。

import threading, time

r = threading.RLock()

print(r.acquire())  # True
print(r)            # <locked _thread.RLock object owner=10928 count=1 at 0x000001F4E8371378>
print(r.acquire(False )) # True 非阻塞再次获取，该线程是属主，直接获得

print(r.release())       # None   释放一次，count 计数减一

def run():
    print("aaaa")
    print(r.acquire(False))  # 此时主线程未释放，无法获取，返回False
    print(r.acquire(), "========") # 阻塞等待，3秒后主线程释放，返回True，获取成功
    print(r)  # <locked _thread.RLock object owner=6868 count=2 at 0x000001F4E8371378>  owner属主改变为子线程

for i in range(1):
    threading.Thread(target=run, name="Thread-%d"%(i+1)).start()

time.sleep(3)         # 主线程3秒后才完全释放RLOCK,此时子线程在等待该锁被释放
print(r.release())

在Rlock的输出信息中， owner为属主线程id，count记录了该锁在属主线程中加锁的次数。继续释放后，全部释放后，该锁为 unlocke未被任何线程占用。该锁才可以被所有线程

Condition

使用场景：多个线程阻塞等待一个condition条件（wait），当其中一条线程发出该通知（notify()），所有在等待该condition的线程被激活执行。

创建一个全局的condition对象后，多个线程都可以使用c.acquire()来获取这个condition对象，也可以通过c.release()释放这个对象，当获取对象后，使用c.wait()可以阻塞这个线程并等待其他拥有该对象线程的通知。

Conditon 默认使用RLock作为参数创建，也可以手动传入lock

con = threading.Condition(threading.Lock)

Rlock和lock的区别在于Rlock可以多次获取(acquire)这个condition，不会发生死锁，最后对应多次释放即可。lock只能获取一次，之后需要进行释放操作，否则将一直阻塞。

示例：多条子线程获取condition等待通知，3秒后主线程获取该锁，并通知所有以获取该信号的线程结束阻塞。

con = threading.Condition()

def fun(con:threading.Condition):
    print("1234")
    with con:   # 在线程中对该锁必须先获取然后释放，使用了con的上下文管理进行acquire和release
        con.wait()
    print("++++")


for i in range(5):
    threading.Thread(target=fun, args=(con, )).start()

time.sleep(3)
with con:             # 3秒后 获取锁对象
    con.notify_all()  # 通知所有等待该condition的线程。该线程release后其他线程才能获取信号
    print("main")
    time.sleep(2)     # 先打印“main”,等待2秒，该con 释放，子线程才会结束等待
print("===end===")

使用notify_all()方法会通知所有等待的线程，也可以调用notify(count)指定可以接受消息线程的个数，只有接受到消息的线程结束阻塞，继续执行。这两种方式类似于广播(全体)和多播(多个成员，非全体)。

semaphore 和 Boundedsemaphore

semaphore是一个信号量，信号量可以初始化为一个指定数值，每次获取(acquire())一个信号量将会减一，当信号量值为0时继续acquire()将会造成阻塞。使用release可以增加一个信号量，增加的数值可以超过起始设置值，起始默认为1。

import threading

sema = threading.Semaphore(2)

sema.acquire()
sema.acquire()

# sema.acquire() # 执行该句将会阻塞，该线程无法继续运行

for i in range(10):
    sema.release()          # release 10次不会报错或者阻塞，每次信号量加1

print("===end===")

Boundedsemaphore对象会将信号量范围限制在指定范围内（0 - maxsize），信号量为0时继续acquire()会阻塞，而达到maxsize时继续release()将会报错。

import threading

sema = threading.BoundedSemaphore(2)

sema.acquire()  # -1
sema.acquire()  # -1

sema.release()  # +1
sema.release()  # +1
sema.release()  # 将会报错，释放

例:使用BoundedSemaphore实现对列表长度控制，并实现线程安全

class Collection:
    def __init__(self, maxsize):
        self._lst = []
        self._BS = threading.BoundedSemathore(maxsize)

    def get_value(self):       # 从列表中拿出一个元素
        self._BS.acquire()      # 当lst没有元素时，信号量也会为0，此时会阻塞，保证pop不会报错
        return self.lst.pop()  # 弹出一个元素返回
    
    def return_value(self, value)    # 将这个元素放回
        self.lst.append(value)       
        # 先append放回，再将信号量加一，如果信号量加一时报错，书名列表超界，弹出最后一个元素再抛出异常
        try:
            self._BS.release()
        except:
            self.lst.pop()
            raise error("列表已满")

get_value和return方法实现了从列表"拿走"和"归还"的操作，当使用线程同时进行添拿走或归还时，就可能发生线程安全问题，例如多条线程获取元素，可能在某一刻pop时列表已经空。我们可议加锁解决这个问题，在此使用了边界信号量来控制这个列表的长度，当信号量为0时再次获取会阻塞，这也避免了pop一个空列表的情况发生。