[Python]-14-多进程

引言

在硬件条件允许的情况下,使用多进程编程可以提高程序的执行效率,这篇文章将介绍如何让Python程序实现多进程。

文章目录

  • 0×1.Unix/Linux系统与Fork()
  • 0×2.使用multiprocessing模块实现多进程
  • a.使用Process类创建进程对象
  • c.使用Queue实现进程间通信

0×1.Unix/Linux系统与Fork()

注:这一部分所介绍的方法,仅适用于Unix/Linux操作系统,并不适用于Windows系统,python在windows系统中的多进程实现可以使用跨平台的multiprocessing模块,请参考本文第二部分的内容;本文所有程序都是在Ubuntu系统环境下执行的,部分代码可能无法在Windows下运行。

说到多进程,就不得不介绍Unix/Linux系统中的一个经典系统调用fork(),当fork被程序调用时,会在程序调用的位置,记录下调用程序当前的所有状态信息(变量值等),再创建一个子进程,将当前程序(父进程)的状态信息传送给这个子进程,这个新建的子进程会复制父进程fork后面的所有程序过程,继续执行下面的内容,而此时,父进程本身也会继续执行fork下面的内容,原本一个进程执行的程序,在fork后变成了两个进程在执行。

Python的os模块封装中就包括了Unix/Linux系统的fork调用,下面用一个小程序来演示一下上面这些理论:

#!/usr/bin/env python3
#coding=utf-8
import os
x=0
s="www.qingsword.com"
rpid=os.fork()
if rpid<0:
print("fork调用失败。")
elif rpid==0:
print("我是子进程(%s),我的父进程是(%s)"%(os.getpid(),os.getppid()))
x+=1
else:
print("我是父进程(%s),我的子进程是(%s)"%(os.getpid(),rpid))
x+=1
print(s,x)

#程序输出
我是父进程(7498),我的子进程是(7499)
www.qingsword.com 1
我是子进程(7499),我的父进程是(7498)
www.qingsword.com 1

从程序输出中不难发现,我们仅执行了一次程序,却同时运行了if判断的elif和else分之的内容,这就是fork的作用;

在rpid=os.fork()这条语句执行之前,程序只有一个进程在执行,此时程序已经初始化了两个变量,当执行fork语句后,程序创建了一个子进程,并且将这两个变量的状态传递给了这个子进程,子进程从if开始执行,直到执行完程序剩下的代码;

rpid=os.fork()之后,父进程也会继续执行,也就意味着,两个进程同时从if的位置开始执行剩下的代码,互不干扰;

os.getpid()能够获取当前进程的PID,os.getppid()能够获取当前进程的父进程PID;

那么,我们是如何知到rpid为0,就是子进程的呢?这是因为fork有三种不同的返回值:

1)在父进程中,fork返回新创建子进程的进程ID;

2)在子进程中,fork返回0;

3)如果出现错误,fork返回一个负值;

这样,我们就可以通过os.fork()这条语句执行后的返回值,来判断进程是父进程还是子进程,在子进程中,返回值是0,在父进程中,返回值是刚才新建的子进程的PID,如果创建进程失败,就会返回一个负值(通常如果系统限制了进程创建,或内存不足,或达到了系统进程上限,就会创建失败并返回一个负值);

在创建了新的进程后,这两个进程(父进程与子进程)执行没有固定的先后顺序,哪个进程先执行要看系统的进程调度策略;

其次,fork后,fork前已经初始化的那些变量会被复制成两份,两个进程中x的值都是0,并且两个进程的变量是独立的,存在不同的地址中,这就是为什么最后打印出来的x值是1的原因(每个进程的if流程都只执行了一次x+=1);

再来看一个实例:

