Python之路(第四十五篇)线程Event事件 条件Condition定时器Timer线程queue

Posted 2022-09-16 nicholas0707

一、事件Event

Event（事件）：事件处理的机制：全局定义了一个内置标志Flag，如果Flag值为 False，那么当程序执行 event.wait方法时就会阻塞，如果Flag值为True，那么event.wait 方法时便不再阻塞。

 

Event其实就是一个简化版的 Condition。Event没有锁，无法使线程进入同步阻塞状态。

Event()

• set(): 将标志设为True，并通知所有处于等待阻塞状态的线程恢复运行状态。

• clear(): 将标志设为False。

• wait(timeout): 如果标志为True将立即返回，否则阻塞线程至等待阻塞状态，等待其他线程调用set()。

• isSet(): 获取内置标志状态，返回True或False。

 

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。

执行过程：在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行。

例子

使用场景：

有多个工作线程尝试链接mysql，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作

from threading import Event,Thread
import time
import random
?
?
def connect_db(e):
    time.sleep(random.randint(0,3))
    e.set()
?
def check_web(e):
    count = 1
    time.sleep(1)
    while count < 4:
        e.wait(0.5)  # 状态为False的时候,只等待0.5s就结束
        if e.is_set():
            print("第%s次链接数据库成功！"%count)
            break
        else:
            print("第%s次链接数据库失败！"%count)
            count += 1
    else:
        raise TimeoutError("数据库链接超时")
?
e = Event()
t1 = Thread(target=check_web,args=(e,))
t2 = Thread(target=connect_db,args=(e,))
t1.start()
t2.start()

　　

二、 条件Condition

可以把Condition理解为一把高级的琐，它提供了比Lock, RLock更高级的功能，允许我们能够控制复杂的线程同步问题。threadiong.Condition在内部维护一个琐对象（默认是RLock），可以在创建Condigtion对象的时候把琐对象作为参数传入。

Condition也提供了acquire, release方法，其含义与琐的acquire, release方法一致，其实它只是简单的调用内部琐对象的对应的方法而已。

Condition还提供了如下方法(特别要注意：这些方法只有在占用琐(acquire)之后才能调用，否则将会报RuntimeError异常。)：

 

• Condition.wait([timeout]):

  wait方法释放内部所占用的琐，同时线程被挂起，直至接收到通知被唤醒或超时（如果提供了timeout参数的话）。当线程被唤醒并重新占有琐的时候，程序才会继续执行下去。

• Condition.notify():

  唤醒一个挂起的线程（如果存在挂起的线程）。注意：notify()方法不会释放所占用的琐。

• Condition.notify_all() Condition.notifyAll()

  唤醒所有挂起的线程（如果存在挂起的线程）。注意：这些方法不会释放所占用的琐。

 

例子

 

?

?
from threading import Thread,Condition
?
def func(con,i):
    con.acquire()
    con.wait()
    print("在执行第%s个线程"%i)
    con.release()
?
?
con = Condition()
for i in range(10):
    Thread(target=func,args=(con,i)).start()
while True:
    num = input(">>>").strip()
    if num.isdigit():
        num = int(num)
    else:break
    con.acquire()
    con.notify(num) #通知有几个线程可以执行
    con.release()
?
# Python提供的Condition对象提供了对复杂线程同步问题的支持。
# Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，
# 还提供了wait和notify方法。线程首先acquire一个条件变量，然后判断一些条件。
# 如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，
# 通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件。
# 不断的重复这一过程，从而解决复杂的同步问题。

　　

?

分析：在程序运行过程中输入数字，输入多少个数字就唤醒多少个线程执行for循环里的线程。

 

三、定时器Timer

Timer:隔一定时间调用一个函数,如果想实现每隔一段时间就调用一个函数的话，就要在Timer调用的函数中，再次设置Timer。Timer是Thread的一个派生类.

 

Timer继承子Thread类，是Thread的子类，也是线程类，具有线程的能力和特征。这个类用来定义多久执行一个函数。

它的实例是能够延迟执行目标函数的线程，在真正执行目标函数之前，都可以cancel它。

Timer()

interval 第一个参数传 间隔时间

function 传执行任务的函数 隔了多少秒后执行这个函数

给函数传参方式 args kwargs

 

Timer用的是Thread模块，每启动一个定时器，启动一个线程

 

 

例子

 

?
from threading import Timer
import time
?
?
def add(x, y):
    print(x + y)
?
?
t = Timer(10,add,args=[4,5,])
t.start()
?
time.sleep(2)
t.cancel()
print("===end===")

　　

?

　分析：start方法执行之后，Timer对象会处于等待状态，等待10秒之后会执行add函数。同时，在执行add函数之前的等待阶段，主线程使用了子线程的cancel方法，就会跳过执行函数结束。

 

例子

?

# 定时器，指定n秒后执行某个操作
from threading import Timer
import time
?
def func():
    print("执行时间同步")
