第35篇进程之间的通信 Queue Pipe 进程池Pool，p.apply()方法，p.apply_async()方法 _进程

万事须己运，他得非我贤。这篇文章主要讲述第35篇进程之间的通信 Queue Pipe 进程池Pool，p.apply()方法，p.apply_async()方法相关的知识，希望能为你提供帮助。

内容大纲：
进程之间的通讯
进程队列
管道

进程之间的数据共享
进程池
使用进程池开启进程
提交任务
获得返回值
回调函数
1.进程队列
先进先出

from multiprocessing import Queue import queue q = Queue() q.put(1) q.put(2) q.put(3) print(q.get()) print(q.get()) print(q.get())

1 2 3

from multiprocessing import Queue import queue q = Queue() q.put(1) q.put(2) q.put(3) print(q.get()) print(q.get()) print(q.get()) print(q.get())#q已经被取空没法取值程序会被夯住

from multiprocessing import Queue import queue q = Queue() q.put(1) q.put(2) q.put(3) print(q.get()) print(q.get()) print(q.get()) print(q.get_nowait())#报错queue.Empty

1 2 3

from multiprocessing import Queue import queue q = Queue() q.put(1) q.put(2) q.put(3) while True: try: print(q.get_nowait()) except queue.Empty: break

1 2 3

from multiprocessing import Queue import queue q = Queue(3)#设置队列的最大容量 q.put(1) q.put(2) q.put(3) q.put(4)#队列已经放满了程序被夯住

from multiprocessing import Queue import queue q = Queue(3)#设置队列的最大容量 while True: try: q.put_nowait(1)#队列放满了报出异常 except queue.Full: breakwhile True: try: print(q.get_nowait()) except queue.Empty: break

1 1 1

#q.empty()q.full()
这两个方法不是很可靠，因为别的进程会随时往队列里面添加或者取走元素

from multiprocessing import Queue import queue q = Queue(3)#设置队列的最大容量 while True: try: q.put_nowait(1)#队列放满了报出异常 except queue.Full: break print(q.empty())#判断队列是否为空 print(q.full())#判断队列是否已满 while True: try: print(q.get_nowait()) except queue.Empty: break print(q.empty()) print(q.full())

False True 1 1 1 True False

from multiprocessing import Process,Queue def consume(q): print(q.get())if __name__ == ‘__main__‘: q = Queue() p = Process(target= consume,args=(q,)) p.start() q.put({‘123‘:456})

{‘123‘: 456}

from multiprocessing import Process,Queue def consume(q): print(‘son ---> ‘,q.get()) q.put(‘abc‘)if __name__ == ‘__main__‘: q = Queue() p = Process(target= consume,args=(q,)) p.start() q.put({‘123‘:456}) p.join() print(‘Foo ---> ‘, q.get())

son - --> {‘123‘: 456} Foo - --> abc

2.什么是生产者消费者模型

import time import random from multiprocessing import Process,Queuedef consumer(q,name): while True: food = q.get()#循环不停的从队列里面取走元素 if food is None:break#取到None，退出循环 time.sleep(random.uniform(0.5,1)) print(‘%s吃了:%s‘%(name,food))def producer(q,name,food): for i in range(10): time.sleep(random.uniform(0.3,0.8)) print(‘%s 生产了:%s%s‘%(name,food,i)) q.put(food+str(i))if __name__ == ‘__main__‘: q = Queue()c1 = Process(target=consumer,args=(q,‘alex‘)) c1.start()p1 = Process(target=producer,args=(q,‘沙县小吃‘,‘鸡腿‘)) p1.start()p1.join()#队列里面放元素设置成一个同步事件. q.put(None)#生产（队列里面添加元素）结束之后最后放一个None

沙县小吃生产了:鸡腿0 沙县小吃生产了:鸡腿1 alex吃了:鸡腿0 沙县小吃生产了:鸡腿2 alex吃了:鸡腿1 沙县小吃生产了:鸡腿3 沙县小吃生产了:鸡腿4 alex吃了:鸡腿2 沙县小吃生产了:鸡腿5 沙县小吃生产了:鸡腿6 alex吃了:鸡腿3 沙县小吃生产了:鸡腿7 alex吃了:鸡腿4 沙县小吃生产了:鸡腿8 alex吃了:鸡腿5 沙县小吃生产了:鸡腿9 alex吃了:鸡腿6 alex吃了:鸡腿7 alex吃了:鸡腿8 alex吃了:鸡腿9

3.可阻塞的队列 JoinableQueue
多了两个方法 q.task_done() q.join()

