文章目录
- 一、基础概念
- 二、同步锁
- 三、线程死锁和递归锁
- 四、同步条件(event)
- 五、信号量
- 六、线程队列(queue)
- 1、常用方法
- 2、queue模块的三种模式
- (1)FIFO队列
- (2)LIFO队列
- (3)按优先级
- 七、生产者消费者模型
一、基础概念
①并发: 系统在一段时间内,处理多个任务的能力
②并行: 系统在同一时刻,处理多个任务的能力
③同步: 当进程执行到一个IO(等待外部数据)的时候,等待,不继续向下运行。
④异步: 当进程执行到一个IO(等待外部数据)的时候,不等待,而是先执行其他代码,一直到数据接收成功,再回来处理。
⑤GIL: 全局解释锁。同一时刻,在同一进程下,如果有多个线程,但CPU只能执行其中一个。
⑥任务分两种: IO密集型和计算密集型。
对于IO密集型的任务,python的多线程可以加快速度,可以采用多进程+协程处理。
对于计算密集型的任务,python的多线程反而因为切换的开销,增加处理时间。
二、同步锁
同步锁也叫互斥锁
(1)未加同步锁情形
import threading
import timedef sub():global numtemp = numtime.sleep(0.001)num = temp -1num = 100l = []for i in range(100):t = threading.Thread(target=sub)t.start()l.append(t)for t in l:t.join()print(num) # 98 输出结果不确定
上述代码中,各个线程在时间片轮转时,自己可能还没执行完,造成num值还未改变,就切换到其他线程了。所以num的值输出不固定(取决于cpu的执行速度)
(2)加同步锁情形
import threading
import timedef sub():global numlock.acquire() # 申请同步锁temp = numtime.sleep(0.001)num = temp - 1lock.release() # 释放同步锁num = 100l = []
lock = threading.Lock() # 创建同步锁对象for i in range(100):t = threading.Thread(target=sub)t.start()l.append(t)for t in l:t.join()print(num) # 0
在执行 lock.acquire() 后,线程不允许被切换,在执行 lock.release() 后,才允许切换线程。
三、线程死锁和递归锁
在线程间共享多个资源的时候,如果两个线程分别占有一部分资源并且同时等待对方的资源,就会造成死锁。这两个线程在无外力作用下将一直等待下去。例:
import threading
import timeclass MyThread(threading.Thread):def actionA(self):A.acquire()print(self.name,"actionA gotA",time.strftime("%X"))time.sleep(1)B.acquire()print(self.name, "actionA gotB", time.strftime("%X"))time.sleep(1)B.release()A.release()def actionB(self):B.acquire()print(self.name, "actionB gotB", time.strftime("%X"))time.sleep(1)A.acquire()print(self.name, "actionB gotA", time.strftime("%X"))time.sleep(1)A.release()B.release()def run(self):self.actionA()self.actionB()if __name__ == '__main__':A = threading.Lock()B = threading.Lock()L = []for i in range(3):t = MyThread()t.start()L.append(t)for i in L:i.join()print("ending...")
第一个线程在执行完actionA方法的B.release()时,第二个线程开始执行actionA方法的A.acquire()。然后两个线程并发运行,直到第一个线程执行actionB方法的A.acquire(),第二个线程执行actionA方法的B.acquire,形成死锁。
解决方法——递归锁
递归锁可以看成有个计数器默认count=0,执行acquire时,count+1;执行release时,count-1。只有当count=0时,才允许新的线程执行acquire后续语句。
import threading
import timeclass MyThread(threading.Thread):def actionA(self):r_lock.acquire() # count=1print(self.name,"actionA gotA",time.strftime("%X"))time.sleep(1)r_lock.acquire() # count=2print(self.name, "actionA gotB", time.strftime("%X"))time.sleep(1)r_lock.release() # count=1r_lock.release() # count=0def actionB(self):r_lock.acquire()print(self.name, "actionB gotB", time.strftime("%X"))time.sleep(1)r_lock.acquire()print(self.name, "actionB gotA", time.strftime("%X"))time.sleep(1)r_lock.release()r_lock.release()def run(self):self.actionA()self.actionB()if __name__ == '__main__':r_lock = threading.RLock()L = []for i in range(3):t = MyThread()t.start()L.append(t)for i in L:i.join()print("ending...")
四、同步条件(event)
event是一个简单的同步对象。
event可以使两个线程同步。
event = threading.Event() 创建同步条件对象
执行event.wait()的线程将等待flag被设定,而阻塞。event.set() 设定flag。当flag被设定的时候,执行event.wait()的线程将不被阻塞,相当于pass。当执行event.clear(),flag将被清除,event.wait()将继续阻塞。多个线程可以等候同一个event对象。
import threading,time
class Boss(threading.Thread):def run(self):print("BOSS:今晚大家都要加班到22:00。")print("first:",event.isSet())# Falseevent.set()time.sleep(3)print("BOSS:<22:00>可以下班了。")print("second:",event.isSet()) # Falseevent.set()class Worker(threading.Thread):def run(self):event.wait()# 一旦flag被设定,event等同于passprint("Worker:哎……命苦啊!")time.sleep(1)event.clear()event.wait()print("Worker:OhYeah!")if __name__=="__main__":event=threading.Event()threads=[]for i in range(5):threads.append(Worker())threads.append(Boss())for t in threads:t.start()for t in threads:t.join()print("ending.....")
五、信号量
Semaphore(count)设定一个计数器count,然后每执行acquire()时,count-1,执行release()时,count+1。当count=0时,再执行acquire()将阻塞线程。相当于同步锁的一个扩展,同步锁的count最大等于1.。
import threading,timeclass myThread(threading.Thread):def run(self):if semaphore.acquire(): # 允许5个线程同时进print(self.name)time.sleep(2)semaphore.release() # 5个线程同时释放锁if __name__=="__main__":semaphore=threading.Semaphore(5) # 相当于申请了5把锁thrs=[]for i in range(100):thrs.append(myThread())for t in thrs:t.start()
六、线程队列(queue)
同一个进程下的多个线程,共享该queue数据
1、常用方法
put(item): 在队列队尾插入一个item。其参数block默认为True,如果队列已满,将阻塞线程。设置block=False,队列已满,将不会阻塞线程,而是引发Full异常。
get(): 将一个值从队列中取出,其参数block默认为True,如果队列已空,将阻塞线程。设置block=False,队列已空,将不会阻塞线程,而是引发Empty异常。
qsize(): 返回队列大小(实际存储数据的个数)
empty(): 如果队列为空,返回True,反之False
full(): 如果队列已满,返回True,反之False
full: 与maxsize大小对应
get_nowait(): 相当于get(block=False)
**put_nowait(item):**相当于put(item,block=False)
task_done() 和 join(): 每当向队列中put()一个item时,未完成任务的计数unfinished_tasks就会加1。调用一次task_done(),unfinished_tasks会减1:
当unfinished_tasks=0时,join()相当于pass。
当unfinished_tasks>0时,join()会阻塞线程。
当调用task_done()次数多于put()时,会引发异常ValueError: task_done() called too many times
2、queue模块的三种模式
(1)FIFO队列
先进先出
import queueq = queue.Queue(3) # FIFO模式
# Queue(maxsize)的参数maxsize为队列最大长度,缺省,则队列长度无限长。q.put(12)
q.put("hello")
q.put({"age":18})while True:data = q.get()print(data)print("---------")'''
12
---------
hello
---------
{'age': 18}
---------
'''
(2)LIFO队列
后进先出,类似于栈
import queueq = queue.LifoQueue() # LIFO模式q.put(12)
q.put("hello")
q.put({"age":18})while True:data = q.get()print(data)print("---------")'''
{'age': 18}
---------
hello
---------
12
---------
'''
(3)按优先级
设置优先级数字越小,越先出来
import queueq = queue.PriorityQueue() # 按优先级模式q.put([3,12])
q.put([2,"hello"])
q.put([4, {"age": 18}])while True:data = q.get()print(data)print("---------")'''
[2, 'hello']
---------
[3, 12]
---------
[4, {'age': 18}]
---------
'''
七、生产者消费者模型
生产数据的线程就是生产者,消费数据的线程就是消费者。为了解决生产者生产速度和消费者消费数据速度不均衡的问题,需要引入一个阻塞队列作为缓冲区。
import threading, queue
import timedef consumer(q):while True:item = q.get()print(f"Consume {item}\n",end="")time.sleep(1)q.task_done()def producer(q):for i in range(10):q.put(i)print("in production...\n", end="")q.join()print("finish!\n",end="")q = queue.Queue()t1 = threading.Thread(target=consumer, args=(q,),daemon=True)
t2 = threading.Thread(target=consumer, args=(q,),daemon=True)t1.start()
t2.start()producer(q)
print("end")