今天学习了多进程中的守护进程、互斥锁、队列。
一、守护进程
主进程创建守护进程
其一:守护进程会在主进程代码执行结束后就终止
其二:守护进程内无法再开启子进程,否则抛出异常:AssertionError:daemonic proesses are not allowed to have children
注意:进程之间是互相独立的,主进程代码运行结束,守护进程随即终止
举例:
#主进程代码运行完毕,守护进程就会结束 from multiprocessing import Process from threading import Thread import time def foo(): print(123) time.sleep(1) print('end123') def bar(): print(456) time.sleep(3) print('end456') if __name__ == '__main__': p1 = Process(target = foo) p2 = Process(target = bar) p1.daemon = True#一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p随即终止运行 p1.start() p2.start() print('main---------')#打印该行则主进程代码结束,则守护进程p1应该被终止,可能会有p1任务执行的打印信息123,因为主进程打印main---------时,p1也执行了,但是随即被终止
二、进程同步(互斥锁)
进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,而共享带来的是竞争,竞争带来的结果就是错乱,如何控制,就是加锁处理。
part1:多个进程共享同一打印终端
#并发运行,效率高,但竞争同一打印终端,带来了打印错乱 from multiprocessing import Process import os,time def work(): print('%s is running'%os.getpid()) time.sleep(2) print('%s is done'%os.getpid()) if __name__ == '__main__' for i in range(3): p = Process(target=work) p.start()
#由并发变成了串行,牺牲了运行效率,但避免了竞争 from multiprocessing import Process,Lock import os,time def work(lock): lock.acquire() print('%s is running' %os.getpid()) time.sleep(2) print('%s is done'%os.getpid()) lock.release() if __name__=='__main__' lock = Lock() for i in range(3): p = Process(target=work,args=(lock,)) p.start()
part2:多个进程共享同一个文件
文件当数据库,模拟抢票
#并发运行,效率高,但竞争写同一文件,数据写入错乱 #文件db的内容为:{"count":1} #注意一定要用双引号,不然json无法识别 from multiprocessing import Process,Lock import time,json,random def search(): dic = json.load(open('db.txt')) print('33[43m剩余票数%s33[0m' %dic'['count']') def get(): dic = json.load(open('db.txt')) time.sleep(0,1) #模拟读数据的网络延迟 if dic['count'] > 0: dic['count'] -=1 time.sleep(0,2) #模拟写数据的网络延迟 json.dump(dic,open('db.txt','w')) print('33[43m购票成功33[0m') def task(lock): search() get() if __name__ == '__main__': lock = Lock() for i in range(100): #模拟并发100个客户端抢票 p = Process(target = task,args = (lock,)) p.start()
#加锁:购票行为由并发变成了串行,牺牲了运行效率,但保证了数据安全 #文件db的内容为:{"count":1} #注意一定要用双引号,不然json无法识别 from multiprocessing import Process,Lock import time,json,random def search(): dic = json.load(open('db.txt')) print('33[43m剩余票数%s33[0m' %dic['count']) def get(): dic = json.load(open('db.txt')) time.sleep(0.1) #模拟读数据的网络延迟 if dic['count'] > 0 : dic['count'] -=1 time.sleep(0.2) #模拟写数据的网络延迟 json.dump(dic,open('db.txt','w')) print('33[43m购票成功33[0m') def task(lock): search() lock.acquire() get() lock.release() if __name__=='__main__' lock=Lock() for i in range(100): #模拟并发100个客户端抢票 p = Process(target = task,args = (lock,)) p.start()
总结:
加锁可以保证多个进程修改同一块数据时,同一时间只能有一个任务可以进行修改,即串行的修改,没错,速度是慢了,但牺牲了速度缺保证了数据安全。
虽然可以用文件共享数据实现进程间通信,但问题是:
1.效率低(共享数据基于文件,而文件是硬盘上的数据)
2.需要自己加锁处理
因此我们最好找寻一种解决方案能够兼顾:1、效率高(多个进程共享一块内存的数据)2、帮我们处理好锁问题。
这就是multiprocessing 模块为我们提供的基于消息的IPC通信机制:队列和管道。
1 队列和管道都是将数据存放于内存中
2 队列又是基于(管道+锁) 实现的,可以让我们从复杂的锁问题中解脱出来,
我们应该尽量避免使用共享数据,进可能使用消息传递和队列,避免处理复杂的同步和锁问题,而且在进程数目增多时,往往可以获得更好的可获展性。
三、队列
进程彼此之间互相隔离,要实现进程间通信(IPC),multiprocessing模块支持两种形式:队列和管道,这两种方式都是使用消息传递的
创建队列的类(底层就是以管道和锁定的方式实现):
Queue([maxsize]):创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递。
参数介绍:
maxsize是队列中允许最大项数,省略则无大小限制。
方法介绍:
主要方法:
1 q.put方法用于以插入数据队列中,put方法还有两个可选参数:blocked和timeout。如果blocked为True(默认值),并且timeout为正值,该方法会阻塞timeout指定的时间,直到该队列有剩余的空间。如果超时,会抛出Queue.Full异常。如果blocked为False,但该Queue已满,会立即抛出Queue.Full异常。 2.q.get方法可以从队列读取并且删除一个元素。同样,get方法有两个可选参数:blocked和timeout。如果blocked为True(默认值),并且timeout为正值,那么在等待时间内没有取到任何元素,会抛出Queue.Empty异常。如果blocked为False,有两种情况存在,如果Queue有一个值可用,则立即返回该值,否则,如果队列为空,则立即抛出Queue.Empty异常。 3. q.get_nowait(): 同q.get(False) 4. q.put_nowait(): 同q.put(False) 5. q.empty(): 调用此方法时q为空则返回True,该结果不可靠,比如在返回True的过程中,如果队列中又加入了项目。 6. q.full(): 调用此方法时q已满则返回True,该结果不可靠,比如在返回True的过程中,如果队列中的项目被取走。 7. q.qsize(): 返回队列中目前项目的正确数量,结果也不可靠,理由同q.empty()和q.full()一样。
应用:
''' multiprocessing模块支持进程间通信的两种主要形式:管道和队列 都是基于消息传递实现的,但是队列接口 ''' from multiprocessing import Process,Queue import time q = Queue(3) #put,get,put_nowait,get_nowait,full,empty q.put(3) q.put(3) q.put(3) print(q.full()) #满了 print(q.get()) print(q.get()) print(q.get()) print(q.empty()) #空了
.