python模块multiprocess多进程

简介

multiprocess提供了Process类&＃xff0c;实现进程相关的功能。但是它基于fork机制&＃xff0c;因此不被windows平台支持。想要在windows中运行&＃xff0c;必须使用if __name__ &＃61;&＃61; &＃39;__main__&＃39;:的方式&＃xff0c;显然这只能用于调试和学习&＃xff0c;不能用于实际环境。

另外&＃xff0c;在multiprocess中你既可以import大写的Process&＃xff0c;也可以import小写的process&＃xff0c;这两者是完全不同的东西。这种情况在Python中很多&＃xff0c;请一定要小心和注意。

下面是一个简单的多进程例子&＃xff0c;Process类的用法和Thread类几乎一模一样。

import os
import multiprocessingdef foo(i):# 同样的参数传递方法print("这里是 ", multiprocessing.current_process().name)print(&＃39;模块名称:&＃39;, __name__)print(&＃39;父进程 id:&＃39;, os.getppid()) # 获取父进程idprint(&＃39;当前子进程 id:&＃39;, os.getpid()) # 获取自己的进程idprint(&＃39;------------------------&＃39;)if __name__ &＃61;&＃61; &＃39;__main__&＃39;:for i in range(5):p &＃61; multiprocessing.Process(target&＃61;foo, args&＃61;(i,))p.start()


运行结果&＃xff1a;

这里是 Process-2
模块名称: __mp_main__
父进程 id: 880
当前子进程 id: 5260
--------------
这里是 Process-3
模块名称: __mp_main__
父进程 id: 880
当前子进程 id: 4912
--------------
这里是 Process-4
模块名称: __mp_main__
父进程 id: 880
当前子进程 id: 5176
--------------
这里是 Process-1
模块名称: __mp_main__
父进程 id: 880
当前子进程 id: 5380
--------------
这里是 Process-5
模块名称: __mp_main__
父进程 id: 880
当前子进程 id: 3520
--------------


进程间的数据共享

在Linux中&＃xff0c;每个子进程的数据都是由父进程提供的&＃xff0c;每启动一个子进程就从父进程克隆一份数据。

创建一个进程需要非常大的开销&＃xff0c;每个进程都有自己独立的数据空间&＃xff0c;不同进程之间通常是不能共享数据的&＃xff0c;要想共享数据&＃xff0c;一般通过中间件来实现。

下面我们尝试用一个全局列表来实现进程间的数据共享&＃xff1a;

from multiprocessing import Processlis &＃61; []def foo(i):lis.append(i)print("This is Process ", i," and lis is ", lis, " and lis.address is ", id(lis))if __name__ &＃61;&＃61; &＃39;__main__&＃39;:for i in range(5):p &＃61; Process(target&＃61;foo, args&＃61;(i,))p.start()print("The end of list_1:", lis)


运行结果&＃xff1a;

The end of list_1: []
This is Process 2 and lis is [2] and lis.address is 40356744
This is Process 1 and lis is [1] and lis.address is 40291208
This is Process 0 and lis is [0] and lis.address is 40291208
This is Process 3 and lis is [3] and lis.address is 40225672
This is Process 4 and lis is [4] and lis.address is 40291208


可以看到&＃xff0c;全局列表lis没有起到任何作用&＃xff0c;在主进程和子进程中&＃xff0c;lis指向内存中不同的列表。

想要在进程之间进行数据共享可以使用Queues、Array和Manager这三个multiprocess模块提供的类。

使用Array共享数据

对于Array数组类&＃xff0c;括号内的“i”表示它内部的元素全部是int类型&＃xff0c;而不是指字符“i”&＃xff0c;数组内的元素可以预先指定&＃xff0c;也可以只指定数组的长度。Array类在实例化的时候必须指定数组的数据类型和数组的大小&＃xff0c;类似temp &＃61; Array(&＃39;i&＃39;, 5)。对于数据类型有下面的对应关系&＃xff1a;

&＃39;c&＃39;: ctypes.c_char, &＃39;u&＃39;: ctypes.c_wchar,
&＃39;b&＃39;: ctypes.c_byte, &＃39;B&＃39;: ctypes.c_ubyte,
&＃39;h&＃39;: ctypes.c_short, &＃39;H&＃39;: ctypes.c_ushort,
&＃39;i&＃39;: ctypes.c_int, &＃39;I&＃39;: ctypes.c_uint,
&＃39;l&＃39;: ctypes.c_long, &＃39;L&＃39;: ctypes.c_ulong,
&＃39;f&＃39;: ctypes.c_float, &＃39;d&＃39;: ctypes.c_double


看下面的例子&＃xff1a;

from multiprocessing import Process
from multiprocessing import Arraydef func(i,temp):temp[0] &＃43;&＃61; 100print("进程%s " % i, &＃39; 修改数组第一个元素后----->&＃39;, temp[0])if __name__ &＃61;&＃61; &＃39;__main__&＃39;:temp &＃61; Array(&＃39;i&＃39;, [1, 2, 3, 4])for i in range(10):p &＃61; Process(target&＃61;func, args&＃61;(i, temp))p.start()


运行结果&＃xff1a;

进程2 修改数组第一个元素后-----> 101
进程4 修改数组第一个元素后-----> 201
进程5 修改数组第一个元素后-----> 301
进程3 修改数组第一个元素后-----> 401
进程1 修改数组第一个元素后-----> 501
进程6 修改数组第一个元素后-----> 601
进程9 修改数组第一个元素后-----> 701
进程8 修改数组第一个元素后-----> 801
进程0 修改数组第一个元素后-----> 901
进程7 修改数组第一个元素后-----> 1001


使用Manager共享数据

