[TOC] ## **进程池** ### **为什么要有进程池** 在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？ 首先，创建进程需要消耗时间，销毁进程也需要消耗时间。 第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，这样反而会影响程序的效率。 因此我们不能无限制的根据任务开启或者结束进程。 ### **进程池的概念** 定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。 如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。 就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果。 ### 多进程和进程池的对比 对于纯计算型的代码 使用进程池更好 —— 真理 对于高IO的代码 直接使用多进程更好 —— 相对论 **结论:** 进程池比起多进程来说 节省了开启进程回收进程资源的时间,给操作系统调度进程降低了难度 ## **用法语法** Pool([numprocess [,initializer [, initargs]]]):创建进程池 ### **参数介绍** 1. numprocess: 要创建的进程数，如果省略，将默认使用cpu_count()的值 2. initializer： 是每个工作进程启动时要执行的可调用对象，默认为None 3. initargs： 是要传给initializer的参数组 ### **主要方法** 1. p.apply(func [, args [, kwargs]]): 在一个池工作进程中执行`func(*args,**kwargs)`,然后返回结果。(**同步**) 2. p.apply_async(func [, args [, kwargs]]): 在一个池工作进程中执行`func(*args,**kwargs)`,然后返回结果。(**异步**) 回调函数后面讲 3. p.close(): 关闭进程池，防止进一步操作。 4. p.jion(): 等待所有工作进程退出。此方法只能在close（）或teminate()之后调用 5. p.map() 接收一个任务函数，和一个iterable。节省了for循环和close、join，是一种简便的写法 ### 其他方法(了解) 方法apply_async和map_async的返回值是AsyncResul的实例obj。实例具有以下方法 1. obj.get(): 返回结果，如果有必要则等待结果到达。 timeout是可选的。如果在指定时间内还没有到达，将引发异常。 如果远程操作中引发了异常，它将在调用此方法时再次被引发。 2. obj.ready(): 如果调用完成，返回True 3. obj.successful(): 如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常 4. obj.wait([timeout]): 等待结果变为可用。 5. obj.terminate()： 立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数 ## 代码实例 ### apply_async代码举例 ~~~ from multiprocessing import Pool import os,time,random def work(i): print('%s is 运行,PID:%s'%(i,os.getpid())) time.sleep(random.random()) return i**i if __name__ == '__main__': p=Pool(3) #进程数量为3的进程池 res_1=[] for i in range(5): res=p.apply_async(func=work,args=(i,)) res_1.append(res) p.close() p.join() print('end'.center(15,'-')) for r in res_1: print(r.get()) ~~~ 说明: ~~~ # 异步运行，根据进程池中有的进程数，每次最多3个子进程在异步执行 # 返回结果之后，将结果放入列表，归还进程，之后再执行新的任务 # 需要注意的是，进程池中的三个进程不会同时开启或者同时结束 # 而是执行完一个就释放一个进程，这个进程就去接收新的任务。 ~~~ 运行结果: ``` 0 is 运行,PID:5732 1 is 运行,PID:5252 2 is 运行,PID:6472 3 is 运行,PID:5732 4 is 运行,PID:6472 ------end------ 1 1 4 27 256 ``` ### map代码举例 以下两段代码完全等价,map帮我们完成了for循环,close,join任务 ~~~ p = Pool(5) for i in range(10): p.apply_async(func=wahaha,args=(i,)) # 异步提交了一个任务 p.close() p.join() ~~~ ~~~ p = Pool(5) p.map(func=wahaha,iterable=range(10)) ~~~ apply_async和map相比，操作复杂，但是可以通过get方法获取返回值 ## **回调函数** ### 需要回调函数的场景： 进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数 我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。 ### 代码举例 ~~~ from multiprocessing import Pool import os,time,random def work1(i): print('%s is 运行,PID:%s'%(i,os.getpid())) time.sleep(random.random()) return i*i def work2(f): # 回调函数用的是主进程中的资源 print(f,os.getpid()) if __name__ == '__main__': p=Pool(3) #进程数量为3的进程池 res_1=[] print('主进程:%s'%os.getpid()) p.apply_async(func=work1,args=(9,),callback=work2) # callback是一个回调函数，接收一个函数地址 p.close() p.join() ~~~ 执行结果: ``` 主进程:8080 9 is 运行,PID:2988 81 8080 ```