import time import random from multiprocessing import Process,JoinableQueuedef consumer(q,name): while True: food = q.get()#循环不行的从队列里面取走元素 # if food is None:break 这句代码在JoinableQueue中就不需要了 time.sleep(random.uniform(0.5,1)) print(‘%s吃了:%s‘%(name,food)) q.task_done()#完成了任务向队列汇报 #只有消费者里面才需要汇报 def producer(q,name,food): for i in range(5): time.sleep(random.uniform(0.3,0.8)) print(‘%s 生产了:%s%s‘%(name,food,i)) q.put(food+str(i))if __name__ == ‘__main__‘:jq = JoinableQueue()#可阻塞的队列c1 = Process(target=consumer,args=(jq,‘alex‘)) c2 = Process(target=consumer,args=(jq,‘taibai‘)) c1.daemon = True c2.daemon = True c1.start() c2.start()p1 = Process(target=producer,args=(jq,‘沙县小吃‘,‘鸡腿‘)) p2 = Process(target=producer,args=(jq,‘黄焖鸡‘,‘炒米粉‘)) p1.start() p2.start()p1.join()#生产者把所有的元素都放到队列里面才停止 p2.join()jq.join()#可阻塞的队列设置成阻塞的

程序执行完成后结束：

沙县小吃生产了:鸡腿0 黄焖鸡生产了:炒米粉0 黄焖鸡生产了:炒米粉1 alex吃了:鸡腿0 沙县小吃生产了:鸡腿1 黄焖鸡生产了:炒米粉2 taibai吃了:炒米粉0 沙县小吃生产了:鸡腿2 alex吃了:炒米粉1 黄焖鸡生产了:炒米粉3 alex吃了:炒米粉2 taibai吃了:鸡腿1 沙县小吃生产了:鸡腿3 黄焖鸡生产了:炒米粉4 沙县小吃生产了:鸡腿4 alex吃了:鸡腿2 taibai吃了:炒米粉3 alex吃了:鸡腿3 taibai吃了:炒米粉4 alex吃了:鸡腿4

4，什么是管道？
管道有左右两端，左边发送右边接收，或者右边发送，左边接收

from multiprocessing import Pipe left,right = Pipe() left.send(‘hello‘) print(right.recv())

hello

from multiprocessing importProcess,Pipe def consumer(pipe): print(pipe[1].recv())#pipe[1].recv()管道的右边接收 if __name__ == ‘__main__‘: pipe = Pipe() Process(target=consumer,args=(pipe,)).start() pipe[0].send(‘hello‘)

hello

from multiprocessing importProcess,Pipe def consumer(left,right): print(right.recv())#pipe[1].r ecv()管道的右边接收 if __name__ == ‘__main__‘: left,right = Pipe()#生成的是一个元组(左端，右端) Process(target=consumer,args=(left,right)).start() left.send(‘hello‘)

hello

管道端口的关闭

from multiprocessing importPipe,Process def consumer(left,right): left.close() while True: try: print(right.recv()) except EOFError: break if __name__ == ‘__main__‘: left,right = Pipe() Process(target=consumer,args=(left,right)).start() right.close() for i in range(5): left.send(‘鸡腿%s‘%i) left.close()

鸡腿0 鸡腿1 鸡腿2 鸡腿3 鸡腿4

总结一下：
队列是基于管道实现的
管道是基于socket实现的
队列+锁是一种简便的IPC机制，是的进程之间的数据变得安全，
什么是IPC inter-process-commucate进程之间的通讯
socket+pickle实现进程之间的通讯，同一台计算机通过文件的收发实现进程之间的通讯

5，什么是进程池？为什么要有进程池？
开启过多的进程并不能够提高效率，反而会降低效率

进程的分类：
计算密集型：
重分占用CPU，多进程可以充利用CPU的多核
适合开启多个进程
IO密集型：
大部分的时间都在阻塞队列，而不是在运行状态，
根本不适合开启多个进程

信号量，多进程，进程池的概念区别：
现在需要生产500件衣服，应该买几台机器？雇佣几名工人？？？

信号量模式：
500件衣服要生产500个任务
雇佣500个人开启了500个进程
购买4台机器4核CPU

多进程模式：
500件衣服要生产500个任务
雇佣500个人开启了500个进程
购买4台机器4核CPU

进程池模式：
500件衣服要生产500个任务
雇佣4个人4个人一人一台机器，不停地生产
购买4台机器4核CPU

importtime frommultiprocessing import Pool,Process def func(num): print(‘生产了第%s件衣服‘%num)if __name__ == ‘__main__‘: start = time.time() p = Pool(4)#创建进程池池子的最大进程是4个进程for i in range(100): p.apply_async(func,args=(i,))#异步提交 func到子进程中执行 p.close()#关闭池，用户不能再向池中提交任务 p.join()#阻塞，直到进程池中所有的进程都执行完毕，主进程才能结束 print(time.time()-start)

生产了第0件衣服 ... 生产了第99件衣服 2.899761915206909#时间消耗

#多进程的方式

importtime frommultiprocessing import Pool,Process def func(num): print(‘生产了第%s件衣服‘%num)if __name__ == ‘__main__‘: start = time.time() p_list = [] for i in range(10): p = Process(target=func,args=(i,)) p.start() p_list.append(p) for p in p_list: p.join() print(time.time()-start)