#!/usr/bin/env python3
#coding=utf-8
import os
x=1
while x<3:
rpid=os.fork()
if rpid==0:
print("第%d次循环,我是子进程(%s),我的父进程是(%s),rpid值为(%s)"					              %(x,os.getpid(),os.getppid(),rpid))
else:
print("第%d次循环,我是父进程(%s),我的子进程是(%s),rpid值为(%s)"					                      %(x,os.getpid(),rpid,rpid))
x+=1

#程序输出
第1次循环,我是父进程(8790),我的子进程是(8791),rpid值为(8791)
第2次循环,我是父进程(8790),我的子进程是(8792),rpid值为(8792)
第1次循环,我是子进程(8791),我的父进程是(8790),rpid值为(0)
第2次循环,我是父进程(8791),我的子进程是(8793),rpid值为(8793)
第2次循环,我是子进程(8792),我的父进程是(8790),rpid值为(0)
第2次循环,我是子进程(8793),我的父进程是(8791),rpid值为(0)

第一次循环的时,当前进程(p8790)fork出了一个新的子进程(p8791),之后,两个进程分别先后进入了第二次循环,父进程(p8790)又fork出了一个新的子进程(p8792),而第一次循环中被创建的子进程(p8791)在第二次循环中也fork出了一个子进程(p8793),成了这个子进程的父进程。

大家可以尝试着增加循环次数,最后可以得到一个结论,循环n次所创建的子进程数量为:下标为k=1,上标为n的∑2^(k-1)

0×2.使用multiprocessing模块实现多进程

第一部分中所使用的fork仅仅是为了让大家了解什么是多进程,以及它的基本工作原理,这一部分将介绍一个跨平台的多进程模块"multiprocessing"。

a.使用Process类创建进程对象

Process是"multiprocessing"模块包含的一个类,可用于创建多进程程序,并可以指定创建的子进程对象去完成哪些代码块的执行。

Process语法结构如下:

Process([group [, target [, name [, args [, kwargs]]]]])

● target:表示这个进程实例所调用对象;

● args:表示调用对象的位置参数元组;

● kwargs:表示调用对象的关键字参数字典;

● name:为当前进程实例的别名;

● group:大多数情况下用不到;

Process类常用方法:

● is_alive():判断进程实例是否还在执行;

● join([timeout]):是否等待进程实例执行结束,或等待多少秒;

● start():启动进程实例(创建子进程);

● run():如果没有给定target参数,对这个对象调用start()方法时,就将执行对象中的run()方法;

● terminate():不管任务是否完成,立即终止;

Process类常用属性:

● name:当前进程实例别名,默认为Process-N,N为从1开始递增的整数;

● pid:当前进程实例的PID值;

● daemon:如果设置为True,父进程终止后子进程将自动终止,且自己不能产生新进程,必须在子进程实例start()之前设置;

下面用一个实例演示一下上面这些概念:

#!/usr/bin/env python3
#coding=utf-8
from multiprocessing import Process
import time
import os

