python 多进程

python 多进程

python 多进程
0.0722019.10.10 09:23:18字数 1,999阅读 378

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)   

相关推荐