python内置模块threading，线程

一、简介

线程模块，线程同时执行，线程谁快谁先执行；线程是一个单独的执行流，不同的线程实际上不会同时执行，只是线程之间切换间隔较短，看起来是同时执行

二、基本使用

整个py文件为主线程，里面的t1 、t2是子线程，当运行py文件时，主线程开始运行，当遇到线程t1、t2执行时，因解释器是从上往下解释，所以t1会先执行，然后是执行t2，print(‘thread1’)、print(‘thread2’)执行后，因为sleep所以t1和t2并为结束，待sleep结束打印后，程序才会结束

import threading
import timedef thread1(text):print('thread1')time.sleep(1)print(text)def thread2(text):print('thread2')time.sleep(2)print(text)t1 = threading.Thread(target=thread1,args=('thread1_sleep',))
t2 = threading.Thread(target=thread2,args=('thread2_sleep',))t1.start()
t2.start()print('主线程结束')

三、方法

threading.Thread(target,name,args,kwargs) 创建线程，target：执行的目标任务名、name：进程名字、args：以元组方式给执行任务传参、kwargs：以字典方式给执行任务传参

thread.start() 运行线程
thread.join(timeout) 阻塞线程运行，只有当join线程运行结束，才会继续往下执行，但是，不妨碍线程之间的切换
thread.name 设置线程名称，低版本使用thread.setName()，设置主线程名称：thread.name = ‘thread213’

thread.daemon 守护线程，主线程结束，就算有子线程正在运行，程序也会结束，避免出现死循环造成程序无法关闭，低版本使用thread.daemon()，不能和join一起使用，join会阻塞主线程

//因为设置了守护线程，thread1_sleep、thread2_sleep不会被打印import threading
import timelock = threading.Lock()def thread1(text):print('thread1')time.sleep(3)print(text)def thread2(text):print('thread2')time.sleep(5)print(text)t1 = threading.Thread(target=thread1,args=('thread1_sleep',))
t2 = threading.Thread(target=thread2,args=('thread2_sleep',))
t2.daemon = True
t1.daemon = True
t1.start()
t2.start()print('主线程结束')

thread.is_alive() 判断线程是否还活着

threading.Lock() 线程锁：能够自行操控线程切换，使线程变得有序；同步锁：最多只有一个线程持有该锁，被加锁的线程在运行时不会将执行权交出去，只有当该线程被解锁时才会将执行权通过系统调度交由其他线程

lock.locaked()：判断该锁对象是否处于上锁状态
lock.acquire(timeout=1)) ：锁住线程，timeout失效时间

lock.release() ：释放线程

import threading'''
1.该例子去掉线程锁，多执行几次会发现线程1和线程2执行顺序发生错乱；加上线程锁后会发现线程1和线程2会按照顺序执行
2.该例子无论加上线程锁还是去掉线程锁，主线程执行的num结果都会是0，是因为join()会阻塞线程运行，只有当join线程执行完后，才会继续往下运行
3.不使用锁，想保证线程1和线程2顺序在执行，可以在线程2调用start之前，线程1使用join
4.lock线程锁嵌套使用会造成锁死，不经常用
'''lock = threading.Lock()num = 0def thread1():global num# lock.acquire()print(f'线程1开始：num={num}')for i in range(20):num+=iprint(f'线程1：num={num}')print(f'线程1结束：num={num}')# lock.release()def thread2():global num# lock.acquire()print(f'线程2开始：num={num}')for i in range(20):num-=iprint(f'线程2：num={num}')print(f'线程2结束：num={num}')# lock.release()t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)t1.start()
t2.start()
t1.join()
t2.join()print(f'主线程结束：num={num}')//可以使用whith优化
with lock:print(f'线程2开始：num={num}')for i in range(20):num-=iprint(f'线程2：num={num}')print(f'线程2结束：num={num}')

threading.Rlock() 递归锁，RLock允许在同一线程中被多次acquire，Lock不允许

rlock.locaked()：判断该锁对象是否处于上锁状态
rlock.acquire(timeout=1)) ：锁住线程，timeout失效时间

rlock.release() ：释放线程

import threading'''
1.RLock和Lock使用方法一样，唯一区别是Lock嵌套使用会出现锁死
2.thread1使用Lock会锁死，使用Rlock不会
'''lock = threading.RLock()num = 0def thread1():global numlock.acquire()lock.acquire()print(f'线程1开始：num={num}')for i in range(20):num+=iprint(f'线程1：num={num}')print(f'线程1结束：num={num}')lock.release()lock.release()def thread2():global numwith lock:print(f'线程2开始：num={num}')for i in range(20):num -= iprint(f'线程2：num={num}')print(f'线程2结束：num={num}')t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)t1.start()
t2.start()
t1.join()
t2.join()print(f'主线程结束：num={num}')

