Python multiprocessing模块 -阿里云开发者社区

一、简介

python多线程有个讨厌的限制，全局解释器锁（global interpreter lock），这个锁的意思是任一时间只能有一个线程使用解释器，跟单cpu跑多个程序一个意思，大家都是轮着用的，这叫“并发”，不是“并行”。手册上的解释是为了保证对象模型的正确性！这个锁造成的困扰是如果有一个计算密集型的线程占着cpu，其他的线程都得等着....，试想你的多个线程中有这么一个线程，得多悲剧，多线程生生被搞成串行；当然这个模块也不是毫无用处，手册上又说了：当用于IO密集型任务时，IO期间线程会释放解释器，这样别的线程就有机会使用解释器了！所以是否使用这个模块需要考虑面对的任务类型

python代码执行由python虚拟机（解释器主循环）来控制。对python虚拟机的访问由GIL控制，GIL保证同一时刻只有一个线程在执行。

python虚拟机执行过程：

1、设置GIL

2、切换到一个线程去运行

3、运行，直至完成指定的字节码指令，或者线程主动让出控制

4、将该线程设置为睡眠状态

5、解锁GIL

6、重复以上所有步骤，运行下一个线程　　

由于GIL的限制，python多线程实际只能运行在单核CPU。如要实现多核CPU并行，只能充分地使用多核CPU的资源（os.cpu_count()查看,在python中大部分情况需要使用多进程。大部分并行模块中，多进程相当于开启多个python解释器，每个解释器对应一个进程。也有一些并行模块通过修改pyhton的GIL机制突破这个限制。

Python提供了multiprocessing。 multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。

仔细说来， multiprocess不是一个模块而是python中一个操作、管理进程的包。之所以叫multi是取自multiple的多功能的意思，在这个包中几乎包含了和进程有关的所有子模块。由于提供的子模块非常多，为了方便大家归类记忆，我将这部分大致分为四个部分：创建进程部分，进程同步部分，进程池部分，进程之间数据共享。

二、multiprocessing.Process模块

Process模块是一个创建进程的模块，借助这个模块，就可以完成进程的创建。

与threading.Thread类似，它利用multiprocessing.Process对象来创建一个进程。

该进程可以运行在Python程序内部编写的函数。

该Process对象与Thread对象的用法相同，也有start(), run(), join()的方法。

此外multiprocessing包中也有Lock/Event/Semaphore/Condition类 (这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。

所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

但在使用这些共享API的时候，我们要注意以下几点:

在UNIX平台上，当某个进程终结之后，该进程需要被其父进程调用wait，否则进程成为僵尸进程(Zombie)

所以，有必要对每个Process对象调用join()方法 (实际上等同于wait)。对于多线程来说，由于只有一个进程，所以不存在此必要性。

multiprocessing提供了threading包中没有的IPC(比如Pipe和Queue)，效率上更高。

应优先考虑Pipe和Queue，避免使用Lock/Event/Semaphore/Condition等同步方式 (因为它们占据的不是用户进程的资源)。

多进程应该避免共享资源。在多线程中，我们可以比较容易地共享资源，比如使用全局变量或者传递参数。

在多进程情况下，由于每个进程有自己独立的内存空间，以上方法并不合适。此时我们可以通过共享内存和Manager的方法来共享资源。

但这样做提高了程序的复杂度，并因为同步的需要而降低了程序的效率。　　

使用格式：

multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）

强调：

1. 需要使用关键字的方式来指定参数

2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

参数介绍：

group为预留参数。
target为可调用对象（函数对象），为子进程对应的活动；相当于multiprocessing.Process子类化中重写的run()方法。
name为线程的名称，默认（None）为"Process-N"。
args、kwargs为进程活动（target）的非关键字参数、关键字参数。
deamon为bool值，表示是否为守护进程。

方法介绍：

run()：表示进程运行的方法。可以在子类中重写此方法。标准run() 方法调用传递给对象构造函数的可调用对象作为目标参数（如果有），分别使用args和kwargs参数中的顺序和关键字参数。

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

join([ 超时])：如果可选参数timeout是None，则该方法将阻塞，直到join()调用其方法的进程终止。如果timeout是一个正数，它最多会阻塞超时秒。请注意，如果方法的进程终止或方法超时，则返回该方法。检查进程exitcode以确定它是否终止。

name：进程的名称。该名称是一个字符串，仅用于识别目的。

is_alive()：返回进程是否存活。从start() 方法返回到子进程终止的那一刻，进程对象仍处于活动状态。

daemon：进程的守护进程标志，一个布尔值。必须在start()调用之前设置，当进程退出时，它会尝试终止其所有守护进程子进程。

pid：返回进程ID。在产生该过程之前，这将是 None。

exitcode：子进程的退出代码。None如果流程尚未终止，这将是。负值-N表示孩子被信号N终止。

属性介绍

p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置

p.name:进程的名称

p.pid：进程的pid

p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)