生产了第1件衣服生产了第9件衣服生产了第2件衣服生产了第3件衣服生产了第4件衣服生产了第5件衣服生产了第0件衣服生产了第8件衣服生产了第6件衣服生产了第7件衣服 5.608381509780884

同步提交与异步提交的区别

import time,os from multiprocessing import Pooldef task(num): time.sleep(1) print(‘%s : %s ‘%(num,os.getpid()))if __name__ == ‘__main__‘: p = Pool(4) for i in range(20): p.apply(task,args=(i,))#提交任务的方式是同步提交

0 : 4424 1 : 2748 2 : 9464 3 : 4176 #后面的pid是不断的重复上面4个pid（进程编号） 4 : 4424 5 : 2748 6 : 9464 7 : 4176 8 : 4424 9 : 2748 10 : 9464 11 : 4176 12 : 4424 13 : 2748 14 : 9464 15 : 4176 16 : 4424 17 : 2748 18 : 9464 19 : 4176

#同提交的方法可以得到返回值

import time,os from multiprocessing import Pooldef task(num): time.sleep(1) print(‘%s : %s ‘%(num,os.getpid())) return num**2if __name__ == ‘__main__‘: p = Pool(4) for i in range(20): res = p.apply(task,args=(i,))#提交任务的方式是同步提交 print(‘--> ‘,res)

0 : 9544 --> 0 1 : 4760 --> 1 2 : 9028 --> 4 3 : 9572 --> 9 4 : 9544 --> 16 5 : 4760 --> 25 6 : 9028 --> 36 7 : 9572 --> 49 8 : 9544 --> 64 9 : 4760 --> 81 10 : 9028 --> 100 11 : 9572 --> 121 12 : 9544 --> 144 13 : 4760 --> 169 14 : 9028 --> 196 15 : 9572 --> 225 16 : 9544 --> 256 17 : 4760 --> 289 18 : 9028 --> 324 19 : 9572 --> 361

#异步提交不能拿到任务的结果但是可以拿到任务提交的情况

import time,os from multiprocessing import Pooldef task(num): time.sleep(1) print(‘%s : %s ‘%(num,os.getpid())) return num**2if __name__ == ‘__main__‘: p = Pool() for i in range(5): res = p.apply_async(task,args=(i,))#提交任务的方式是异步提交 print(‘--> ‘,res)

--> < multiprocessing.pool.ApplyResult object at 0x0000000DBF84C438> --> < multiprocessing.pool.ApplyResult object at 0x0000000DBF84C518> --> < multiprocessing.pool.ApplyResult object at 0x0000000DBF84C5C0> --> < multiprocessing.pool.ApplyResult object at 0x0000000DBF84C6A0> --> < multiprocessing.pool.ApplyResult object at 0x0000000DBF84C780>

import time,os from multiprocessing import Pooldef task(num): time.sleep(1) print(‘%s : %s ‘%(num,os.getpid()))if __name__ == ‘__main__‘: p = Pool() for i in range(5): p.apply_async(task,args=(i,))#提交任务的方式是同步提交 p.close()#关闭池子不能再往里面添加任务 p.join()#进程池设置成阻塞任务完成了主进程成才能关闭

0 : 10100 1 : 624 2 : 8288 3 : 9532 4 : 10100

进程池总结：
p = Pool()实例化的时候进程的个数默认值是cpu的个数u，或者设置成cpu+1
提交任务：
同步提交：apply(函数名，args =())
#有返回值，返回值是子函数逇返回值
#一个任务接着一个任务按顺序同步执行，没有任何并发的结果
异步提交：apply_async
#返回值是任务提交的结果
#p.close()
#p.join()
#必须先close()再join()，p设置成阻塞，直到p中所有的进程都执行完毕，才结束主进程

#这种法法取值，与同步提交没有区别

import time,os from multiprocessing import Pooldef task(num): time.sleep(1) print(‘%s : %s ‘%(num,os.getpid())) return num**2if __name__ == ‘__main__‘: p = Pool() for i in range(5): res = p.apply_async(task,args=(i,))#提交任务的方式是同步提交 print(res.get())

0: 8836 0 1: 9248 1 2: 9780 4 3: 2588 9 4: 8836 16

#将计算的结果放进列表

import time,os from multiprocessing import Pooldef task(num): time.sleep(1) print(‘%s : %s ‘%(num,os.getpid())) return num**2if __name__ == ‘__main__‘: p = Pool() res_lsit = [] for i in range(5): res = p.apply_async(task,args=(i,))#提交任务的方式是同步提交 res_lsit.append(res)#将计算的结果放进列表 for res in res_lsit: print(res.get())

0 : 9424 0 1 : 912 1 2 : 8572 4 3 : 7436 9 4 : 9424 16

#p.map(函数名,参数)

import time,os from multiprocessing import Pooldef task(num): time.sleep(1) print(‘%s : %s ‘%(num,os.getpid())) return num**2if __name__ == ‘__main__‘: p = Pool() p.map(task,range(5))