2023年8月2日发(作者：)

python多线程并发100_python并发编程之多线程（2）⼀ threading模块介绍multiprocess模块的完全模仿了threading模块的接⼝，⼆者在使⽤层⾯，有很⼤的相似性，因⽽不再详细介绍thread模块python不推荐使⽤这个模块，推荐更⾼级的threading。thread模块和对象函数 描述start_new_thread(function,args) 产⽣新线程，指定函数和参数allocate_lock() 分配⼀个LockType类型的锁对象exit() 线程退出LockType对象的⽅法acquire() 尝试获取对象对象locked 如果获取了返回True否则返回Falserelease() 释放锁#coding:utf-8importthreadfrom time importsleep,ctimeloops=[4,2]#睡眠时间defloop(nloop,nsec,lock):print 'start loop',nloop,'at:',ctime()sleep(nsec)print 'loop',nloop,'done at:',ctime()e()#释放锁对象defmain():print 'starting at :',ctime()locks=[]nloops=range(len(loops))#创建nloops数组#创建锁并加⼊组for i innloops:lock=te_lock()#创建 lock (不能直接加⼊)e()#尝试获取(lock)#加⼊组for i innloops:_new_thread(loop,(i,loops[i],locks[i]))#创建线程#获取锁对象 成功True 失败Falsefor i innloops:while locks[i].locked():pass#如果锁对象被释放 表⽰解锁了 可以继续执⾏print 'all DONE at:',ctime()if __name__ == '__main__':main()thread模块threading模块中的Thread类有很多thread模块⾥没有的⽅法，⼀般使⽤时可以选择⼏种⽅法⾥的⼀种：创建⼀个Thread实例，传给它⼀个函数；创建⼀个Thread实例，传给它⼀个可调⽤的类对象；从Thread派⽣⼀个⼦类，创建这个⼦类的实例。可以看看它有哪些⽅法函数 描述start() 开始线程的执⾏run() 定义线程的功能的函数(⼀般会被⼦类重写)join(timeout=None) 程序挂起，知道线程结束，如果给了timeout，最多阻塞timeout秒getName() 返回线程的名字setName(name) 设置线程的名字isAlive() 布尔标志，表⽰这个线程是否还在运⾏中isDaemon() 返回线程的daemon标志setDaemon(daemonic) 把线程的daemon标志设置成daemonic⽤threading模块重写我们上次的例⼦：importthreadingfrom time importsleep, ctimeloops= [4, 2]defloop(nloop, nsec):print 'start loop%s at: %sn' %(nloop, ctime()),sleep(nsec)print 'loop%s done at: %sn' %(nloop, ctime()),defmain():print 'starting at: %sn' %ctime(),threads=[]nloops=range(len(loops))for i innloops:t= (target =loop,args=(i,loops[i]))(t)for i innloops:threads[i].start()for i innloops:threads[i].join()print 'all DONE at: %sn' %ctime(),if __name__ == '__main__':main()threading模块⼆ 开启线程的两种⽅式#⽅式⼀from threading importThreadimporttimedefsayhi(name):(2)print('%s say hello' %name)if __name__ == '__main__':t=Thread(target=sayhi,args=('egon',))()print('主线程')⽅式⼀⽅式⼀#⽅式⼆from threading importThreadimporttimeclassSayhi(Thread):def __init__(self,name):super().__init__()=namedefrun(self):(2)print('%s say hello' %)if __name__ == '__main__':t= Sayhi('egon')()print('主线程')⽅式⼆⽅式⼆三 在⼀个进程下开启多个线程与在⼀个进程下开启多个⼦进程的区别from threading importThreadfrom multiprocessing importProcessimportosdefwork():print('hello')if __name__ == '__main__':#在主进程下开启线程t=Thread(target=work)()print('主线程/主进程')'''打印结果:hello主线程/主进程'''#在主进程下开启⼦进程t=Process(target=work)()print('主线程/主进程')'''打印结果:主线程/主进程hello'''谁的开启速度快谁的开启速度快from threading importThreadfrom multiprocessing importProcessimportosdefwork():print('hello',())if __name__ =='__main__':#part1:在主进程下开启多个线程,每个线程都跟主进程的pid⼀样t1=Thread(target=work)t2=Thread(target=work)()()print('主线程/主进程pid',())#part2:开多个进程,每个进程都有不同的pidp1=Process(target=work)p2=Process(target=work)()()print('主线程/主进程pid',())瞅⼀瞅pidpidfrom threading importThreadfrom multiprocessing importProcessimportosdefwork():globalnn=0if __name__ == '__main__':#n=100#p=Process(target=work)#()#()#print('主',n) #毫⽆疑问⼦进程p已经将⾃⼰的全局的n改成了0,但改的仅仅是它⾃⼰的,查看⽗进程的n仍然为100n=1t=Thread(target=work)()()print('主',n) #查看结果为0,因为同⼀进程内的线程之间共享进程内的数据同⼀进程内的线程共享该进程的数据？同⼀进程内的线程共享该进程的数据？四 练习练习⼀：#_*_coding:utf-8_*_#!/usr/bin/env pythonimportmultiprocessingimportthreadingimportsockets=(_INET,_STREAM)(('127.0.0.1',8080))(5)defaction(conn):whileTrue:data=(1024)print(data)(())if __name__ == '__main__':whileTrue:conn,addr=()p=(target=action,args=(conn,))()多线程并发的socket服务端多线程并发的socket服务端#_*_coding:utf-8_*_#!/usr/bin/env pythonimportsockets=(_INET,_STREAM)t(('127.0.0.1',8080))whileTrue:msg=input('>>:').strip()if not msg:(('utf-8'))data=(1024)print(data)客户端客户端练习⼆：三个任务，⼀个接收⽤户输⼊，⼀个将⽤户输⼊的内容格式化成⼤写，⼀个将格式化后的结果存⼊⽂件from threading importThreadmsg_l=[]format_l=[]deftalk():whileTrue:msg=input('>>:').strip()if not msg:continuemsg_(msg)defformat_msg():whileTrue:ifmsg_l:res=msg_()format_(())defsave():whileTrue:ifformat_l:with open('','a',encoding='utf-8') as f:res=format_()('%sn' %res)if __name__ == '__main__':t1=Thread(target=talk)t2=Thread(target=format_msg)t3=Thread(target=save)()()()View Code五 线程相关的其他⽅法Thread实例对象的⽅法# isAlive(): 返回线程是否活动的。# getName(): 返回线程名。# setName(): 设置线程名。threading模块提供的⼀些⽅法：# tThread(): 返回当前的线程变量。# ate(): 返回⼀个包含正在运⾏的线程的list。正在运⾏指线程启动后、结束前，不包括启动前和终⽌后的线程。# Count(): 返回正在运⾏的线程数量，与len(ate())有相同的结果。from threading importThreadimportthreadingfrom multiprocessing importProcessimportosdefwork():(3)print(t_thread().getName())if __name__ == '__main__':#在主进程下开启线程t=Thread(target=work)()print(t_thread().getName())print(t_thread()) #主线程print(ate()) #连同主线程在内有两个运⾏的线程print(_count())print('主线程/主进程')'''打印结果:MainThread[<_mainthread started>, ]主线程/主进程Thread-1'''View Code主线程等待⼦线程结束from threading import Threadimport timedef sayhi(name):(2)print('%s say hello' %name)if __name__ == '__main__':t=Thread(target=sayhi,args=('egon',))()()print('主线程')print(_alive())'''egon say hello主线程False'''六 守护线程⽆论是进程还是线程，都遵循：守护xxx会等待主xxx运⾏完毕后被销毁需要强调的是：运⾏完毕并⾮终⽌运⾏#1.对主进程来说，运⾏完毕指的是主进程代码运⾏完毕#2.对主线程来说，运⾏完毕指的是主线程所在的进程内所有⾮守护线程统统运⾏完毕，主线程才算运⾏完毕详细解释：#1 主进程在其代码结束后就已经算运⾏完毕了(守护进程在此时就被回收),然后主进程会⼀直等⾮守护的⼦进程都运⾏完毕后回收⼦进程的资源(否则会产⽣僵⼫进程)，才会结束，#2 主线程在其他⾮守护线程运⾏完毕后才算运⾏完毕(守护线程在此时就被回收)。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，⽽进程必须保证⾮守护线程都运⾏完毕后才能结束。from threading import Threadimport timedef sayhi(name):(2)print('%s say hello' %name)if __name__ == '__main__':t=Thread(target=sayhi,args=('egon',))mon(True) #必须在()之前设置()print('主线程')print(_alive())'''主线程True'''from threading importThreadimporttimedeffoo():print(123)(1)print("end123")defbar():print(456)(3)print("end456")t1=Thread(target=foo)t2=Thread(target=bar)=()()print("main-------")迷惑⼈的例⼦迷惑⼈的例⼦七 Python GIL(Global Interpreter Lock)⼋ 同步锁三个需要注意的点：#1.线程抢的是GIL锁，GIL锁相当于执⾏权限，拿到执⾏权限后才能拿到互斥锁Lock，其他线程也可以抢到GIL，但如果发现Lock仍然没有被释放则阻塞，即便是拿到执⾏权限GIL也要⽴刻交出来#是等待所有，即整体串⾏，⽽锁只是锁住修改共享数据的部分，即部分串⾏，要想保证数据安全的根本原理在于让并发变成串⾏，join与互斥锁都可以实现，毫⽆疑问，互斥锁的部分串⾏效率要更⾼#3. ⼀定要看本⼩节最后的GIL与互斥锁的经典分析GIL VS Lock机智的同学可能会问到这个问题，就是既然你之前说过了，Python已经有⼀个GIL来保证同⼀时间只能有⼀个线程来执⾏了，为什么这⾥还需要lock?⾸先我们需要达成共识：锁的⽬的是为了保护共享的数据，同⼀时间只能有⼀个线程来修改共享的数据然后，我们可以得出结论：保护不同的数据就应该加不同的锁。最后，问题就很明朗了，GIL 与Lock是两把锁，保护的数据不⼀样，前者是解释器级别的(当然保护的就是解释器级别的数据，⽐如垃圾回收的数据)，后者是保护⽤户⾃⼰开发的应⽤程序的数据，很明显GIL不负责这件事，只能⽤户⾃定义加锁处理，即Lock过程分析：所有线程抢的是GIL锁，或者说所有线程抢的是执⾏权限线程1抢到GIL锁，拿到执⾏权限，开始执⾏，然后加了⼀把Lock，还没有执⾏完毕，即线程1还未释放Lock，有可能线程2抢到GIL锁，开始执⾏，执⾏过程中发现Lock还没有被线程1释放，于是线程2进⼊阻塞，被夺⾛执⾏权限，有可能线程1拿到GIL，然后正常执⾏到释放Lock。。。这就导致了串⾏运⾏的效果既然是串⾏，那我们执⾏()()()这也是串⾏执⾏啊，为何还要加Lock呢，需知join是等待t1所有的代码执⾏完，相当于锁住了t1的所有代码，⽽Lock只是锁住⼀部分操作共享数据的代码。因为Python解释器帮你⾃动定期进⾏内存回收，你可以理解为python解释器⾥有⼀个独⽴的线程，每过⼀段时间它起wake up做⼀次全局轮询看看哪些内存数据是可以被清空的，此时你⾃⼰的程序⾥的线程和 py解释器⾃⼰的线程是并发运⾏的，假设你的线程删除了⼀个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能⼀个其它线程正好⼜重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了。为了解决类似的问题，python解释器简单粗暴的加了锁，即当⼀个线程运⾏时，其它⼈都不能动，这样就解决了上述的问题， 这可以说是Python早期版本的遗留问题。from threading import Threadimport os,timedef work():global ntemp=(0.1)n=temp-1if __name__ == '__main__':n=100l=[]for i in range(100):p=Thread(target=work)(p)()for p in l:()print(n) #结果可能为99锁通常被⽤来实现对共享资源的同步访问。为每⼀个共享资源创建⼀个Lock对象，当你需要访问该资源时，调⽤acquire⽅法来获取锁对象(如果其它线程已经获得了该锁，则当前线程需等待其被释放)，待资源访问完后，再调⽤release⽅法释放锁：from threading importThread,Lockimportos,timedefwork():e()temp=(0.1)n=e()if __name__ == '__main__':lock=Lock()n=100l=[]for i in range(100):p=Thread(target=work)(p)()for p inl:()print(n) #结果肯定为0，由原来的并发执⾏变成串⾏，牺牲了执⾏效率保证了数据安全分析：#1.100个线程去抢GIL锁，即抢执⾏权限#2. 肯定有⼀个线程先抢到GIL(暂且称为线程1)，然后开始执⾏，⼀旦执⾏就会拿到e()#3. 极有可能线程1还未运⾏完毕，就有另外⼀个线程2抢到GIL，然后开始运⾏，但线程2发现互斥锁lock还未被线程1释放，于是阻塞，被迫交出执⾏权限，即释放GIL#4.直到线程1重新抢到GIL，开始从上次暂停的位置继续执⾏，直到正常释放互斥锁lock，然后其他的线程再重复2 3 4的过程GIL锁与互斥锁综合分析(重点！！！)互斥锁与join的区别：#不加锁:并发执⾏,速度快,数据不安全from threading importcurrent_thread,Thread,Lockimportos,timedeftask():globalnprint('%s is running'%current_thread().getName())temp=(0.5)n=temp-1if __name__ == '__main__':n=100lock=Lock()threads=[]start_time=()for i in range(100):t=Thread(target=task)(t)()for t inthreads:()stop_time=()print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 is runningThread-2 Thread-100 is running主:0.5216 n:99'''#不加锁:未加锁部分并发执⾏,加锁部分串⾏执⾏,速度慢,数据安全from threading importcurrent_thread,Thread,Lockimportos,timedeftask():#未加锁的代码并发运⾏(3)print('%s start to run' %current_thread().getName())globaln#加锁的代码串⾏运⾏e()temp=(0.5)n=e()if __name__ == '__main__':n=100lock=Lock()threads=[]start_time=()for i in range(100):t=Thread(target=task)(t)()for t inthreads:()stop_time=()print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 is runningThread-2 Thread-100 is running主:53.2942 n:0'''#有的同学可能有疑问:既然加锁会让运⾏变成串⾏,那么我在start之后⽴即使⽤join,就不⽤加锁了啊,也是串⾏的效果啊#没错:在start之后⽴刻使⽤jion,肯定会将100个任务的执⾏变成串⾏,毫⽆疑问,最终n的结果也肯定是0,是安全的,但问题是#start后⽴即join:任务内的所有代码都是串⾏执⾏的,⽽加锁,只是加锁的部分即修改共享数据的部分是串⾏的#单从保证数据安全⽅⾯,⼆者都可以实现,但很明显是加锁的效率更⾼.from threading importcurrent_thread,Thread,Lockimportos,timedeftask():(3)print('%s start to run' %current_thread().getName())globalntemp=(0.5)n=temp-1if __name__ == '__main__':n=100lock=Lock()start_time=()for i in range(100):t=Thread(target=task)()()stop_time=()print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 start to runThread-2 start Thread-100 start to run主:350.6937336921692 n:0 #耗时是多么的恐怖'''互斥锁与join的区别(重点！！！)import threadingR=()e()'''对公共数据的操作'''e()九 死锁现象与递归锁进程也有死锁与递归锁。所谓死锁： 是指两个或两个以上的进程或线程在执⾏过程中，因争夺资源⽽造成的⼀种互相等待的现象，若⽆外⼒作⽤，它们都将⽆法推进下去。此时称系统处于死锁状态或系统产⽣了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁from threading importThread,LockimporttimemutexA=Lock()mutexB=Lock()classMyThread(Thread):defrun(self):1()2()deffunc1(self):e()print('033[41m%s 拿到A锁033[0m' %)e()print('033[42m%s 拿到B锁033[0m' %)e()e()deffunc2(self):e()print('033[43m%s 拿到B锁033[0m' %)(2)e()print('033[44m%s 拿到A锁033[0m' %)e()e()if __name__ == '__main__':for i in range(10):t=MyThread()()'''Thread-1 拿到A锁Thread-1 拿到B锁Thread-1 拿到B锁Thread-2 拿到A锁然后就卡住，死锁了'''死锁解决⽅法，递归锁，在Python中为了⽀持在同⼀线程中多次请求同⼀资源，python提供了可重⼊锁RLock。这个RLock内部维护着⼀个Lock和⼀个counter变量，counter记录了acquire的次数，从⽽使得资源可以被多次require。直到⼀个线程所有的acquire都被release，其他的线程才能获得资源。上⾯的例⼦如果使⽤RLock代替Lock，则不会发⽣死锁：mutexA=mutexB=()#⼀个线程拿到锁，counter加1,该线程内⼜碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为⽌⼗ 信号量Semaphore同进程的⼀样Semaphore管理⼀个内置的计数器，每当调⽤acquire()时内置计数器-1；调⽤release() 时内置计数器+1；计数器不能⼩于0；当计数器为0时，acquire()将阻塞线程直到其他线程调⽤release()。实例：(同时只有5个线程可以获得semaphore,即可以限制最⼤连接数为5)：from threading importThread,Semaphoreimportthreadingimporttime#def func():#if e():#print(tThread().getName() + ' get semaphore')#(2)#e()deffunc():e()print('%s get sm' %t_thread().getName())(3)e()if __name__ == '__main__':sm=Semaphore(5)for i in range(23):t=Thread(target=func)()View Code与进程池是完全不同的概念，进程池Pool(4)，最⼤只能产⽣4个进程，⽽且从头到尾都只是这四个进程，不会产⽣新的，⽽信号量是产⽣⼀堆线程/进程⼗⼀ Event同进程的⼀样线程的⼀个关键特性是每个线程都是独⽴运⾏且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定⾃⼰下⼀步的操作,这时线程同步问题就会变得⾮常棘⼿。为了解决这些问题,我们需要使⽤threading库中的Event对象。 对象包含⼀个可由线程设置的信号标志,它允许线程等待某些事件的发⽣。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待⼀个Event对象, ⽽这个Event对象的标志为假,那么这个线程将会被⼀直阻塞直⾄该标志为真。⼀个线程如果将⼀个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果⼀个线程等待⼀个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执⾏()：返回event的状态值；()：如果 ()==False将阻塞线程；()： 设置event的状态值为True，所有阻塞池的线程激活进⼊就绪状态， 等待操作系统调度；()：恢复event的状态值为False。例如，有多个⼯作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些⼯作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采⽤机制来协调各个⼯作线程的连接操作⼗⼆ 条件Condition(了解)使得线程等待，只有满⾜某条件时，才释放n个线程import threadingdef run(n):e()()print("run the thread: %s" %n)e()if __name__ == '__main__':con = ion()for i in range(10):t = (target=run, args=(i,))()while True:inp = input('>>>')if inp == 'q':e()(int(inp))e()defcondition_func():ret=Falseinp= input('>>>')if inp == '1':ret=Truereturnretdefrun(n):e()_for(condition_func)print("run the thread: %s" %n)e()if __name__ == '__main__':con=ion()for i in range(10):t= (target=run, args=(i,))()View Code⼗三 定时器定时器，指定n秒后执⾏某操作from threading import Timerdef hello():print("hello, world")t = Timer(1, hello)() # after 1 seconds, "hello, world" will be printed⼗四 线程queuequeue队列 ：使⽤import queue，⽤法与进程Queue⼀样queue is especially useful in threaded programming when information must be exchanged safely between multiple (maxsize=0) #先进先出importqueueq=()('first')('second')('third')print(())print(())print(())'''结果(先进先出):firstsecondthird''' 先进先出eue(maxsize=0) #last in fisrt outimportqueueq=eue()('first')('second')('third')print(())print(())print(())'''结果(后进先出):thirdsecondfirst'''eue 后进先出tyQueue(maxsize=0) #存储数据时可设置优先级的队列importqueueq=tyQueue()#put进⼊⼀个元组,元组的第⼀个元素是优先级(通常是数字,也可以是⾮数字之间的⽐较),数字越⼩优先级越⾼((20,'a'))((10,'b'))((30,'c'))print(())print(())print(())'''结果(数字越⼩优先级越⾼,优先级⾼的优先出队):(10, 'b')(20, 'a')(30, 'c')'''tyQueue put进⼊⼀个元组,元组的第⼀个元素是优先级(通常是数字,也可以是⾮数字之间的⽐较),数字越⼩优先级越⾼其他⼗五 Python标准模块--s#1 介绍s模块提供了⾼度封装的异步调⽤接⼝ThreadPoolExecutor：线程池，提供异步调⽤ProcessPoolExecutor: 进程池，提供异步调⽤Both implement the same interface, which is defined by the abstract Executor class.#2 基本⽅法#submit(fn, *args, **kwargs)异步提交任务#map(func, *iterables, timeout=None, chunksize=1)取代for循环submit的操作#shutdown(wait=True)相当于进程池的()+()操作wait=True，等待池内所有任务执⾏完毕回收完资源后才继续wait=False，⽴即返回，并不会等待池内的任务执⾏完毕但不管wait参数为何值，整个程序都会等到所有任务执⾏完毕submit和map必须在shutdown之前#result(timeout=None)取得结果#add_done_callback(fn)回调函数 ProcessPoolExecutor ThreadPoolExecutor map的⽤法 回调函数#介绍The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls sPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but alsomeans that only picklable objects can be executed sPoolExecutor(max_workers=None, mp_context=None)An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workersisNone or not given, it will default to the number of processors on the machine. If max_workers is lower orequal to 0, then aValueError will be raised.#⽤法from s importThreadPoolExecutor,ProcessPoolExecutorimportos,time,randomdeftask(n):print('%s isruning' %())(t(1,3))return n**2if __name__ == '__main__':executor=ProcessPoolExecutor(max_workers=3)futures=[]for i in range(11):future=(task,i)(future)wn(True)print('+++>')for future infutures:print(())ProcessPoolExecutorProcessPoolExecutor#介绍ThreadPoolExecutor isan Executor subclass that uses a pool of threads to execute calls PoolExecutor(max_workers=None, thread_name_prefix='')An Executor subclass that uses a pool of at most max_workers threads to execute calls din version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine,multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number ofworkers should be higher than the number of workers version 3.6: The thread_name_prefix argument was added to allow users to control the names forworker threads created by the pool foreasier debugging.#⽤法与ProcessPoolExecutor相同ThreadPoolExecutorThreadPoolExecutorfrom s importThreadPoolExecutor,ProcessPoolExecutorimportos,time,randomdeftask(n):print('%s isruning' %())(t(1,3))return n**2if __name__ == '__main__':executor=ThreadPoolExecutor(max_workers=3)#for i in range(11):#future=(task,i)(task,range(1,12)) #map取代了for+submitmap的⽤法map的⽤法from s importThreadPoolExecutor,ProcessPoolExecutorfrom multiprocessingimportPoolimportrequestsimportjsonimportosdefget_page(url):print(' get %s' %((),url))respone=(url)if _code == 200:return {'url':url,'text':}defparse_page(res):res=()print(' parse %s' %((),res['url']))parse_res='url: size:[%s]n' %(res['url'],len(res['text']))with open('','a') as f:(parse_res)if __name__ == '__main__':urls=['','','','/','/']##for url in urls:#_async(get_page,args=(url,),callback=pasrse_page)#()#()p=ProcessPoolExecutor(3)for url inurls:(get_page,url).add_done_callback(parse_page)#parse_page拿到的是⼀个future对象obj，需要⽤()拿到结果回调函数回调函数