#两个子进程将会调用的两个方法
def  worker_1(interval):
print("worker_1,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid()))
t_start=time.time()
time.sleep(interval) #程序将会被挂起interval秒
t_end=time.time()
print("worker_1,执行时间为%0.2f秒"%(t_end-t_start))

def  worker_2(interval):
print("worker_2,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid()))
t_start=time.time()
time.sleep(interval)
t_end=time.time()
print("worker_2,执行时间为%0.2f秒"%(t_end-t_start))

#输出当前程序的ID
print("进程ID:%s"%os.getpid())

#创建两个进程对象,target指向这个进程对象要执行的对象名称,args后面的元组中,是要传递给worker_1方法的参数,因为worker_1方法就一个interval参数,这里传递一个整数2给它,如果不指定name参数,默认的进程对象名称为Process-N,N为一个递增的整数
p1=Process(target=worker_1,args=(2,))
p2=Process(target=worker_2,name="qing2",args=(1,))

#使用"进程对象名称.start()"来创建并执行一个子进程,这两个进程对象在start后,就会分别去执行worker_1和worker_2方法中的内容
p1.start()
p2.start()

#同时父进程仍然往下执行,如果p2进程还在执行,将会返回True
print("p2.is_alive=%s"%p2.is_alive())

#输出p1和p2进程的别名和pid
print("p1.name=%s"%p1.name)
print("p1.pid=%s"%p1.pid)
print("p2.name=%s"%p2.name)
print("p2.pid=%s"%p2.pid)

#join括号中不携带参数,表示父进程在这个位置要等待p1进程执行完成后,再继续执行下面的语句,一般用于进程间的数据同步,如果不写这一句,下面的is_alive判断将会是True,在WingIDE中也可能看不到worker_1执行完成后的print输出,因为主程序很可能先执行完成,但在shell(cmd)里面调用这个程序时可以完整的看到这个过程,大家可以尝试着将下面的这条语句改成p1.join(1),因为p2需要2秒以上才可能执行完成,父进程等待1秒很可能不能让p1完全执行完成,所以下面的print会输出True,即p1仍然在执行
p1.join()
print("p1.is_alive=%s"%p1.is_alive())

#程序输出
进程ID:7609
p2.is_alive=True
p1.name=Process-1 #p1使用了默认的别名
p1.pid=7610
p2.name=qing2
p2.pid=7611
worker_1,父进程(7609),当前进程(7610)
worker_2,父进程(7609),当前进程(7611)
worker_2,执行时间为1.00秒
#如果不使用join,在WingIDE中很可能看不到这一句输出,并且下面的is_alive会显示True
worker_1,执行时间为2.00秒
p1.is_alive=False

再来看一个实例,下面这个程序,使用一个for循环获取当前程序正在运行的子进程列表:

#!/usr/bin/env python3
#coding=utf-8
import multiprocessing
from multiprocessing import Process
import time
import os

def  worker_1(interval):
print("worker_1,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid()))
t_start=time.time()
time.sleep(interval)
t_end=time.time()
print("worker_1,执行时间为%0.2f秒"%(t_end-t_start))

print("进程ID:%s"%os.getpid())
#获取当前计算机CPU计算核心个数
print("当前计算机CPU计算核心个数:",multiprocessing.cpu_count())

#初始化4个进程对象,p1仅延迟1秒,前三个延迟3秒
p1=Process(target=worker_1,args=(3,))
p2=Process(target=worker_1,args=(3,))
p3=Process(target=worker_1,args=(3,))
p4=Process(target=worker_1,args=(1,))
p1.start()
p2.start()
p3.start()
p4.start()
p4.join() #等待p4执行完成后再往下执行

#因为p4已经执行完成,active_children()列表中只有前三个子进程
for ps in multiprocessing.active_children():
print(ps.name,ps.pid)
p1.join()
p2.join()
p3.join()
print("END")

#程序输出
进程ID:8327
当前计算机CPU计算核心个数: 4
worker_1,父进程(8327),当前进程(8328)
worker_1,父进程(8327),当前进程(8329)
worker_1,父进程(8327),当前进程(8330)
worker_1,父进程(8327),当前进程(8331)
worker_1,执行时间为1.00秒
Process-3 8330
Process-1 8328
Process-2 8329
worker_1,执行时间为3.00秒
worker_1,执行时间为3.00秒
worker_1,执行时间为3.00秒
END

下面的代码演示了Process类的deamon属性的用法:

#!/usr/bin/env python3
#coding=utf-8
from multiprocessing import Process
import time
import os

def  worker_1(interval):
print("worker_1,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid()))
t_start=time.time()
time.sleep(interval)
t_end=time.time()
print("worker_1,执行时间为%0.2f秒"%(t_end-t_start))

if __name__=="__main__":
t_start=time.time()
print("当前程序进程(%s)"%os.getpid())
p1=Process(target=worker_1,args=(2,))
p1.daemon=True
#设置成True,只要父进程运行结束,子进程也会随即结束
p1.start()
#p1.join() 添加这个方法就能让父进程等待子进程执行完成
t_stop=time.time()
print("(%s)执行结束,耗时%0.2f"%(os.getpid(),t_stop-t_start))

#程序输出,从输出中可以看到,因为父进程短时间就执行结束了,所以子进程都还没来得及运行
当前程序进程(13921)
(13921)执行结束,耗时0.01

如果想要设置daemon参数为True,又想在父进程结束前执行完子进程,可以在主进程中加入join方法,等待子进程执行完成。

现在我们已经可以任意的创建子进程去执行指定目标代码块了,实际上,创建新的进程还能够使用类的方式,可以自定义一个类,继承Process类,每次实例化这个类的时候,就等同于实例化一个进程对象,请看下面的实例:

#!/usr/bin/env python3
#coding=utf-8
from multiprocessing import Process
import time
import os

#继承Process类
class Process_Class(Process):
#因为Process类本身也有__init__方法,这个子类相当于重写了这个方法,但这样就会带来一个问题,我们并没有完全的初始化一个Process类,所以就不能使用从这个类继承的一些方法和属性,最好的方法就是将继承类本身传递给Process.__init__方法,完成这些初始化操作
def __init__(self,interval):
Process.__init__(self)
self.interval=interval

#重写了Process类的run()方法
def run(self):
print("子进程(%s) 开始执行,父进程为(%s)"%(os.getpid(),os.getppid()))
t_start=time.time()
time.sleep(self.interval)
t_stop=time.time()
print("(%s)执行结束,耗时%0.2f秒"%(os.getpid(),t_stop-t_start))

if __name__=="__main__":
t_start=time.time()
print("当前程序进程(%s)"%os.getpid())
p1=Process_Class(2)
#对一个不包含target属性的Process类执行start()方法,就会运行这个类中的run()方法,所以这里会执行p1.run()
p1.start()
p1.join()
t_stop=time.time()
print("(%s)执行结束,耗时%0.2f"%(os.getpid(),t_stop-t_start))

#程序输出
当前程序进程(12153)
子进程(12154) 开始执行,父进程为(12153)
(12154)执行结束,耗时2.00秒
(12153)执行结束,耗时2.01

b.使用进程池Pool

当需要创建的子进程数量不多时,可以直接利用multiprocessing中的Process动态成生多个进程,但如果是上百甚至上千个目标,手动的去创建进程的工作量巨大,此时就可以用到multiprocessing模块提供的Pool方法。

初始化Pool时,可以指定一个最大进程数,当有新的请求提交到Pool中时,如果池还没有满,那么就会创建一个新的进程用来执行该请求;但如果池中的进程数已经达到指定的最大值,那么该请求就会等待,直到池中有进程结束,才会创建新的进程来执行,请看下面的实例:

#!/usr/bin/env python3
#coding=utf-8
from multiprocessing import Pool
import os,time,random

def worker(msg):
t_start=time.time()
print(msg,"开始执行")
#random.random()随机生成0~1之间的浮点数
time.sleep(random.random()*2)
t_stop=time.time()
print(msg,"执行完毕,耗时%0.2f"%(t_stop-t_start))

po=Pool(3) #定义一个进程池,最大进程数3
for i in range(0,5):
#Pool.apply_async(要调用的目标,(传递给目标的参数元祖,)),每次循环将会创建一个新的子进程去调用目标,直到达到进程池最大进程数的限制
po.apply_async(worker,(i,))
print("www.qingsword.com")
po.close() #关闭进程池,关闭后po不再接收新的请求
po.join() #等待po中所有子进程执行完成,必须放在close语句之后
print("End")

#程序输出
www.qingsword.com
0 开始执行
1 开始执行
2 开始执行  #因为Pool(3)所以并行运行了三个子进程
2 执行完毕,耗时0.96
3 开始执行  #直到上面三个子进程中的一个运行完成,再创建新的子进程
3 执行完毕,耗时0.08
4 开始执行
0 执行完毕,耗时1.61
4 执行完毕,耗时0.67
1 执行完毕,耗时1.86
End

multiprocessing.Pool常用函数解析:

● apply_async(func[, args[, kwds[, callback]]]) :使用非阻塞方式调用func(并行执行,堵塞方式必须等待上一个进程退出才能执行下一个进程),args为传递给func的参数列表,kwds为传递给func的关键字参数列表;

● apply(func[, args[, kwds]]):使用阻塞方式调用func(稍后会用实例演示,看输出就能明白两者区别);

● close():关闭Pool,使其不再接受新的任务;

● terminate():不管任务是否完成,立即终止;

● join():主进程阻塞,等待子进程的退出, 必须在close或terminate之后使用;

下面来看一个阻塞方式调用的实例:

#!/usr/bin/env python3
#coding=utf-8
from multiprocessing import Pool
import os,time,random

def worker(msg):
t_start=time.time()
print(msg,"开始执行")
time.sleep(random.random()*2)
t_stop=time.time()
print(msg,"执行完毕,耗时%0.2f"%(t_stop-t_start))

po=Pool(3)
for i in range(0,5):
po.apply(worker,(i,)) #唯一的区别在这里
print("www.qingsword.com")
po.close()
po.join()
print("End")

#程序输出,从输出中可以看出,每个子进程执行完成后再创建下一个子进程,父进程也被阻塞,必须等待所有子进程调用完成后,再继续执行下面的内容
0 开始执行
0 执行完毕,耗时0.27
1 开始执行
1 执行完毕,耗时0.06
2 开始执行
2 执行完毕,耗时1.19
3 开始执行
3 执行完毕,耗时1.36
4 开始执行
4 执行完毕,耗时0.70
www.qingsword.com
End

c.使用Queue实现进程间通信

可以使用multiprocessing模块的Queue实现多进程之间的数据传递,Queue本身是一个消息列队程序,首先用一个小实例来演示一下Queue的工作原理:

#!/usr/bin/env python3
#coding=utf-8
from multiprocessing import Queue
q=Queue(3) #初始化一个Queue对象,最多可接收三条put消息
q.put("消息1")
q.put("消息2")
print(q.full())  #False
q.put("消息3")
print(q.full()) #True

#因为消息列队已满下面的try都会抛出异常,第一个try会等待2秒后再抛出异常,第二个Try会立刻抛出异常
try:
q.put("消息4",True,2)
except:
print("消息列队已满,现有消息数量:%s"%q.qsize())

try:
q.put_nowait("消息4")
except:
print("消息列队已满,现有消息数量:%s"%q.qsize())

#推荐的方式,先判断消息列队是否已满,再写入
if not q.full():
q.put_nowait("消息4")

#读取消息时,先判断消息列队是否为空,再读取
if not q.empty():
for i in range(q.qsize()):
print(q.get_nowait())

#程序输出
False
True
消息列队已满,现有消息数量:3
消息列队已满,现有消息数量:3
消息1
消息2
消息3

下面是Queue对象中所包含的常用方法的功能简介:

初始化Queue()对象时(例如:q=Queue()),若括号中没有指定最大可接收的消息数量,或数量为负值,那么就代表可接受的消息数量没有上限(直到内存的尽头);

● Queue.qsize():返回当前队列包含的消息数量;

● Queue.empty():如果队列为空,返回True,反之False ;

● Queue.full():如果队列满了,返回True,反之False;

● Queue.get([block[, timeout]]):获取队列中的一条消息,然后将其从列队中移除,block默认值为True;

1)如果block使用默认值,且没有设置timeout(单位秒),消息列队如果为空,此时程序将被阻塞(停在读取状态),直到从消息列队读到消息为止,如果设置了timeout,则会等待timeout秒,若还没读取到任何消息,则抛出"Queue.Empty"异常;

2)如果block值为False,消息列队如果为空,则会立刻抛出"Queue.Empty"异常;

● Queue.get_nowait():相当Queue.get(False);

● Queue.put(item,[block[, timeout]]):将item消息写入队列,block默认值为True;

1)如果block使用默认值,且没有设置timeout(单位秒),消息列队如果已经没有空间可写入,此时程序将被阻塞(停在写入状态),直到从消息列队腾出空间为止,如果设置了timeout,则会等待timeout秒,若还没空间,则抛出"Queue.Full"异常;

2)如果block值为False,消息列队如果没有空间可写入,则会立刻抛出"Queue.Full"异常;