p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）

需要注意的是start()，join()，is_alive()， terminate()和exitcode方法只能由创建进程对象的过程调用。

三 Process类的使用

注意：在windows中Process()必须放到# if __name__ == '__main__':下

详细解释

Since Windows has no fork, the multiprocessing module starts a new Python process and imports the calling module.

If Process() gets called upon import, then this sets off an infinite succession of new processes (or until your machine runs out of resources).

This is the reason for hiding calls to Process() inside

if __name__ == "__main__"

since statements inside this if-statement will not get called upon import.

由于Windows没有fork，多处理模块启动一个新的Python进程并导入调用模块。

如果在导入时调用Process（），那么这将启动无限继承的新进程（或直到机器耗尽资源）。

这是隐藏对Process（）内部调用的原，使用if __name__ == “__main __”，这个if语句中的语句将不会在导入时被调用。

创建并开启子进程的两种方式

方法一：multiprocessing.Process子类化

#开进程的方法一:

import time

import random

from multiprocessing import Process

def piao(name):

print('%s piaoing' %name)

time.sleep(random.randrange(1,5))

print('%s piao end' %name)

p1=Process(target=piao,args=('egon',)) #必须加,号

p1.start()

print('主线程')

方法二：通过函数　

#开进程的方法二:

import time

import random

from multiprocessing import Process

class Piao(Process):

def __init__(self,name):

super().__init__()

self.name=name

def run(self):

print('%s piaoing' %self.name)

time.sleep(random.randrange(1,5))

print('%s piao end' %self.name)

p1=Piao('egon')

p1.start() #start会自动调用run

print('主线程')

一些方法的示例用法：

from multiprocessing import Process

import time

import os

def info():

print('module name:', __name__)

print('parent process:', os.getppid())

print('process id:', os.getpid())

def f(name):

info()

time.sleep(3)

print('hello', name)

if __name__ == '__main__':

info()

p = Process(target=f, args=('bob',))

# p.daemon = False

print(p.daemon)

p.start()

p.join(1)

print('name:', p.name)

print('is_alive:', p.is_alive())

print('exitcode:', p.exitcode)

在上述逻辑中，子进程会休息3s然后再打印一句话才结束，同时设定join(1)阻塞1s，阻塞在1s后结束，我们的并没有守护主进程，然后主进程结束后，子进程依然alive；

如果想要守护主进程，设定p.daemon = True:

if __name__ == '__main__':

info()

p = Process(target=f, args=('bob',))

p.daemon = True

print(p.daemon)

p.start()

# p.join(1)

print('name:', p.name)

print('is_alive:', p.is_alive())

print('exitcode:', p.exitcode)

在上述逻辑中，子进程会休息3s然后再打印一句话才结束，我们的设定守护主进程，然后主进程结束后，打印的is_alive: True这句话其实是在主进程里运行的，所以此时子进程确实是alive，但是主进程结束后子进程也结束了，不会运行info() 函数；　　

进程直接的内存空间是隔离的

from multiprocessing import Process

n=100 #在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就可以了

def work():

global n

n=0

print('子进程内: ',n)

if __name__ == '__main__':

p=Process(target=work)

p.start()

print('主进程内: ',n)

对于要创建多个子进程的情形，更简洁的办法是采用进程池：multiprocessing.Pool(processes=None, initializer=None, initargs=(), maxtasksperchild=None)

processes ：使用的工作进程的数量，如果processes是None那么使用 os.cpu_count()返回的数量。

initializer：如果initializer不是None，那么每一个工作进程在开始的时候会调用initializer(*initargs)。

maxtasksperchild：工作进程退出之前可以完成的任务数，完成后用一个新的工作进程来替代原进程，来让闲置的资源被释放。maxtasksperchild默认是None，意味着只要Pool存在工作进程就会一直存活。

