multiprocessing总结

Posted 2020-09-27

由于GIL的存在，python的多线程并不是真正的多线程。如果想充分的时候多核CPU的资源，在Cpython中大部分情况下需要使用到多进程（multiprocess）。

Python通过“multiprocessing”来实现多进程并发的功能。

 

multiprocessing支持的功能：

1. 子进程
2. 进程间通信/共享
3. 进程同步

在使用multiprocessing时候的注意点：

• 在UNIX平台上，当某个进程终结之后，该对其父进程调用wait，否则进程将成为僵尸进程（Zombie），所以每个Process对象都需要调用join()方法。
• multiprocessing提供了IPC（inter process communication），推荐使用Queue或pipe
• 多进程应该避免资源共享，因为会增加复杂性和降低程序效率
• 在windows下，子进程的启动必须在 if __name__ == "__main__":语句后面

创建多进程的方法：

1，直接传入

2，从Process类继承

 

直接传入代码：

 1 from multiprocessing import Process
 2 
 3 def foo(num):
 4     print("sub_process :", num)
 5 
 6 if __name__ =="__main__":
 7     for i in range(10):
 8         p = Process(target=foo,args=(i,))
 9         p.start()
10         p.join()

通过类实现代码：

 1 from multiprocessing import Process
 2 
 3 class Myprocess(Process):
 4     def __init__(self,arg):
 5         super(Myprocess,self).__init__()
 6         self.arg = arg
 7 
 8     def run(self):
 9         print("sub_process :" , self.arg)
10 
11 if __name__ == "__main__":
12     for i in range(10):
13         p = Myprocess(i)
14         p.start()
15         

 

-----------------------------------------------------------------------------------------------------------------------------------------------------

Process类

构造方法：

Process（[group , [target ,[name ,[ args ,[kwargs ] ] ] ] ] ）

　　group：线程组

　　target：要执行的函数

　　name：进程名字

　　args/kwargs：函数的参数

实例方法：

　　is_alive()：返回进程是否正在运行

　　join([timeout[)：阻塞当前上下文环境的进程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。

　　start()：进程准备就绪，等待CPU调用

　　run()：start()调用run方法

　　terminate()：不管任务是否完成，立即停止工作进程

 

属性：

　　name：进程名

　　pid：进程号

 

-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

 

pool类：（进程池）

　　进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，之道进程池中有可用进程为止。

进程池数量最好等于CPU核心数量。

 

构造方法：

Pool（[processes , [ initializer , [ initargs , [ maxtasksperchild , [ context ] ] ] ] ]）

• processes：使用的工作进程数量，如果processes是None，那么使用os.cpu_count（）返回的数量
• initializer：如果initializer是None，那么每一个工作进程在开始的时候会调用initializer（*initargs）
• maxtasksperichild：工作进程退出之前可以完成的任务数，完成后用一个新的工作进程来替换原进程，来让限制的资源被释放。默认值为None，意味着只要pool存在，工作进程就会一直存活。
• context：用在指定工作进程启动时的上下文，一般使用multiprocessing.Pool（）或者一个context对象的pool（）方法来创建一个池，两种方法都适当的设置了context

 

实例方法：

　　apply （fun , [ args , [ kwargs ] ] ）：同步进程池

　　apply_async（fun , [args, [ kwargs  [ callback , [error_callback ] ] ] ] ）异步进程池

　　close（）关闭进程池，阻止更多的任务提交到pool，待任务完成后，工作进程会退出。

　　terminate（）：结束工作进程，不在处理未完成的任务

　　join（）：wait工作线程的退出，在调用join（）前，必须调用close（） or terminate（）。这是因为被终止的进程需要被父进程调用wait（join与wait等价），否则进程会成为僵尸进程。

 

例子：（异步进程池）

　　

 1 def foo(num):
 2     time.sleep(2)
 3     print(‘Run process : %d‘ % num)
 4 
 5 def bar(arg):
 6     print("Error")
 7 
 8 if __name__ == "__main__":
 9     pool = Pool(8)
10     for i in range(16):
11         pool.apply_async(func=foo , args=(i,) , error_callback=bar)
12     pool.close()
13     pool.join()
14     print("---Run Over---")

 

例子：（同步进程池）

 1 #!/usr/bin/env python3
 2 #-*- coding:utf-8 -*-
 3 
 4 from multiprocessing import Pool
 5 import time
 6 
 7 def foo(num):
 8     time.sleep(2)
 9     print(‘Run process : %d‘ % num)
10 
11 def bar(arg):
12     print("Error")
13 
14 if __name__ == "__main__":
15     pool = Pool(8)
16     for i in range(16):
17         pool.apply(func=foo,args=(i,))
18     pool.close()
19     pool.join()
20     print("---Run Over---")

 

区别：

异步进程池在调用进程的时候没有顺序，同步有顺序，但是同步牺牲了效率。