python并发编程之多线程编程
一、threading模块介绍
multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性，因而不再详细介绍
二、开启线程的两种方式
方式一：
from threading import Thread
import time
import random
def task(name):
print('%s is running' %name)
time.sleep(random.randint(1,3))
print('%s is end' %name)
#注意：在windows中Process()必须放到# if __name__ == '__main__':下
if __name__ == '__main__':
t1=Thread(target=task,args=('ALex',))
t1.start()
print('主函数')
#linux下的就不需要if __name__ == '__main__':
t1=Thread(target=task,args=('renwu1',))
t1.start()
print('主函数')
方式二：
from threading import Thread
import time
import random
class task(Thread):
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
print('%s is runnging' %self.name)
time.sleep(random.randint(1,3))
print('%s is end' %self.name)
t1=task('renwu1')
t1.start()
print('主函数')
三、在一个进程下开启多个线程与在一个进程下开启多个子进程的区别
1、比较谁的开启速度快
from threading import Thread
from multiprocessing import Process
def task():
print('hello')
t1=Thread(target=task)
t1.start()
print('主函数/线程')
p1=Process(target=task)
p1.start()
print('主函数/进程')
2、查看pid
from threading import Thread
from multiprocessing import Process
import os
def task():
print('hello', os.getpid())
# 在主进程下开多个线程,每个线程都跟主进程的pid一样
t1=Thread(target=task)
t2=Thread(target=task)
t1.start()
t2.start()
print('主函数pid/线程',os.getpid())
# 在主进程下开多个进程,每个进程都有不同的pid
p1=Process(target=task)
p2=Process(target=task)
p1.start()
p2.start()
print('主函数pid/进程',os.getpid())
3、同一进程内的线程共享改进程内的数据
from threading import Thread
from multiprocessing import Process
def task():
global n
n=0
n=1
p=Process(target=task)
p.start()
p.join()
print('主函数/进程',n)
t=Thread(target=task)
t.start()
t.join()
四、线程相关的其他方法
Thread实例对象的方法
isAlive(): 返回线程是否活动的。
getName(): 返回线程名。
setName(): 设置线程名。
threading模块提供的一些方法：
threading.currentThread(): 返回当前的线程变量。
threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
举个栗子：
from threading import Thread
import threading
import time
def work():
time.sleep(3)
print(threading.current_thread().getName())
t=Thread(target=work)
t.start()
print(threading.current_thread().getName())
print(threading.current_thread())
print(threading.enumerate())
print(threading.active_count())
print('主线程/主进程')
主线程等待子线程结束：
from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
t=Thread(target=sayhi,args=('eGon',))
t.start()
t.join() #试试注释这一行内容
print('主线程')
print(t.is_alive())
五、守护线程
无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁
需要强调的是：运行完毕并非终止运行
1.对主进程来说，运行完毕指的是主进程代码运行完毕
2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕
详细解释：
1、主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，
2、主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束
from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
t=Thread(target=sayhi,args=('egon',))
t.setDaemon(True) #必须在t.start()之前设置
t.start()
print('主线程')
print(t.is_alive())
六、python GIL(Global Interpreter Lock) 链接：https://blog.51cto.com/blogger/publish
七、同步锁
1、有三个需要注意的点：
1.线程抢的是GIL锁，GIL锁相当于执行权限，拿到执行权限后才能拿到互斥锁Lock，其他线程也可以抢到GIL，但如果发现Lock仍然没有被释放则阻塞，即便是拿到执行权限GIL也要立刻交出来
2.join是等待所有，即整体串行，而锁只是锁住修改共享数据的部分，即部分串行，要想保证数据安全的根本原理在于让并发变成串行，join与互斥锁都可以实现，毫无疑问，互斥锁的部分串行效率要更高
3. 一定要看本小节最后的GIL与互斥锁的经典分析
2、GIL VS Lock
有的同学可能会问到这个问题，就是既然你之前说过了，Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要lock?
首先我们需要达成共识：锁的目的是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据
然后，我们可以得出结论：保护不同的数据就应该加不同的锁。
最后，问题就很明朗了，GIL与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，只能用户自定义加锁处理，即Lock
过程分析：所有线程抢的是GIL锁，或者说所有线程抢的是执行权限
线程1抢到GIL锁，拿到执行权限，开始执行，然后加了一把Lock，还没有执行完毕，即线程1还未释放Lock，有可能线程2抢到GIL锁，开始执行，执行过程中发现Lock还没有被线程1释放，于是线程2进入阻塞，被夺走执行权限，有可能线程1拿到GIL，然后正常执行到释放Lock。这就导致了串行运行的效果
既然是串行，那我们执行
t1.start()
t1.join
t2.start()
t2.join()
这也是串行执行啊，为何还要加Lock呢，需知join是等待t1所有的代码执行完，相当于锁住了t1的所有代码，而Lock只是锁住一部分操作共享数据的代码。
PS:
因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的。
如果此时你自己的程序里的线程和py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了。
为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，这样就解决了上述的问题，这可以说是Python早期版本的遗留问题。
锁通常被用来实现对共享资源的同步访问。
为每一个共享资源创建一个Lock对象，当你需要访问该资源时，调用acquire方法来获取锁对象（如果其它线程已经获得了该锁，则当前线程需等待其被释放），待资源访问完后，再调用release方法释放锁：
import threading
R=threading.Lock()
R.acquire()
#对公共数据的操作
R.release()
3、GIL锁与互斥锁的综合分析
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
mutexA=Lock()
mutexB=Lock()
class MyThread(Thread):
def run(self):
self.f1()
self.f2()
def f1(self):
mutexA.acquire()
print('%s拿到了A锁' %self.name)
mutexB.acquire()
print('%s拿到了B锁' % self.name)
mutexB.release()
mutexA.release()
def f2(self):
mutexB.acquire()
print('%s拿到了B锁' % self.name)
time.sleep(0.1)
mutexA.acquire()
print('%s拿到了A锁' % self.name)
mutexA.release()
mutexB.release()
for i in range(10):
t=MyThread()
t.start()
解决死锁的方法：递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。
上面的例子如果使用RLock代替Lock，则不会发生死锁：
mutexA=mutexB=threading.RLock()
一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止
九、线程的信号量Semaphore
备注：同进程的一样
Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。
实例：(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5)：
from threading import Thread,Semaphore,current_thread
import time,random
sm=Semaphore(5)
def task():
with sm:
print('%s is laing' %current_thread().getName())
time.sleep(random.randint(1,3))
for i in range(20):
t=Thread(target=task)
t.start()
信号量与进程池是完全不同的概念；
进程池Pool(4)，指最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程。
十、Event
同进程的一样，线程的一个关键特性是每个线程都是独立运行且状态不可预测。
如果程序中的其他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。
为了解决这些问题,我们需要使用threading库中的Event对象。
对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。
在初始情况下,Event对象中的信号标志被设置为假。
如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。
一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。
如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行
event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。
十一、定时器
定时器：指定n秒后执行某操作
from threading import Timer
def hello():
print("hello, world")
t = Timer(1, hello)
t.start() # 1秒以后就会有'hello world'
应用：验证码定时器
from threading import Timer
import random,time
class Code:
def __init__(self):
self.make_cache()
def make_cache(self,interval=5):
self.cache=self.make_code()
print(self.cache)
self.t=Timer(interval,self.make_cache)
self.t.start()
def make_code(self,n=4):
res=''
for i in range(n):
s1=str(random.randint(0,9))
s2=chr(random.randint(65,90))
res+=random.choice([s1,s2])
return res
def check(self):
while True:
inp=input('>>: ').strip()
if inp.upper() == self.cache:
print('验证成功',end='\n')
self.t.cancel()
break
obj=Code()
obj.check()
十二、线程queue
queue队列 ：使用import queue，用法与进程Queue一样
class queue.Queue(maxsize=0) #先进先出
import queue
q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
class queue.LifoQueue(maxsize=0) #last in fisrt out
import queue
q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列
import queue
q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))
print(q.get())
print(q.get())
print(q.get())
结果(数字越小优先级越高,优先级高的优先出队)相关文章