?
?
while True:
    Timer(5,func).start() #注意这个线程是异步的，先执行这句，然后马上就执行time.sleep()，
    # timer在等待5秒后执行func,而time.sleep()此时也等待了5秒，再次执行timer()
    time.sleep(5)

　　

 

四、线程queue

queue 模块下提供了几个阻塞队列，这些队列主要用于实现线程通信。在 queue 模块下主要提供了三个类，分别代表三种队列，它们的主要区别就在于进队列、出队列的不同。

关于这三个队列类的简单介绍如下：

• Queue.qsize()：返回队列的实际大小，也就是该队列中包含几个元素。

• Queue.empty()：判断队列是否为空。

• Queue.full()：判断队列是否已满。

• Queue.put(item, block=True, timeout=None)：向队列中放入元素。如果队列己满，且 block 参数为 True（阻塞），当前线程被阻塞，timeout 指定阻塞时间，如果将 timeout 设置为 None，则代表一直阻塞，直到该队列的元素被消费；如果队列己满，且 block 参数为 False（不阻塞），则直接引发 queue.FULL 异常。

• Queue.put_nowait(item)：向队列中放入元素，不阻塞。相当于在上一个方法中将 block 参数设置为 False。

• Queue.get(item, block=True, timeout=None)：从队列中取出元素（消费元素）。如果队列已满，且 block 参数为 True（阻塞），当前线程被阻塞，timeout 指定阻塞时间，如果将 timeout 设置为 None，则代表一直阻塞，直到有元素被放入队列中； 如果队列己空，且 block 参数为 False（不阻塞），则直接引发 queue.EMPTY 异常。

• Queue.get_nowait(item)：从队列中取出元素，不阻塞。相当于在上一个方法中将 block 参数设置为 False。

• Queue.Queue(maxsize=0) #FIFO， 用来定义队列的长度，如果maxsize小于1就表示队列长度无限,

• Queue.LifoQueue(maxsize=0) #LIFO， 如果maxsize小于1就表示队列长度无限

• task_done()#意味着之前入队的一个任务已经完成。由队列的消费者线程调用。每一个get()调用得到一个任务，接下来的task_done()调用告诉队列该任务已经处理完毕如果当前一个join()正在阻塞，它将在队列中的所有任务都处理完时恢复执行（即每一个由put()调用入队的任务都有一个对应的task_done()调用）。

• join()#阻塞调用线程，直到队列中的所有任务被处理掉。只要有数据被加入队列，未完成的任务数就会增加。当消费者线程调用task_done(（意味着有消费者取得任务并完成任务），未完成的任务数就会减少。当未完成的任务数降到0，join()解除阻塞。

 

三种不同的队列类

q=Queue(maxsize):创建一个FIFO(first-in first-out,先进先出)队列。maxsize是队列中金额以放入的项的最大数量。
如果省略maxsize参数或将它置为0，队列大小将无穷大。
?
q=LifoQueue(maxsize)：创建一个LIFO（last-in first-out，后进先出）队列(栈)。
?
q=PriorityQueue(maxsize)：创建一个优先级队列，其中项按照优先级从低到高依次排好。
使用这种队列时，项应该是(priority,data)形式的元组，其中priority时一个标识优先级的数字。

例子

# 进程中使用队列 from multiprocessing import Queue
import queue
#
# q1 = queue.Queue() #先进先出
# q1.get_nowait() #不阻塞直接取，如果取时是空则报错
# q1.put_nowait(2)  #不阻塞直接存，如果存时是满的则报错
# q1.get()
# q1.put(1)
?
?
q2 = queue.LifoQueue() #其实是一个栈，先进后出
q2.put("a")
q2.put("b")
q2.put("c")
print(q2.get())
?
q3 = queue.PriorityQueue()  #优先级队列，根据设置的优先级确定取出的对象
q3.get((20,"a"))
q3.get((10,"c"))
q3.get((0,"b"))
q3.get((-10,"d"))  #数字越小越优先取出来
q3.get((30,"e"))
print(q3.get())

　　

?

生产者消费者模型

import threading
import time
from queue import Queue
?
?
def put_id():
    i = 0
    while True:
        i = i + 1
        print("添加数据", i, id_queue.qsize())
        time.sleep(1)
        id_queue.put(i)
?
?
def get_id(m):
    while True:
        i = id_queue.get()
        print("线程", m, ‘取值‘, i)
?
?
if __name__ == "__main__":
    id_queue = Queue(10)
?
    Th1 = threading.Thread(target=put_id, )
    Th2 = threading.Thread(target=get_id, args=(2,))
    Th3 = threading.Thread(target=get_id, args=(3,))
    Th5 = threading.Thread(target=get_id, args=(4,))
    Th4 = threading.Thread(target=get_id, args=(5,))
?
    Th1.start()
    Th2.start()
    Th3.start()
    Th4.start()
    Th5.start()