Python并发编程 05 锁、同步条件、信号量、线程队列、生产者消费者模型

一、基础概念

①并发： 系统在一段时间内，处理多个任务的能力
②并行： 系统在同一时刻，处理多个任务的能力
③同步： 当进程执行到一个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")