python 多进程
python 多进程
Python的多线程只能运行在单核上,各个线程以并发的方法异步运行。而多进程可以利用CPU的多核,进程数取决于计算机CPU的处理器个数,由于运行在不同的核上,各个进程的运行是并行的。
在python中,如果使用多进程,需要使用multiprocessing这个库。multiprocessing模块用来开启子进程,并在子进程中执行我们定制的任务(比如函数),该模块与多线程模块threading的编程接口类似。同时该模块提供了process、Queue、Lock等组件,用于进程的创建和进程间通信。
当进程数量大于CPU的内核数量时,等待运行的进程会等到其他进程运行完让出内核为止。如果CPU单核,就无法运行多进程并行。可以使用multiprocessing库查看CPU核数。
>>>from multiprocessing import cpu_count
>>>
>>>cpu_count()
8
可知,本机的CPU核数为8。
进程创建
multiprocessing模块提供了一个Process类来构造一个子进程,结合queue来实现进程间通讯。使用Process,需要根据实际需要手动去动态创建多个进程,操作不是很方便,实际中多使用进程池。
由于进程启动的开销比较大,使用多进程的时候会导致大量内存空间被消耗。为了防止这种情况发生可以使用进程池(由于启动线程的开销比较小,所以不需要线程池这种概念,多线程只会频繁得切换cpu导致系统变慢,并不会占用过多的内存空间)。
进程池内部维护一个进程序列,当使用时,去进程池(Pool)中获取一个进程,如果进程池序列中没有可供使用的进程,那么程序就会等待,直到进程池中有可用进程为止。
>>>from multiprocessing import Pool
>>>dir(Pool())
[‘Process‘,‘__class__‘, ‘__delattr__‘, ‘__dict__‘, ‘__dir__‘, ‘__doc__‘, ‘__enter__‘,‘__eq__‘, ‘__exit__‘, ‘__format__‘, ‘__ge__‘, ‘__getattribute__‘, ‘__gt__‘,‘__hash__‘, ‘__init__‘, ‘__init_subclass__‘, ‘__le__‘, ‘__lt__‘, ‘__module__‘,‘__ne__‘, ‘__new__‘, ‘__reduce__‘, ‘__reduce_ex__‘, ‘__repr__‘, ‘__setattr__‘,‘__sizeof__‘, ‘__str__‘, ‘__subclasshook__‘, ‘__weakref__‘, ‘_cache‘, ‘_ctx‘,‘_get_tasks‘, ‘_guarded_task_generation‘, ‘_handle_results‘, ‘_handle_tasks‘,‘_handle_workers‘, ‘_help_stuff_finish‘, ‘_initargs‘, ‘_initializer‘,‘_inqueue‘, ‘_join_exited_workers‘, ‘_maintain_pool‘, ‘_map_async‘,‘_maxtasksperchild‘, ‘_outqueue‘, ‘_pool‘, ‘_processes‘, ‘_quick_get‘,‘_quick_put‘, ‘_repopulate_pool‘, ‘_result_handler‘, ‘_setup_queues‘, ‘_state‘,‘_task_handler‘, ‘_taskqueue‘, ‘_terminate‘, ‘_terminate_pool‘,‘_worker_handler‘, ‘_wrap_exception‘, ‘apply‘, ‘apply_async‘, ‘close‘, ‘imap‘,‘imap_unordered‘, ‘join‘, ‘map‘, ‘map_async‘, ‘starmap‘, ‘starmap_async‘,‘terminate‘]
>>>
创建一个进程池的方法是
pool=Pool(processes=X)
Pool有一个processes参数,这个参数可以不设置,如果不设置函数会跟根据计算机的实际情况(cpu_count)来决定要运行多少个进程,我们也可自己设置。
使用Pool类,常用的方法有
map
map()与内置的map函数用法行为基本一致,第一个参数是函数,第二个参数是一个可迭代对象,将可迭代对象中的元素作为参数依次传入函数中。如
[ newtest]# catpool.py
#encoding=utf-8
from multiprocessing import Pool
import time,os
import random
def myfunc(url):
time.sleep(random.random()*3)
print("now process is"+str(os.getpid())+" get anele "+str(url))
if __name__ =="__main__":
urls=[var for var in range(5)]
pool=Pool(processes=3) #创建有3个进程数量的进程池
pool.map(myfunc,urls)
print("main function pid is"+ str(os.getpid()))
pool.close() #关闭进程池,不再接受新的进程
pool.join() #主进程等待子进程结束
[ newtest]#
运行结果
[]# python pool.py
nowprocess is 24790 get an ele 0
nowprocess is 24790 get an ele 3
nowprocess is 24791 get an ele 1
nowprocess is 24790 get an ele 4
nowprocess is 24792 get an ele 2
mainfunction pid is 24789
[]#
Windows下,进程的创建语句必需写在if __name__ == "__main__":下。
close方法是 等待所有进程结束后,才关闭进程池。join方法是主进程等待所有子进程执行完毕(阻塞主\父进程),必须在close或terminate()之后。从结果看通过这种方法创建的进程是阻塞型进程(其他子进程执行完毕,主进程(pid=24789)才继续向下执行)。
apply_async()
创建非阻塞型进程,原型为
apply_async(func[, args[, kwds[, callback]]])
如
[]# vi poolnonblock.py
#encoding=utf-8
frommultiprocessing import Pool
importtime,os
importrandom
defmyfunc(url):
time.sleep(random.random()*3)
print("now process is"+str(os.getpid())+" get anele "+str(url))
if__name__ == "__main__":
pool=Pool(processes=3)
for i in range(10):
pool.apply_async(myfunc,(i,)) #使用元祖类型传参
print("main function pid is "+str(os.getpid()))
pool.close() #关闭进程池,不再接受新的进程
pool.join() #主进程等待子进程结束
运行结果
[]# python poolnonblock.py
mainfunction pid is 31299
nowprocess is 31301 get an ele 1
nowprocess is 31300 get an ele 0
nowprocess is 31301 get an ele 3
nowprocess is 31302 get an ele 2
nowprocess is 31301 get an ele 5
nowprocess is 31301 get an ele 7
nowprocess is 31300 get an ele 4
nowprocess is 31302 get an ele 6
nowprocess is 31301 get an ele 8
nowprocess is 31300 get an ele 9
因为子进程为非阻塞,主函数(主进程)会自己执行自个的,不搭理子进程的执行,所以主进程不会等待for循环执行完毕后才输出“main function pid is 31299”。
倘若没有pool.join()这一句,则主进程执行完毕后,子进程也就终止了,如
[]# python poolblock.py
mainfunction pid is 13702
nowprocess is 13703 get an ele 0
[]# python poolblock.py
mainfunction pid is 13973
[]#
我们可以把这一句放到主进程最后一可执行语句前面,这样的效果等同于创建阻塞子进程。如
if__name__ == "__main__":
pool=Pool(processes=3)
for i in range(10):
pool.apply_async(myfunc,(i,)) #使用元祖类型传参
pool.close() #关闭进程池,不再接受新的进程
pool.join() #主进程等待子进程结束
print("main function pid is "+str(os.getpid()))
运行结果
[~]# python poolnonblock.py
nowprocess is 12586 get an ele 1
nowprocess is 12586 get an ele 3
nowprocess is 12586 get an ele 4
nowprocess is 12587 get an ele 2
nowprocess is 12587 get an ele 6
nowprocess is 12585 get an ele 0
nowprocess is 12587 get an ele 7
now processis 12587 get an ele 9
nowprocess is 12586 get an ele 5
nowprocess is 12585 get an ele 8
mainfunction pid is 12584
又因为进程池中只能容纳有3个对象实例,小于服务器的核数(核数为4),某一进程执行完毕后,不会创建新的子进程,是刚刚空闲出来的进程去执行新的任务,进程池中的各进程pid是不变的。
若改变使得进程池中进程实例大于服务器的核数,如
….
pool=Pool(processes=5)
for i in range(10):
pool.apply_async(myfunc,(i,))
….
运行结果
[]# python poolblock.py
mainfunction pid is 30563
nowprocess is 30566 get an ele 2
nowprocess is 30568 get an ele 4
nowprocess is 30565 get an ele 1
nowprocess is 30568 get an ele 6
nowprocess is 30567 get an ele 3
nowprocess is 30567 get an ele 9
nowprocess is 30566 get an ele 5
nowprocess is 30568 get an ele 8
nowprocess is 30564 get an ele 0
nowprocess is 30565 get an ele 7
可以看到,这两种情况都维持执行的进程总数为processes,但只有后者,在当一个进程执行完毕后会添加新的进程进去。
apply ()
创建阻塞型进程。原型
apply(func[, args[, kwds]])
如修改为
…
pool=Pool(processes=3)
for iin range(4):
pool.apply(myfunc,(i,))
…
运行结果
[]# python poolblock.py
now processis 8200 get an ele 0
nowprocess is 8201 get an ele 1
nowprocess is 8202 get an ele 2
nowprocess is 8200 get an ele 3
mainfunction pid is 8199
可见主进程被阻塞到子进程执行完毕后才继续运行。阻塞型进程不需要pool.join()这一句。
我们也可以定义一系列函数,通过循环让不同的进程执行不同的函数。如
pool=Pool(processes=3)
for funcin func_list:
function_list= [func1,func2,func3] #函数名组成的列表
pool.apply_async(func)