21-Python-多进程

1、为什么需要多进程

由于GIL的存在，python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。

multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法相同，也有start(), run(), join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition类 (这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

2、多进程的创建方法

 1 import multiprocessing
 2 import time
 3 
 4 
 5 def run(name):
 6     time.sleep(2)
 7     print("hello", name)
 8 
 9 
10 # 多进程
11 if __name__ == "__main__":
12     for i in range(10):
13         p = multiprocessing.Process(target=run, args=("druid %s" % i, ))
14         p.start()
15         p.join()

3、进程间通信

不同进程间内存是不共享的，要想实现两个进程间的数据交换，可以用以下方法：

3.1、Queues

 1 from multiprocessing import Queue, Process
 2 import os
 3 
 4 
 5 def f(pq):
 6     pq.put([42, None, "druid"])
 7     print("子进程的父进程ID:", os.getppid())
 8     print("子进程的当前进程ID:", os.getpid())
 9 
10 
11 if __name__ == "__main__":
12     print("父进程ID:", os.getppid())
13     print("当前进程ID:", os.getpid())
14     q = Queue()
15     p = Process(target=f, args=(q,))
16     p.start()
17     print(q.get())
18 
19     p.join()

3.2、Pipes

The Pipe() function returns a pair of connection objects connected by a pipe which by default is duplex (two-way). For example:

 1 from multiprocessing import Process, Pipe
 2 
 3 
 4 def f(conn):
 5     conn.send("33[31;1m msg1 from child process33[0m")
 6     conn.send("33[31;1m msg2 from child process33[0m")
 7     print("from parent:", conn.recv())
 8     conn.close()
 9 
10 
11 if __name__ == "__main__":
12     parent_conn, child_conn = Pipe()
13     p = Process(target=f, args=(child_conn,))
14     p.start()
15 
16     print("from child process:", parent_conn.recv())
17     print("from child process:", parent_conn.recv())
18     # print("from child process:", parent_conn.recv())  # 如果接收次数多余发送次数，就会卡住
19     parent_conn.send("33[32;1m msg from parent process33[0m")
20 
21     p.join()

3.3、Managers

Queues和Pipes只是实现了进程间数据的传递，Managers真正实现了进程间的数据共享，即多个进程可以修改同一份数据。

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example:

 1 from multiprocessing import Process, Manager
 2 import os
 3 
 4 
 5 def f(d, l):
 6     d[os.getpid()] = os.getpid()
 7     l.append(os.getpid())
 8     print(l)
 9 
10 
11 if __name__ == "__main__":
12     with Manager() as manager:
13         d = manager.dict()
14         l = manager.list(range(5))
15         p_list = []  # 用来存放进程
16         for i in range(10):
17             p = Process(target=f, args=(d, l))
18             p.start()
19             p_list.append(p)
20         for res in p_list:
21             res.join()
22 
23         print(d)
24         print(l)

4、进程池

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

进程池中有两个方法：

- apply
- apply_async

4.1、apply方法

 1 from multiprocessing import Pool
 2 import os
 3 import time
 4 
 5 
 6 def func(i):
 7     time.sleep(2)
 8     print("子进程ID:", os.getpid())  # 每隔两秒打印一次结果
 9     return i+100
10 
11 
12 if __name__ == "__main__":
13     pool = Pool(processes=5)  # 允许进程池同时放入5个进程
14     print("主进程ID:", os.getpid())
15     for i in range(10):
16         pool.apply(func=func, args=(i,))  # 串行
17 
18     print("end")
19     pool.close()
20     pool.join()  # 进程池中进程执行完毕后再关闭，如果不加，那么程序直接关闭

4.2、apply_async方法

 1 from multiprocessing import Pool
 2 import os
 3 import time
 4 
 5 
 6 def func(i):
 7     time.sleep(2)
 8     print("子进程ID:", os.getpid())
 9     return i+100
10 
11 
12 if __name__ == "__main__":
13     pool = Pool(processes=5)  # 允许进程池同时放入5个进程
14     print("主进程ID:", os.getpid())
15     for i in range(10):
16         pool.apply_async(func=func, args=(i,))  # 并行
17 
18     print("end")  # 先打印主进程ID和"end"，然后每次打印5个子进程ID
19     pool.close()
20     pool.join()  # 进程池中进程执行完毕后再关闭，如果不加，那么程序直接关闭

4.3、apply_async+回调函数

 1 from multiprocessing import Pool
 2 import os
 3 import time
 4 
 5 
 6 def func(i):
 7     time.sleep(2)
 8     print("子进程ID:", os.getpid())
 9     return i+100
10 
11 
12 def bar(arg):
13     print("--->exec done: %s, SubPID:%s" % (arg, os.getpid()))
14 
15 
16 if __name__ == "__main__":
17     pool = Pool(processes=5)  # 允许进程池同时放入5个进程
18     print("主进程ID:", os.getpid())
19     for i in range(10):
20         pool.apply_async(func=func, args=(i,), callback=bar)  # callback为回调函数，前面的函数执行完了才会执行回调函数
21 
22     print("end")
23     pool.close()
24     pool.join()  # 进程池中进程执行完毕后再关闭，如果不加，那么程序直接关闭