摘要:
1.死锁与递归锁
2.信号量
3.Event
4.Timer
5.GIL
6.Queue
7.什么时候该用多线程和多进程
一. 死锁与递归锁
所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程.
死锁例子:
死锁-------------------
from threading import Thread,Lock,RLock
import time
mutexA = Lock()
mutexB = Lock()
class MyThread(Thread):
def run(self):
self.f1()
self.f2()
def f1(self):
mutexA.acquire()
print('\033[33m%s 拿到A锁 '%self.name)
mutexB.acquire()
print('\033[45%s 拿到B锁 '%self.name)
mutexB.release()
mutexA.release()
def f2(self):
mutexB.acquire()
print('\033[33%s 拿到B锁 ' % self.name)
time.sleep(1) #睡一秒就是为了保证A锁已经被别人那到了
mutexA.acquire()
print('\033[45m%s 拿到B锁 ' % self.name)
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t = MyThread()
t.start() #一开启就会去调用run方法
解决死锁的方法就是使用递归锁:在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次acquire,每次acquire,counter都加1,直到一个线程所有的acquire都被release(counter计算器为0)，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁.
mutexA=mutexB=threading.RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止
from threading import Thread,Lock,RLock
import time
mutexB = mutexA = RLock()
class MyThread(Thread):
def run(self):
self.f1()
self.f2()
def f1(self):
mutexA.acquire()
print('\033[33m%s 拿到A锁 '%self.name)
mutexB.acquire()
print('\033[45%s 拿到B锁 '%self.name)
mutexB.release()
mutexA.release()
def f2(self):
mutexB.acquire()
print('\033[33%s 拿到B锁 ' % self.name)
time.sleep(1) #睡一秒就是为了保证A锁已经被别人拿到了
mutexA.acquire()
print('\033[45m%s 拿到B锁 ' % self.name)
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t = MyThread()
t.start() #一开启就会去调用run方法
递归锁
总结死锁和递归锁的区别:
1.死锁只能acquire一次.递归锁可以acquire多次
二. 信号量
Semaphore管理一个内置的计数器.
Semaphore与进程池看起来类似，但是是完全不同的概念.
进程池：Pool(4)，最大只能产生四个进程，而且从头到尾都只是这四个进程，不会产生新的.
信号量：信号量是产生的一堆进程/线程，即产生了多个任务都去抢那一把锁.
信号量其实也是一把锁,只是这个锁,可以有多个.
解析:
Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。
from threading import Thread,Semaphore,currentThread
import time
import random sm = Semaphore(3) # 生成信号量的大小 def task():
sm.acquire() #加锁
print("%s 正在上厕所"%currentThread().getName())
time.sleep(random.randint(3,5))
sm.release()
print("\033[1;34m%s 从厕所出来了\033[0m"%currentThread().getName()) if __name__ == '__main__':
for i in range(10):
t = Thread(target=task)
t.start()
-----------------------------------------------------------------------------------------
"""
信号量的大小,就是每次执行任务的数量.每次同时只能执行3个,要等到前一个或者多个释放了锁以后,其他任务才能抢锁继续运行 """
信号量例子
三. Event
线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行
Event常用方法属性:
from threading import Event
Event.is_Set() #返回event的状态值
Event.wait() #如果 event.isSet()==False将阻塞线程；
Event.set() #设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
Event.clear() #恢复
例:有多个工作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作.
from threading import Thread,Event,currentThread
import time
event = Event() def conn():
count = 0 while not event.is_set():
if count >=3:
return
print("%s tyr connectioning"%currentThread().getName())
event.wait(2)
count +=1
print("%s connected"%currentThread().getName()) def checking():
print("%s checking sql"%currentThread().getName())
time.sleep(5)
event.set() if __name__ == '__main__':
for i in range(3):
t = Thread(target=conn)
t.start() c = Thread(target=checking)
c.start()
-----------------------------------------------------------------------------------------
"""
Thread-1 tyr connectioning
Thread-2 tyr connectioning
Thread-3 tyr connectioning
Thread-4 checking sql
Thread-3 tyr connectioning
Thread-2 tyr connectioning
Thread-1 tyr connectioning
一直在尝试连接到服务.
Thread-2 tyr connectioning
Thread-1 tyr connectioning
Thread-3 tyr connectioning
等到checking发送set信号后,才正式连接上
Thread-2 connected
Thread-1 connected
Thread-3 connected """
多线程连接MySQL
四. Timer定时器
Timer定时器:经过指定的时间后,运行指定的程序.
输入随机的验证码,5秒后,自动刷新验证码.
from threading import Thread,Timer
import random class Mycode: def __init__(self):
self.cache_code() def make_code(self):
res=""
for i in range(4):
str1=str(random.randint(0,9))
str2=chr(random.randint(65,90))
res+=random.choice([str1,str2]) # 每次从列表列表中随机取出一个字符串,拼接到res上
return res def cache_code(self,n=5):
self.code=self.make_code() # 获取验证码
print(self.code)
t=Timer(n,self.cache_code) # 5秒之后,调用自己
t.start() def check_code(self):
while True:
inp_str=input(">>>>").strip()
if inp_str.upper()==self.code:
print("输入正确")
else:
print("输入错误") if __name__=='__main__':
c=Mycode()
t=Thread(target=c.check_code)
t.start()
输入随机验证码
五. GIL
GIL本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。可以肯定的一点是：保护不同的数据的安全，就应该加不同的锁。要想了解GIL，首先确定一点：每次执行python程序，都会产生一个独立的进程。例如python test.py，python aaa.py，python bbb.py会产生3个不同的python进程
在python解释器中,不仅 有我们启动的主线程任务或者由此主线程开启的其他线程任务,还有解释器开启的垃圾回收等解释器级别的线程，总之，所有线程都运行在这一个进程内，毫无疑问.所有的数据都是共享的,所有线程的任务，都需要将任务的代码当做参数传给解释器的代码去执行，即所有的线程要想运行自己的任务，首先需要解决的是能够访问到解释器的代码。
那么流程就是这样:
多个线程先访问到解释器的代码，即拿到执行权限，然后将target的代码交给解释器的代码去执行
解释器的代码是所有线程共享的，所以垃圾回收线程也可能访问到解释器的代码而去执行，这就导致了一个问题:对于同一个数据100，可能线程1执行x=100的同时，而垃圾回收执行的是回收100的操作，解决这种问题没有什么高明的方法，就是加锁处理，如下图的GIL，保证python解释器同一时间只能执行一个任务的代码
五-1.GIL与Lock
Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要lock?首先，我们需要达成共识：锁的目的是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据,然后，我们可以得出结论：保护不同的数据就应该加不同的锁。最后，问题就很明朗了，GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock，如下图
分析:
1、100个线程去抢GIL锁，即抢执行权限
2、肯定有一个线程先抢到GIL（暂且称为线程1），然后开始执行，一旦执行就会拿到lock.acquire()
3、极有可能线程1还未运行完毕，就有另外一个线程2抢到GIL，然后开始运行，但线程2发现互斥锁lock还未被线程1释放，于是阻塞，*交出执行权限，即释放GIL
4、直到线程1重新抢到GIL，开始从上次暂停的位置继续执行，直到正常释放互斥锁lock，然后其他的线程再重复2 3 4的过程
例子:
from threading import Thread,Lock
import time def work(mutexA):
global n
mutexA.acquire()
temp =n
time.sleep(0.1)
n = temp-1
mutexA.release() if __name__ == '__main__':
n=100
mutexA=Lock()
li =[]
for i in range(100):
t = Thread(target=work,args=(mutexA,))
li.append(t)
t.start() for i in li:
i.join()
print("主线程",n)
----------------------------------------------------------------------------------------
主线程 0
如果去掉mutexA锁,结果肯定是99,因为在无锁的情况下,所有的线程都已经运行到time.sleep()这里了.这时候所有线程的n都是100.
六. queue
queue队列 ：使用import queue，用法与进程Queue一样
queue.Queue(maxsize=0) #先进先出
# 1.队列-----------
import queue
q = queue.Queue(3) #先进先出
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
先进先出
queue.LifoQueue(maxsize=0) #先进后出
# 2.堆栈----------
q = queue.LifoQueue() #先进后出（或者后进先出）
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
-------------------------------------------------------------------------------------------
third
second
first
先进后出
queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列
# ----------------
'''3.put进入一个元组，元组的第一个元素是优先级
（通常也可以是数字，或者也可以是非数字之间的比较）
数字越小，优先级越高'''
q = queue.PriorityQueue()
q.put((20,'a'))
q.put((10,'b')) #先出来的是b,数字越小优先级越高嘛
q.put((30,'c'))
print(q.get())
print(q.get())
print(q.get())
-------------------------------------------------------------------------------------------
b
a
c
优先级
七. 在什么时候使用多线程和多进程
1、cpu到底是用来做计算的，还是用来做I/O的？
2、多cpu，意味着可以有多个核并行完成计算，所以多核提升的是计算性能
3、每个cpu一旦遇到I/O阻塞，仍然需要等待，所以多核对I/O操作没什么用处一个工人相当于cpu，此时计算相当于工人在干活，I/O阻塞相当于为工人干活提供所需原材料的过程，工人干活的过程中如果没有原材料了，则工人干活的过程需要停止，直到等待原材料的到来。如果你的工厂干的大多数任务都要有准备原材料的过程（I/O密集型），那么你有再多的工人，意义也不大，还不如一个人，在等材料的过程中让工人去干别的活，反过来讲，如果你的工厂原材料都齐全，那当然是工人越多，效率越高.
结论:
1、对计算来说，cpu越多越好，但是对于I/O来说，再多的cpu也没用
2、当然对运行一个程序来说，随着cpu的增多执行效率肯定会有所提高（不管提高幅度多大，总会有所提高），这是因为一个程序基本上不会是纯计算或者纯I/O，所以我们只能相对的去看一个程序到底是计算密集型还是I/O密集型，从而进一步分析python的多线程到底有无用武之地假设我们有四个任务需要处理，处理方式肯定是要玩出并发的效果，解决方案可以是：
方案一：开启四个进程
方案二：一个进程下，开启四个线程
--------------------------------------------------------------------------------------------
单核情况下，分析结果:
如果四个任务是计算密集型，没有多核来并行计算，方案一徒增了创建进程的开销，方案二胜
如果四个任务是I/O密集型，方案一创建进程的开销大，且进程的切换速度远不如线程，方案二胜
--------------------------------------------------------------------------------------------
多核情况下，分析结果：
如果四个任务是计算密集型，多核意味着并行计算，在python中一个进程中同一时刻只有一个线程执行用不上多核，方案一胜
如果四个任务是I/O密集型，再多的核也解决不了I/O问题，方案二胜
--------------------------------------------------------------------------------------------
结论： 现在的计算机基本上都是多核，python对于计算密集型的任务开多线程的效率并不能带来多大性能上的提升，甚至不如串行(没有大量切换)，但是，对于IO密集型的任务效率还是有显著提升的。
七-1. 多线程性能测试
1.多核也就是多个CPU
（1）cpu越多，提高的是计算的性能
（2）如果程序是IO操作的时候（多核和单核是一样的），再多的cpu也没有意义。
2.实现并发
第一种：一个进程下，开多个线程
第二种：开多个进程
3.多进程：
优点：可以利用多核
缺点：开销大
4.多线程
优点：开销小
缺点：不可以利用多核
5多进程和多进程的应用场景
1.计算密集型：也就是计算多，IO少
如果是计算密集型，就用多进程（如金融分析等）
2.IO密集型：也就是IO多，计算少
如果是IO密集型的，就用多线程（一般遇到的都是IO密集型的）
#计算型
from multiprocessing import Process
from threading import Thread
import time
import os
def work():
n= 0
for i in range(100000000):
n +=i if __name__ == '__main__':
start=time.time()
li = []
for i in range(os.cpu_count()):
# p =Process(target=work) # 多进程耗时:5.942339897155762
t = Thread(target=work) #多线程耗时:21.759244680404663
# li.append(p)
li.append(t)
# p.start()
t.start()
for i in li:
i.join() print(time.time() - start)
计算型多进程效率高
#I/O型
from multiprocessing import Process
from threading import Thread
import time
import os
def work():
time.sleep(0.1)
print("python") if __name__ == '__main__':
start=time.time()
li = []
for i in range(300):
# p =Process(target=work) # 多进程耗时:24.05037522315979
t = Thread(target=work) #多线程耗时: 0.12700724601745605
# li.append(p)
li.append(t)
# p.start()
t.start()
for i in li:
i.join() print(time.time() - start)
I/O操作多线程效率高
多线程用于IO密集型，如socket，爬虫，web
多进程用于计算密集型，如金融分析