context: 用在制定工作进程启动时的上下文，一般使用 multiprocessing.Pool() 或者一个context对象的Pool()方法来创建一个池，两种方法都适当的设置了context。　　

如，将socket通信变成并发的形式

server.py

from socket import *

from multiprocessing import Pool

server=socket(AF_INET,SOCK_STREAM)

server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)

server.bind(('127.0.0.1',8080))

server.listen(5)

def talk(conn,):

while True:

try:

msg=conn.recv(1024)

if not msg:break

conn.send(msg.upper())

except Exception:

break

if __name__ == '__main__': #windows下start进程一定要写到这下面

pool = Pool(processes=10)

while True:

conn,client_addr=server.accept()

pool.apply_async(talk,args=(conn,))

client.py

from socket import *

client=socket(AF_INET,SOCK_STREAM)

client.connect(('127.0.0.1',8080))

while True:

msg=input('>>: ').strip()

if not msg:continue

client.send(msg.encode('utf-8'))

msg=client.recv(1024)

print(msg.decode('utf-8'))

四、僵尸进程与孤儿进程（了解）

100

101

102

103

104

105

106

107

108

109

110

111

参考博客：http://www.cnblogs.com/Anker/p/3271773.html

一：僵尸进程（有害）

　　僵尸进程：一个进程使用fork创建子进程，如果子进程退出，而父进程并没有调用wait或waitpid获取子进程的状态信息，那么子进程的进程描述符仍然保存在系统中。这种进程称之为僵死进程。详解如下

我们知道在unix/linux中，正常情况下子进程是通过父进程创建的，子进程在创建新的进程。子进程的结束和父进程的运行是一个异步过程,即父进程永远无法预测子进程到底什么时候结束，如果子进程一结束就立刻回收其全部资源，那么在父进程内将无法获取子进程的状态信息。

因此，UNⅨ提供了一种机制可以保证父进程可以在任意时刻获取子进程结束时的状态信息：

1、在每个进程退出的时候，内核释放该进程所有的资源，包括打开的文件，占用的内存等。但是仍然为其保留一定的信息（包括进程号the process ID，退出状态the termination status of the process，运行时间the amount of CPU time taken by the process等）

2、直到父进程通过wait / waitpid来取时才释放. 但这样就导致了问题，如果进程不调用wait / waitpid的话，那么保留的那段信息就不会释放，其进程号就会一直被占用，但是系统所能使用的进程号是有限的，如果大量的产生僵死进程，将因为没有可用的进程号而导致系统不能产生新的进程. 此即为僵尸进程的危害，应当避免。

　　任何一个子进程(init除外)在exit()之后，并非马上就消失掉，而是留下一个称为僵尸进程(Zombie)的数据结构，等待父进程处理。这是每个子进程在结束时都要经过的阶段。如果子进程在exit()之后，父进程没有来得及处理，这时用ps命令就能看到子进程的状态是“Z”。如果父进程能及时处理，可能用ps命令就来不及看到子进程的僵尸状态，但这并不等于子进程不经过僵尸状态。如果父进程在子进程结束之前退出，则子进程将由init接管。init将会以父进程的身份对僵尸状态的子进程进行处理。

二：孤儿进程（无害）

　　孤儿进程：一个父进程退出，而它的一个或多个子进程还在运行，那么那些子进程将成为孤儿进程。孤儿进程将被init进程(进程号为1)所收养，并由init进程对它们完成状态收集工作。

　　孤儿进程是没有父进程的进程，孤儿进程这个重任就落到了init进程身上，init进程就好像是一个民政局，专门负责处理孤儿进程的善后工作。每当出现一个孤儿进程的时候，内核就把孤儿进程的父进程设置为init，而init进程会循环地wait()它的已经退出的子进程。这样，当一个孤儿进程凄凉地结束了其生命周期的时候，init进程就会代表党和政府出面处理它的一切善后工作。因此孤儿进程并不会有什么危害。

我们来测试一下（创建完子进程后，主进程所在的这个脚本就退出了，当父进程先于子进程结束时，子进程会被init收养，成为孤儿进程，而非僵尸进程），文件内容

import os

import sys

import time

pid = os.getpid()

ppid = os.getppid()

print 'im father', 'pid', pid, 'ppid', ppid

