一 : 概述
进程间通信(IPC)的方式有N种,这里我们学习FIFO队列和管道
二 : 队列的创建和使用
队列可以由multiprocessing.Queue创建,它是多线程安全的,可以实现多进程之间的数据传递.
创建 : Queue([maxsize]) , maxsize是队列中允许的最大元素数.默认为0,意为无限制.
将Queue类实例化得到对象q,q具有如下方法:
q.get( [ block [ , timeout ] ] ), 这个方法用于返回q中的第一个元素,如果为空,则阻塞,直到有元素可用为止,参数block用于控制阻塞行为,默认为True,若设置为False,则不会阻塞,引发Queue.Empty异常,参数timeout为超时时间,用在阻塞模式中,若在指定的时间内没有元素可用,引发Queue.Empty异常.
q.get_nowait()相当于block设定为False的get()方法.
q.put( item, [ , block [ , timeout ] ] ) ,将item放入队列,如果队列已满,则阻塞,直到有空间可用为止,,参数block用于控制阻塞行为,默认为True,若设置为False,则不会阻塞,引发Queue.Full异常,参数timeout为超时时间,用在阻塞模式中,若在指定的时间内没有元素可用,引发Queue.Full异常.
q.put_nowait(item),相当于block设定为False的put()方法.
q.qsize(),返回队列中目前项目的数量,不过该函数的结果并不可靠.
q.empty(),调用此方法时,如果队列为空,则返回True,不过结果不可靠,因为在返回结果和使用队列中间,可能有其他进程对队列进行了修改.
q.full(),与empt()方法相反,如果队列已经满了,则返回True,也是不可靠的,原因相同.
q.close(),关闭队列,无法再put()和get()
''' multiprocessing模块支持进程间通信的两种主要形式:管道和队列 都是基于消息传递实现的,但是队列接口 ''' from multiprocessing import Queue q=Queue(3) #put ,get ,put_nowait,get_nowait,full,empty q.put(3) q.put(3) q.put(3) # q.put(3) # 如果队列已经满了，程序就会停在这里，等待数据被别人取走，再将数据放入队列。 # 如果队列中的数据一直不被取走，程序就会永远停在这里。 try: q.put_nowait(3) # 可以使用put_nowait，如果队列满了不会阻塞，但是会因为队列满了而报错。 except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去，但是会丢掉这个消息。 print('队列已经满了') # 因此，我们再放入数据之前，可以先看一下队列的状态，如果已经满了，就不继续put了。 print(q.full()) #满了 print(q.get()) print(q.get()) print(q.get()) # print(q.get()) # 同put方法一样，如果队列已经空了，那么继续取就会出现阻塞。 try: q.get_nowait(3) # 可以使用get_nowait，如果队列满了不会阻塞，但是会因为没取到值而报错。 except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。 print('队列已经空了') print(q.empty()) #空了
加入进程通信中
import time from multiprocessing import Process, Queue def f(q): q.put([time.asctime(), 'from Eva', 'hello']) #调用主函数中p进程传递过来的进程参数 put函数为向队列中添加一条数据。 if __name__ == '__main__': q = Queue() #创建一个Queue对象 p = Process(target=f, args=(q,)) #创建一个进程 p.start() print(q.get()) p.join()
三 : 使用队列实现生产者消费者模型
1. 什么是生产者消费者模型
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
2. 为什么使用这个模型
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
3. 基于队列实现该模型
from multiprocessing import Process,Queue import time,random,os def consumer(q): while True: res=q.get() time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) def producer(q): for i in range(10): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() print('主进程')
但是这样子写有一个问题,因为消费者函数一直在运行,所以主程序也不会关闭,这时候,我们需要在生产完毕之后再向队列总加入一个None作为结束信号,即可中断程序.
from multiprocessing import Process,Queue import time,random,os def consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) def producer(q): for i in range(10): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) q.put(None) #发送结束信号 if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() print('主')
这个结束信号也可以放在主程序中发送,这就需要先运行完毕生产者,确保结束信号加在了队列最后
from multiprocessing import Process,Queue import time,random,os def consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) def producer(q): for i in range(2): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() p1.join() q.put(None) #发送结束信号 print('主') 主进程在生产者生产完毕后发送结束信号None
当存在多个生产者和消费者的时候,代码是酱紫的
from multiprocessing import Process,Queue import time,random,os def consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) def producer(name,q): for i in range(2): time.sleep(random.randint(1,3)) res='%s%s' %(name,i) q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=('包子',q)) p2=Process(target=producer,args=('骨头',q)) p3=Process(target=producer,args=('泔水',q)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) c2=Process(target=consumer,args=(q,)) #开始 p1.start() p2.start() p3.start() c1.start() p1.join() #必须保证生产者全部生产完毕,才应该发送结束信号 p2.join() p3.join() q.put(None) #有几个消费者就应该发送几次结束信号None q.put(None) #发送结束信号 print('主') 多个消费者的例子：有几个消费者就需要发送几次结束信号
需要注意的是当存在两个消费者的时候,我们在最后放入了两个中断信号,这是因为中断信号就是给消费者用的,当前一个消费者进程get了一个None之后,这个None就被取出了,这种方式很low,当消费者多的时候,需要put多个None,所以python中实现了更cool的方法.
JoinableQueue([maxsize]) ,可以创建一个可连接的队列,它的特别之处在于消费者消费数据之后会反馈给生产者一个信号,,通知是使用共享的信号和条件来实现的.
这个类是Queue的子类,其中单独定义了两个方法:task_done() 和join()
task_done() : 消费者使用此方法发出信号,表示get()取出的项目已经被处理,如果调用该方法的次数大于从队列删除的项目的数量,将引发ValueError异常.
join() : 生产者使用该方法阻塞,知道队列中的所有项目都被处理,阻塞将持续到task_done()返回信号的次数等于队列中的项目数.
from multiprocessing import Process,JoinableQueue import time,random,os def consumer(q): while True: res=q.get() time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走了 def producer(name,q): for i in range(10): time.sleep(random.randint(1,3)) res='%s%s' %(name,i) q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) q.join() #生产完毕，使用此方法进行阻塞，直到队列中所有项目均被处理。 if __name__ == '__main__': q=JoinableQueue() #生产者们:即厨师们 p1=Process(target=producer,args=('包子',q)) p2=Process(target=producer,args=('骨头',q)) p3=Process(target=producer,args=('泔水',q)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) c2=Process(target=consumer,args=(q,)) c1.daemon=True c2.daemon=True #开始 p_l=[p1,p2,p3,c1,c2] for p in p_l: p.start() p1.join() p2.join() p3.join() print('主') #主进程等--->p1,p2,p3等---->c1,c2 #p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据 #因而c1,c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了。应该随着主进程的结束而结束,所以设置成守护进程就可以了。 JoinableQueue队列实现消费之生产者模型
四 : 使用管道进行通信
创建 : 1 . from multiprocessing import Pipe 2. con1,con2 = Pipe(duplex) ,Pipe()返回的创建的对象是一个包含两个元素的元组, 这两个元素表示管道两端的连接对象,需要注意:必须在创建Process对象之前创建管道.
参数 : 默认为True,表示全双工,意为两端既可以收,也可以发,如果该参数设置为False,con1只能用于接收,con2只能用于发送(不能调换,这个方向是固定的).
主要方法 : send(obj) 通过管道发送对象
recv() 接收另一端发送的内容,如果没有可以接收的,则会阻塞,如果另一端关闭,则会报EOFError.
close() 关闭连接,如果被垃圾回收,则会自动执行该方法.
from multiprocessing import Process, Pipe def f(conn): conn.send("Hello The_Third_Wave") conn.close() if __name__ == '__main__': parent_conn, child_conn = Pipe() p = Process(target=f, args=(child_conn,)) p.start() print(parent_conn.recv()) p.join() pipe初使用
from multiprocessing import Process, Pipe def f(parent_conn,child_conn): parent_conn.close() #不写close将不会引发EOFError while True: try: print(child_conn.recv()) except EOFError: child_conn.close() if __name__ == '__main__': parent_conn, child_conn = Pipe() p = Process(target=f, args=(parent_conn,child_conn,)) p.start() child_conn.close() parent_conn.send('hello') p.join() parent_conn.close()
from multiprocessing import Process,Pipe def consumer(p,name): produce, consume=p produce.close() while True: try: baozi=consume.recv() print('%s 收到包子:%s' %(name,baozi)) except EOFError: break def producer(seq,p): produce, consume=p consume.close() for i in seq: produce.send(i) if __name__ == '__main__': produce,consume=Pipe() c1=Process(target=consumer,args=((produce,consume),'c1')) c1.start() seq=(i for i in range(10)) producer(seq,(produce,consume)) produce.close() consume.close() c1.join() print('主进程') pipe实现生产者消费者模型
五 : 使用Manager模块进行进程之间的数据共享
进程之间数据是独立的,可以借助队列和管道实现通信,二者都是基于消息传递的,虽然进程间数据独立,但是可以通过Manager模块实现数据共享,事实上这个模块的功能远不止于此.
A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies. A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array.
from multiprocessing import Manager,Process,Lock def work(d,lock): with lock: #不加锁而操作共享的数据,肯定会出现数据错乱 d['count']-=1 if __name__ == '__main__': lock=Lock() with Manager() as m: dic=m.dict({'count':100}) p_l=[] for i in range(100): p=Process(target=work,args=(dic,lock)) p_l.append(p) p.start() for p in p_l: p.join() print(dic) Manager例子
五 : 进程池
当有n多个进程的时候 ,我们可以用进程池来管理这些进程.
创建 : 1. from multiprocessing import Pool 2. p = Pool([numprocess [,initializer [, initargs]]])
参数: numprocess : 传入的是一个int类型的数字,表示池中的最大进程数量,如果不给出则会通过os.cpu_count()获取,如果获取不到,则会赋值1,我们自定义这个数字一般是cpu数目+1.
initializer : 是每个工作进程启动时要执行的可调用对象,默认为None.
initargs : 传给initializer的参数组
主要方法:
1. map(func, iterable[, chunksize=None])
Pool类中的map方法，与内置的map函数用法行为基本一致，它会使进程阻塞直到结果返,返回值是个list.
2. apply(func [, args [, kwargs]])
同步得执行任务,并返回结果.主进程会被阻塞直到函数执行结束
3. apply_async(func[, args=()[, kwds={}[, callback=None]]])
与apply用法一致，但它是非阻塞的且支持结果返回后进行回调,异步得执行任务.
4. close()
关闭进程池,使其不再接受新的任务,但是不妨碍工作进程继续执行现有的任务.
5. terminal()
结束工作进程,不再处理未处理的任务.
6. join()
主进程阻塞等待子进程的退出,要在close或者terminal之后使用.
map进程池和多进程做任务的效率对比 :
from multiprocessing import Pool,Process import time def func(i): i += 1 print(i) if __name__ == '__main__': p = Pool(4)# 创建进程池，池中有4个进程待命 start = time.time() t = [i for i in range(100)] p.map(func,t)# p.close() p.join()# 等待进程池中所有进程都执行完所有任务。 print(time.time() - start)# 打印进程池做任务的时间 start = time.time() l = [] for i in range(100): p = Process(target=func,args=(i,)) p.start() l.append(p) [i.join() for i in l]# 等待所有进程工作结束 print(time.time() - start)# 打印开启100个进程做任务的时间 print('--'*20+'>') map进程池和多进程做任务的效率对比
同步调用和异步调用的区别 :
import os,time from multiprocessing import Pool def work(n): print('%s run' %os.getpid()) time.sleep(1) return n**2 if __name__ == '__main__': p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务 for i in range(10): res=p.apply(work,args=(i,)) # 同步调用，直到本次任务执行完毕拿到res，等待任务work执行的过程中可能有阻塞也可能没有阻塞 # 但不管该任务是否存在阻塞，同步调用都会在原地等着 # 进程池的同步调用
import os import time import random from multiprocessing import Pool def work(n): print('%s run' %os.getpid()) time.sleep(random.random()) return n**2 if __name__ == '__main__': p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务 res_l=[] for i in range(10): res=p.apply_async(work,args=(i,)) # 异步运行，根据进程池中有的进程数，每次最多3个子进程在异步执行 # 返回结果之后，将结果放入列表，归还进程，之后再执行新的任务 # 需要注意的是，进程池中的三个进程不会同时开启或者同时结束 # 而是执行完一个就释放一个进程，这个进程就去接收新的任务。 res_l.append(res) # 异步apply_async用法：如果使用异步提交的任务，主进程需要使用jion，等待进程池内任务都处理完，然后可以用get收集结果 # 否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了 p.close() p.join() for res in res_l: print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get # 进程池的异步调用
回调函数 :
需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数.
from multiprocessing import Pool import requests import json import os def get_page(url): print('<进程%s> get %s' %(os.getpid(),url)) respone=requests.get(url) if respone.status_code == 200: return {'url':url,'text':respone.text} def pasrse_page(res): print('<进程%s> parse %s' %(os.getpid(),res['url'])) parse_res='url:<%s> size:[%s] ' %(res['url'],len(res['text'])) with open('db.txt','a') as f: f.write(parse_res) if __name__ == '__main__': urls=[ 'https://www.baidu.com', 'https://www.python.org', 'https://www.openstack.org', 'https://help.github.com/', 'http://www.sina.com.cn/' ] p=Pool(3) res_l=[] for url in urls: res=p.apply_async(get_page,args=(url,),callback=pasrse_page) res_l.append(res) p.close() p.join() print([res.get() for res in res_l]) #拿到的是get_page的结果,其实完全没必要拿该结果,该结果已经传给回调函数处理了 ''' 打印结果: <进程3388> get https://www.baidu.com <进程3389> get https://www.python.org <进程3390> get https://www.openstack.org <进程3388> get https://help.github.com/ <进程3387> parse https://www.baidu.com <进程3389> get http://www.sina.com.cn/ <进程3387> parse https://www.python.org <进程3387> parse https://help.github.com/ <进程3387> parse http://www.sina.com.cn/ <进程3387> parse https://www.openstack.org [{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html> ...',...}] ''' 使用多进程请求多个url来减少网络等待浪费的时间
如果在主进程中等待进程池中所有任务都执行完毕后，再统一处理结果，则无需回调函数