● Queue.put_nowait(item):相当Queue.put(item, False);

在了解了这些基本原理后,来看看Queue如何实现不同进程间数据的传递:

#!/usr/bin/env python3
#coding=utf-8
from multiprocessing import Queue,Process
import os,time,random

#用于读取列队中的消息打印到屏幕
def reader(q):
print("reader启动(%s),父进程为(%s)"%(os.getpid(),os.getppid()))
#一个死循环,结束时需要使用terminate()方法强行结束,q.get(True)中设置了block值为True,并且没有指定等待时间
while True:
print("reader从Queue获取到消息:%s"%q.get(True))

#用于向列队中put消息
def writer(q):
print("writer启动(%s),父进程为(%s)"%(os.getpid(),os.getppid()))
L=(x for x in "qing")
for i in L:
q.put(i)
time.sleep(random.random()*3)
#设置一个延迟来模拟进程put消息的时间差

if __name__=="__main__":
print("(%s) start"%os.getpid())
q=Queue() #初始化一个Queue对象没有指定消息列队上限
#将q对象作为参数传递给下面两个进程
pr=Process(target=reader,args=(q,))
pw=Process(target=writer,args=(q,))
pr.start()
pw.start()
pw.join()
pr.terminate() #因为pr不能自己终止,所以只能强行终止
print("(%s) End"%os.getpid())

