多线程之（信号量，Event，定时器）

Posted 2020-12-19 zoling7

一、信号量（Semaphore）

---

信号量也是一把锁，可以指定信号量为5，对比互斥锁同一时间只能有一个任务抢到锁去执行，信号量同一时间可以有5个任务拿到锁去执行，如果说互斥锁是合租房屋的人去抢一个厕所，那么信号量就相当于一群路人争抢公共厕所，公共厕所有多个坑位，这意味着同一时间可以有多个人上公共厕所，但公共厕所容纳的人数是一定的，这便是信号量的大小。

from threading import Thread,Semaphore,currentThread
import time,random

# 同时只有3个线程可以获得semaphore,即可以限制最大连接数为3
s = Semaphore(3)    # 坑位

def task():
    # s.acquire()
    # print("%s come in." % currentThread().getName())
    # s.release()

    with s:     # 上下文管理，可以省略加锁和释放，效果和上面一样
        print("%s come in." % currentThread().getName())
        time.sleep(random.randint(1,3))     # 模拟每个人上的时间，每个人时间不同

if __name__ == ‘__main__‘:
    for i in range(10):
        t = Thread(target=task)
        t.start()

解析：

# Semaphore管理一个内置的计数器，
# 每当调用acquire()时内置计数器-1；
# 调用release() 时内置计数器+1；
# 计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

与进程池是完全不同的概念，进程池Pool(4)，最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程。

 

二、Event

---

线程的一个关键特性是每个线程都是独立运行且状态不可预测。

如果程序中的其他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。

为了解决这些问题，我们需要使用threading库中的Event对象。

对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。

在初始情况下，Event对象中的信号标志被设置为假。

如果有线程等待一个Event对象，而这个Event对象的标志为假，那么这个线程将会被一直阻塞直至该标志为真。

一个线程如果将一个Event对象的信号标志设置为真，它将唤醒所有等待这个Event对象的线程。

如果一个线程等待一个已经被设置为真的Event对象，那么它将忽略这个事件，继续执行。

from threading import Event

# event.isSet()：返回event的状态值；

# event.wait()：如果 event.isSet()==False将阻塞线程；

# event.set()：设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

# event.clear()：恢复event的状态值为False。

from threading import Thread,Event
import time
# event无非就是实现了（线程之间的同步），一个线程通知另外一个线程，说我已经准备好我的活了，你可以接着运行你其他的任务了

event = Event()
# event.wait()    # 在原地等
# event.set()     # 这个运行算是等完了，发这个信号了，上面的wait就不用等着了

def student(name):
    print("学生%s正在听课" % name)
    event.wait()    # 只要没有人下发下课的信号，就一直卡着
    print("学生%s课间活动" % name)

def teacher(name):
    print("老师%s正在授课" % name)
    time.sleep(7)
    event.set()     # 发信号

if __name__ == ‘__main__‘:
    stu1 = Thread(target=student,args=("娃娃鱼",))
    stu2 = Thread(target=student,args=("托儿所",))
    stu3 = Thread(target=student,args=("儿童劫",))
    tea1 = Thread(target=teacher,args=("菊花信",))

    stu1.start()
    stu2.start()
    stu3.start()
    tea1.start()

event.wait()与event.set()用法

from threading import Thread,Event
import time

event = Event()

def student(name):
    print("学生%s正在听课" % name)
    event.wait(2)    # 里面可以设置超时时间，没必要等到睡7秒
    print("学生%s课间活动" % name)

def teacher(name):
    print("老师%s正在授课" % name)
    time.sleep(7)
    event.set()     # 发信号

if __name__ == ‘__main__‘:
    stu1 = Thread(target=student,args=("娃娃鱼",))
    stu2 = Thread(target=student,args=("托儿所",))
    stu3 = Thread(target=student,args=("儿童劫",))
    tea1 = Thread(target=teacher,args=("菊花信",))

    stu1.start()
    stu2.start()
    stu3.start()
    tea1.start()

event.wait(2)

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

from threading import Thread,Event,currentThread
import time

def conn_mysql():
    n = 1
    while not event.is_set():
        if n > 3:
            raise TimeoutError("链接超时...")

        print("<%s>第%s次尝试链接..." % (currentThread().getName(),n))
        event.wait(0.5)
        n += 1
        print("<%s> 链接成功..." % currentThread().getName())


def check_mysql():
    print("33[36;1m<%s>正在检查...33[0m" % currentThread().getName())
    time.sleep(5)   # 模拟检测时间
    event.set()

if __name__ == ‘__main__‘:
    event = Event()
    for i in range(3):  # 启三个线程
        t = Thread(target=conn_mysql)
        t.start()
    th = Thread(target=check_mysql)
    th.start()

 

三、定时器

---

 定时器：隔多长时间以后触发一个任务的执行。

from threading import Timer

def task(name):
    print("hello %s." % name)

# 里面有几个参数，interval代表时间间隔(秒级)，function代表隔了多少秒之后要运行的那个函数，args，kwargs
t = Timer(5,task,args=("托儿所",))
t.start()
# 5s 后执行 hello 托儿所.

验证码实现：

# 验证码
from threading import Timer
import random

class Code:
    def __init__(self):
        self.make_cache()   # init先执行，立马拿到一个缓存的验证码
        self.check()

    def make_cache(self,interval=60):
        """缓存的功能，缓存每次产生的验证码"""
        self.cache = self.make_code()  # 生成的验证码缓存下来
        print(self.cache)
        self.t = Timer(interval,self.make_cache)    # 每隔60s重新执行一遍，又刷新一遍原来的缓存
        self.t.start()

    def make_code(self,n=4):
        res = ""
        for i in range(n):  # 代表验证码的位数
            s1 = str(random.randint(0,9))    # 随机数转成 str，是为了一会要拼起来
            s2 = chr(random.randint(65,90))  # 65~90的数字，正好是 ASCII表的小写字母到大写字母
            res += random.choice([s1,s2])      # 把s1,s2串起来
        return res

    def check(self):
        while True:
            code = input("请输入你的验证码：").strip().upper()
            if code == self.cache:
                print("验证码输入正确。")
                self.t.cancel()
                break
            else:
                print("验证码输入错误，请重新输入。")
                continue

if __name__ == ‘__main__‘:
    obj = Code()