threading.Condition() 条件锁，条件锁是在递归锁的基础上增加了暂停线程运行的功能，可以使用wait()与notify()来控制线程执行的个数

clock.acquire(timeout=1)) ：锁住线程，timeout失效时间
clock.release() ：释放线程
lockObject.wait(timeout=None) ：将当前线程设置为“等待”状态，只有该线程接到“通知”或者超时时间到期之后才会继续运行，在“等待”状态下的线程将允许系统根据策略自行切换到其他线程中运行
lockObject.wait_for(task, timeout=None) ：和wait 类似，唯一区别是可以传入一个函数，注意：task参数应当传入一个可调用对象，且返回结果为bool类型
lockObject.notify(n=1) ：通知一个当前状态为“等待”的线程继续运行，也可以通过参数n通知多个

lockObject.notify_all()：通知所有当前状态为“等待”的线程继续运行

'''
生产者：厨师
消费者：顾客要求：
1、顾客先下单，下单后分别由不同的厨师接单，且同一种菜不能接单重复
2、厨师做菜、顾客等菜、顾客吃菜之间不能相互影响
3、直到顾客菜吃完程序结束逻辑：
1、先把现有厨师创建好（包括每个厨师会做的菜），并使用条件锁，等待接单
2、顾客开始下单，使用条件锁，结束厨师等待接单（厨师开始接单），订单存在三种状态：0已下单、1已接单、2正在做、3开始上菜、4已上菜、5已吃完，为避免不同厨师
接单重复，需要在厨师接单时，把对应的订单改为已接单状态
3、厨师开始做菜，并且顾客开始等待上菜
4、顾客等待上菜时，开启条件锁，等待上菜
5、厨师做自己的菜时，每做好一道菜，开始给顾客上菜，开启条件锁，结束顾客等待，给顾客上菜，并释放锁，开始做下一道菜（之所以菜做好，开启条件锁，为了避免做菜和等菜相互影响）
6、顾客接到菜，释放锁，并判断订单和上菜数量是否一致，不一致继续等待上菜，一致说明菜已经上完，同时顾客开始吃菜（之所以在接到菜就释放锁，为了避免等菜和吃菜之间相互影响）
7、厨师做完自己的订单，结束工作；顾客吃完自己的菜，结束程序（还可以加上结账、离店、以及店铺座位不够顾客等待排队等逻辑）注意：示例中多次使用set取订单之间的差集和并集，这样会导致多个顾客，下单重复时只会存在一个单子；若需多个顾客，此部分代码不能使用set，需要手动实现 
'''import threading
import time
import randomlock = threading.Condition()#厨师列表
producers= [{'name': '厨师一','food': [{'name':'玉带虾仁','status':0 #0未做菜，1正在做菜}, {'name':'油发豆莛','status':0}, {'name': '红扒鱼翅','status':0}, {'name': '白扒通天翅','status':0}, {'name': '孔府一品锅','status':0}, {'name': '花揽桂鱼','status':0}, {'name': '纸包鸡','status':0}, {'name': '锅烧鸡','status':0}, {'name': '红扒鱼翅','status':0}, {'name': '山东菜丸','status':0}]
}, {'name': '厨师二','food': [{'name': '白菜卷肉','status':0}, {'name': '白汁鱼肚','status':0}, {'name': '板栗烧鸡','status':0}, {'name': '棒棒鸡','status':0}, {'name': '菠饺银肺汤','status':0}, {'name':  '陈皮鳝段','status':0}, {'name': '陈皮兔丁','status':0}, {'name': '虫草鸭子','status':0}, {'name': '川北凉粉','status':0}, {'name': '椿芽烘蛋','status':0}]
}, {'name': '厨师三','food': [{'name': '白菜卷肉','status':0}, {'name': '白汁鱼肚','status':0}, {'name': '板栗烧鸡','status':0}, {'name': '棒棒鸡','status':0}, {'name': '菠饺银肺汤','status':0},{'name': '花揽桂鱼','status':0}, {'name': '纸包鸡','status':0}, {'name': '锅烧鸡','status':0}, {'name': '红扒鱼翅','status':0}, {'name': '山东菜丸','status':0}]
}]#顾客列表
customers = [{'name': '顾客一','food': ['玉带虾仁', '油发豆莛', '红扒鱼翅', '棒棒鸡', '菠饺银肺汤', '陈皮鳝段']
}]#订单列表
order_food = []
order_food_name = []class Producer(threading.Thread):def __init__(self, name, food):super().__init__()self.name = nameself.food = foodprint(f'【{name}】菜单：{food}')def run(self):lock.acquire()print(f'【{self.name}】在等待接单...')lock.wait() #等待接单all_order_name = list(map(lambda order: order['status'] == 0 and order['foodName'], order_food))food_name = list(map(lambda food: food['status'] == 0 and food['name'], self.food))my_order_name = list(set(all_order_name).intersection(set(food_name)))for my_order in my_order_name:order_food_index = order_food_name.index(my_order)#把订单改为已经接单，避免其他厨师重复接单order_food[order_food_index]['status'] = 1if len(my_order_name):print(f'【{self.name}】已经接单 {my_order_name}')lock.release()if len(my_order_name):for order_item in my_order_name:order_food_index = order_food_name.index(order_item)food_index = food_name.index(order_item)#菜还未开始做if order_food[order_food_index]['status'] == 1:print(f'【{self.name}】开始做【{order_item}】')# 改变厨师状态，厨师正在做菜self.food[food_index]['status'] = 1# 改变订单状态，订单正在做order_food[order_food_index]['status'] = 2# 做菜时间time.sleep(random.randint(2, 6))# 厨师已把菜做完self.food[food_index]['status'] = 0lock.acquire()# 订单开始上菜print(f'【{self.name}】已做完【{order_item}】，开始上菜')order_food[order_food_index]['status'] = 3# 上菜定时time.sleep(random.randint(2, 6))lock.notify() #通知顾客菜已经做好lock.release()class Consumer(threading.Thread):def __init__(self, name, food):super().__init__()self.name = name#已经下单的菜self.my_food = food#已经吃完的菜self.eat_food = []#已经上菜的菜self.up_food = []'''下单'''def placeOrder(self):lock.acquire()print(f'【{self.name}】正在下单：{self.my_food}')time.sleep(random.randint(2, 6))for order in self.my_food:order_food.append({'name': self.name,'foodName': order,'status': 0,  # 0已下单，1已接单, 2正在做，3开始上菜，4已上菜，5已吃完})order_food_name.append(order)print(f'【{self.name}】已下单：{self.my_food}')lock.notify_all()  # 通知所有厨师已经下单lock.release()'''等待上菜'''def waitUpFood(self):#订单和上菜数量不一致，说明还有菜没上while len(self.my_food) != len(self.up_food):lock.acquire()wait_food = list(set(self.my_food).difference(self.up_food))print(f'【{self.name}]正在等待上菜{wait_food}')lock.wait()  # 等待上菜# 获取已做完的菜self.up_foo = []for order in order_food:if order['status'] == 3 and order['name'] == self.name:food_name = order['foodName']self.up_food.append(food_name)order_food_index = order_food_name.index(food_name)order_food[order_food_index]['status'] = 4print(f'【{self.name}]已上菜{self.up_food}')lock.release()else:print(f'【{self.name}]菜已上齐')def run(self):self.placeOrder()while True:if len(self.eat_food) == len(self.my_food):print(f'【{self.name}]菜已吃完')breakself.waitUpFood()not_eat_food = list(set(self.up_food).difference(self.eat_food))if len(not_eat_food):for food_name in not_eat_food:print(f'【{self.name}]正在吃【{food_name}】...')# 吃菜定时time.sleep(random.randint(2, 6))print(f'【{self.name}]已经吃完【{food_name}】')order_food_index = order_food_name.index(food_name)order_food[order_food_index]['status'] = 5self.eat_food.append(food_name)threads = []#创建生产者-厨师
for producer in producers:threads.append(Producer(producer['name'], producer['food']))#创建消费者-顾客
for customer in customers:threads.append(Consumer(customer['name'], customer['food']))for thread in threads:thread.start()

threading.Event() 事件锁，事件锁是基于条件锁来做的，它与条件锁的区别在于一次只能放行全部，不能放行任意个数量的子线程继续运行

elock.clear() 将事件锁设为红灯状态，即所有线程暂停运行
elock.is_set() 用来判断当前事件锁状态，红灯为False，绿灯为True
elock.set() 将事件锁设为绿灯状态，即所有线程恢复运行

elock.wait(timeout=None) 将当前线程设置为“等待”状态，只有该线程接到“绿灯通知”或者超时时间到期之后才会继续运行，在“等待”状态下的线程将允许系统根据策略自行切换到其他线程中运行

import threading
import random
import time'''红绿灯'''
class TrafficLight(threading.Thread):def __init__(self,time=60):super().__init__()self.time = timeself.red_light = bool(random.randint(0,1))#根据定时，切换红绿灯，默认60sdef createSignalLight (self):self.red_light = not self.red_lightsignal_light = threading.Timer(self.time, self.createSignalLight)signal_light.start()'''汽车'''
class Car(threading.Thread):def __init__(self):super().__init__()self.car = []#创建车辆，1-3秒随机时间生成车辆def createCar(self):self.car.append({'name':f'汽车{len(self.car)+1}','status':0,#汽车状态，0未到路口，1已到路口，2等待通过，3正在通过，4已通过})signal_light = threading.Timer(random.randint(1,3),self.createCar)signal_light.start()'''路口'''
class DivertedTraffic(TrafficLight,Car):def __init__(self):self.event_lock = threading.Event()TrafficLight.__init__(self)Car.__init__(self)self.createCar()time.sleep(random.randint(1,3))self.createSignalLight()def run(self):signal_light = threading.Timer(1, self.run)signal_light.start()for car in self.car:if car['status'] == 0:car['status'] = 1print(f'【{car["name"]}】已到该路口')if self.red_light:self.event_lock.clear()car['status'] = 2print(f'红灯：【{car["name"]}】等待通过')self.event_lock.wait()self.event_lock.set()car['status'] = 3print(f'【{car["name"]}】正在通过该路口')time.sleep(random.randint(1, 3))car['status'] = 4print(f'绿灯：【{car["name"]}】已通过')thread = DivertedTraffic()thread.run()

threading.Semaphore(value) 信号量锁，用来限制线程执行个数，使用场景：读写文件，读是可以有多个线程线程，而写一般只有一个；爬虫的时候，有时候爬取速度太快了，会导致被网站禁止，所以这个时候就需要控制线程数量来控制爬虫频率

slock.acquire(blocking=True, timeout=1) 上锁，内置计数器-1,直到为0的时候阻塞

slock.release() 解锁，内置计数器+1，并让某个线程的acquire()从阻塞变为不阻塞

import threading
import timedef toilet(sem, name):sem.acquire()time.sleep(3)print(f"当前时间: {time.ctime()}, person_{name} 上厕所完毕!", )sem.release()sem.release()if __name__ == "__main__":sem = threading.Semaphore(3)   # 最多同时运行3个线程for i in range(11):t = threading.Thread(target=toilet, args=(sem, str(i)))t.start()

threading.BoundedSemaphor(value) 信号量锁会检查内部计数器的值，并保证它不会大于初始值，如果超了，就引发一个 ValueError。如：threading.Semaphore例子换成threading.BoundedSemaphor会报错
threading.current_thread() 返回当前线程对象
threading.active_count() 返回当前存活的线程对象的数量
threading.get_ident() 返回线程pid
threading.enumerate() 返回线程数量
threading.main_thread() 返回主线程对象
threading.Timer(interval,func,args,kwargs) 定时器，系统会给Timer的实例化对象分配一条线程，interval 用于设置等待时间、func要执行的函数或方法、args/kwargs该函数或方法要用到的位置参数或关键字参数

timer.cancel() 取消定时

timer.start() 启动定时

import threading
import random
import time'''红绿灯'''
class TrafficLight():def __init__(self,time=60):self.time = timeself.red_light = bool(random.randint(0,1))#根据定时，切换红绿灯，默认60sdef createSignalLight (self):self.red_light = not self.red_lightsignal_light = threading.Timer(self.time, self.createSignalLight)signal_light.start()'''汽车'''
class Car():def __init__(self):self.wait_car = []#创建车辆，1-3秒随机时间生成车辆def createCar(self):self.wait_car.append(f'汽车{len(self.wait_car)+1}')signal_light = threading.Timer(random.randint(1,3),self.createCar)signal_light.start()'''路口'''
class DivertedTraffic(TrafficLight,Car):def __init__(self):TrafficLight.__init__(self,5)Car.__init__(self)self.createCar()time.sleep(random.randint(1,3))self.createSignalLight()def run(self):if self.red_light:print(f'红灯：等待车辆 {len(self.wait_car)} ')else:print(f'绿灯：放行车辆 {len(self.wait_car)} ')self.wait_car=[]signal_light = threading.Timer(1, self.run)signal_light.start()thread = DivertedTraffic()thread.run()

四、其他

class封装线程

import threadingclass MyThread(threading.Thread):def __init__(self,threadName):super().__init__()self.thread_name = threadName#重写run方法def run(self):print(self.thread_name)t1 = MyThread('Thread1')
t2 = MyThread('Thread2')
t1.start()
t2.start()