通过Manager类也可以实现进程间数据的共享。Manager()返回的manager对象提供一个服务进程&＃xff0c;使得其他进程可以通过代理的方式操作Python对象。manager对象支持 list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value ,Array等多种格式。

from multiprocessing import Process
from multiprocessing import Managerdef func(i, dic):dic["num"] &＃61; 100&＃43;iprint(dic.items())if __name__ &＃61;&＃61; &＃39;__main__&＃39;:dic &＃61; Manager().dict()for i in range(10):p &＃61; Process(target&＃61;func, args&＃61;(i, dic))p.start()p.join()


运行结果&＃xff1a;

[(&＃39;num&＃39;, 100)]
[(&＃39;num&＃39;, 101)]
[(&＃39;num&＃39;, 102)]
[(&＃39;num&＃39;, 103)]
[(&＃39;num&＃39;, 104)]
[(&＃39;num&＃39;, 105)]
[(&＃39;num&＃39;, 106)]
[(&＃39;num&＃39;, 107)]
[(&＃39;num&＃39;, 108)]
[(&＃39;num&＃39;, 109)]


使用queues的Queue类共享数据

multiprocessing是一个包&＃xff0c;它内部又一个queues模块&＃xff0c;提供了一个Queue队列类&＃xff0c;可以实现进程间的数据共享&＃xff0c;如下例所示&＃xff1a;

import multiprocessing
from multiprocessing import Process
from multiprocessing import queuesdef func(i, q):ret &＃61; q.get()print("进程%s从队列里获取了一个%s&＃xff0c;然后又向队列里放入了一个%s" % (i, ret, i))q.put(i)if __name__ &＃61;&＃61; "__main__":lis &＃61; queues.Queue(20, ctx&＃61;multiprocessing)lis.put(0)for i in range(10):p &＃61; Process(target&＃61;func, args&＃61;(i, lis,))p.start()


运行结果&＃xff1a;

进程1从队列里获取了一个0&＃xff0c;然后又向队列里放入了一个1
进程4从队列里获取了一个1&＃xff0c;然后又向队列里放入了一个4
进程2从队列里获取了一个4&＃xff0c;然后又向队列里放入了一个2
进程6从队列里获取了一个2&＃xff0c;然后又向队列里放入了一个6
进程0从队列里获取了一个6&＃xff0c;然后又向队列里放入了一个0
进程5从队列里获取了一个0&＃xff0c;然后又向队列里放入了一个5
进程9从队列里获取了一个5&＃xff0c;然后又向队列里放入了一个9
进程7从队列里获取了一个9&＃xff0c;然后又向队列里放入了一个7
进程3从队列里获取了一个7&＃xff0c;然后又向队列里放入了一个3
进程8从队列里获取了一个3&＃xff0c;然后又向队列里放入了一个8


关于queue和Queue&＃xff0c;在Python库中非常频繁的出现&＃xff0c;很容易就搞混淆了。甚至是multiprocessing自己还有一个Queue类(大写的Q)&＃xff0c;一样能实现queues.Queue的功能&＃xff0c;导入方式是from multiprocessing import Queue。

进程锁

为了防止和多线程一样的出现数据抢夺和脏数据的问题&＃xff0c;同样需要设置进程锁。与threading类似&＃xff0c;在multiprocessing里也有同名的锁类RLock&＃xff0c;Lock&＃xff0c;Event&＃xff0c;Condition和 Semaphore&＃xff0c;连用法都是一样样的&＃xff0c;这一点非常友好&＃xff01;

from multiprocessing import Process
from multiprocessing import Array
from multiprocessing import RLock, Lock, Event, Condition, Semaphore
import timedef func(i,lis,lc):lc.acquire()lis[0] &＃61; lis[0] - 1time.sleep(1)print(&＃39;say hi&＃39;, lis[0])lc.release()if __name__ &＃61;&＃61; "__main__":array &＃61; Array(&＃39;i&＃39;, 1)array[0] &＃61; 10lock &＃61; RLock()for i in range(10):p &＃61; Process(target&＃61;func, args&＃61;(i, array, lock))p.start()


运行结果&＃xff1a;

say hi 9
say hi 8
say hi 7
say hi 6
say hi 5
say hi 4
say hi 3
say hi 2
say hi 1
say hi 0


进程池Pool类

进程启动的开销比较大&＃xff0c;过多的创建新进程会消耗大量的内存空间。仿照线程池的做法&＃xff0c;我们可以使用进程池控制内存开销。

比较幸运的是&＃xff0c;Python给我们内置了一个进程池&＃xff0c;不需要像线程池那样要自己写&＃xff0c;你只需要简单的from multiprocessing import Pool导入就行。进程池内部维护了一个进程序列&＃xff0c;需要时就去进程池中拿取一个进程&＃xff0c;如果进程池序列中没有可供使用的进程&＃xff0c;那么程序就会等待&＃xff0c;直到进程池中有可用进程为止。

进程池中常用的方法&＃xff1a;

• apply() 同步执行&＃xff08;串行&＃xff09;
• apply_async() 异步执行&＃xff08;并行&＃xff09;
• terminate() 立刻关闭进程池
• join() 主进程等待所有子进程执行完毕。必须在close或terminate()之后。
• close() 等待所有进程结束后&＃xff0c;才关闭进程池。

from multiprocessing import Pool
import timedef func(args):time.sleep(1)print("正在执行进程 ", args)if __name__ &＃61;&＃61; &＃39;__main__&＃39;:p &＃61; Pool(5) # 创建一个包含5个进程的进程池for i in range(30):p.apply_async(func&＃61;func, args&＃61;(i,))p.close() # 等子进程执行完毕后关闭进程池# time.sleep(2)# p.terminate() # 立刻关闭进程池p.join()


参考&＃xff1a;

• 多进程multiprocess
• python 进程、线程、协程详解
• Python - Multithreaded Programming