pid = os.fork()

#执行pid=os.fork()则会生成一个子进程

#返回值pid有两种值：

# 如果返回的pid值为0，表示在子进程当中

# 如果返回的pid值>0，表示在父进程当中

if pid > 0:

print 'father died..'

sys.exit(0)

# 保证主线程退出完毕

time.sleep(1)

print 'im child', os.getpid(), os.getppid()

执行文件，输出结果：

im father pid 32515 ppid 32015

father died..

im child 32516 1

看，子进程已经被pid为1的init进程接收了，所以僵尸进程在这种情况下是不存在的，存在只有孤儿进程而已，孤儿进程声明周期结束自然会被init来销毁。

三：僵尸进程危害场景：

　　例如有个进程，它定期的产生一个子进程，这个子进程需要做的事情很少，做完它该做的事情之后就退出了，因此这个子进程的生命周期很短，但是，父进程只管生成新的子进程，至于子进程退出之后的事情，则一概不闻不问，这样，系统运行上一段时间之后，系统中就会存在很多的僵死进程，倘若用ps命令查看的话，就会看到很多状态为Z的进程。严格地来说，僵死进程并不是问题的根源，罪魁祸首是产生出大量僵死进程的那个父进程。因此，当我们寻求如何消灭系统中大量的僵死进程时，答案就是把产生大量僵死进程的那个元凶枪毙掉（也就是通过kill发送SIGTERM或者SIGKILL信号啦）。枪毙了元凶进程之后，它产生的僵死进程就变成了孤儿进程，这些孤儿进程会被init进程接管，init进程会wait()这些孤儿进程，释放它们占用的系统进程表中的资源，这样，这些已经僵死的孤儿进程就能瞑目而去了。

四：测试

#1、产生僵尸进程的程序test.py内容如下

#coding:utf-8

from multiprocessing import Process

import time,os

def run():

print('子',os.getpid())

if __name__ == '__main__':

p=Process(target=run)

p.start()

print('主',os.getpid())

time.sleep(1000)

#2、在unix或linux系统上执行

[root@vm172-31-0-19 ~]# python3 test.py &

[1] 18652

[root@vm172-31-0-19 ~]# 主 18652

子 18653

[root@vm172-31-0-19 ~]# ps aux |grep Z

USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND

root 18653 0.0 0.0 0 0 pts/0 Z 20:02 0:00 [python3] <defunct> #出现僵尸进程

root 18656 0.0 0.0 112648 952 pts/0 S+ 20:02 0:00 grep --color=auto Z

[root@vm172-31-0-19 ~]# top #执行top命令发现1zombie

top - 20:03:42 up 31 min, 3 users, load average: 0.01, 0.06, 0.12

Tasks: 93 total, 2 running, 90 sleeping, 0 stopped, 1 zombie

%Cpu(s): 0.0 us, 0.3 sy, 0.0 ni, 99.7 id, 0.0 wa, 0.0 hi, 0.0 si, 0.0 st

KiB Mem : 1016884 total, 97184 free, 70848 used, 848852 buff/cache

KiB Swap: 0 total, 0 free, 0 used. 782540 avail Mem

PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND

root 20 0 29788 1256 988 S 0.3 0.1 0:01.50 elfin

#3、

等待父进程正常结束后会调用wait／waitpid去回收僵尸进程

但如果父进程是一个死循环，永远不会结束，那么该僵尸进程就会一直存在，僵尸进程过多，就是有害的

解决方法一：杀死父进程

解决方法二：对开启的子进程应该记得使用join，join会回收僵尸进程

参考python2源码注释

class Process(object):

def join(self, timeout=None):

'''

Wait until child process terminates

'''

assert self._parent_pid == os.getpid(), 'can only join a child process'

assert self._popen is not None, 'can only join a started process'

res = self._popen.wait(timeout)

if res is not None:

_current_process._children.discard(self)

join方法中调用了wait，告诉系统释放僵尸进程。discard为从自己的children中剔除

解决方法三：http://blog.csdn.net/u010571844/article/details/50419798

Python multiprocessing模块

一、简介

二、multiprocessing.Process模块

三 Process类的使用

四、僵尸进程与孤儿进程（了解）

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Python multiprocessing模块

一、简介

二、multiprocessing.Process模块

三 Process类的使用

四、僵尸进程与孤儿进程（了解）

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像