#程序输出
(11592) start
reader启动(11593),父进程为(11592)
writer启动(11594),父进程为(11592)
reader从Queue获取到消息:q
reader从Queue获取到消息:i
reader从Queue获取到消息:n
reader从Queue获取到消息:g
(11592) End

上面这种方法只适用于普通的Process进程创建,如果要使用Pool创建进程,就需要使用multiprocessing.Manager()中的Queue(),而不是multiprocessing.Queue(),否则会得到一条如下的错误信息:

RuntimeError: Queue objects should only be shared between processes through inheritance.

下面的实例演示了进程池中的进程如何通信:

#!/usr/bin/env python3
#coding=utf-8

#修改import中的Queue为Manager
from multiprocessing import Manager,Pool
import os,time,random

def reader(q):
print("reader启动(%s),父进程为(%s)"%(os.getpid(),os.getppid()))
for i in range(q.qsize()):
print("reader从Queue获取到消息:%s"%q.get(True))

def writer(q):
print("writer启动(%s),父进程为(%s)"%(os.getpid(),os.getppid()))
L=(x for x in "qing")
for i in L:
q.put(i)

if __name__=="__main__":
print("(%s) start"%os.getpid())
q=Manager().Queue() #使用Manager中的Queue来初始化
po=Pool()
#使用阻塞模式创建进程,这样就不需要在reader中使用死循环了,可以让writer完全执行完成后,再用reader去读取
po.apply(writer,(q,))
po.apply(reader,(q,))
po.close()
po.join()
print("(%s) End"%os.getpid())

#程序输出
(11971) start
writer启动(11978),父进程为(11971)
reader启动(11977),父进程为(11971)
reader从Queue获取到消息:q
reader从Queue获取到消息:i
reader从Queue获取到消息:n
reader从Queue获取到消息:g
(11971) End
  • 版权声明: 本文源自互联网, 于3个月前,由整理发表,共 677字。
  • 原文链接:点此查看原文