篇首语:本文由编程笔记#小编为大家整理,主要介绍了Python多进程编程相关的知识,希望对你有一定的参考价值。
一 多进程编程
Python实现多进程的方式有两种:一种方法是os模块中的fork方法,另一种是使用multiprocessing模块。
前者仅适用于LINUX/UNIX操作系统,对Windows不支持,后者则是跨平台的实现方式。
第一种方式:使用os模块中的fork方式实现多进程
import os
if __name__ == \'__main__\':
print \'current Process (%s) start ...\'%(os.getpid())
pid = os.fork()
if pid < 0:
print \'error in fork\'
elif pid == 0:
print \'I am child process(%s) and my parent process is (%s)\',(os.getpid(),os.getppid())
else:
print \'I(%s) created a chlid process (%s).\',(os.getpid(),pid)
第二种方式:multiprocessing
由于GIL的存在,python中的多线程其实并不是真正的多线程,如果想要充分地使用多核CPU的资源,在python中大部分情况需要使用多进程。Python提供了非常好用的多进程包multiprocessing,只需要定义一个函数,Python会完成其他所有事情。借助这个包,可以轻松完成从单进程到并发执行的转换。multiprocessing支持子进程、通信和共享数据、执行不同形式的同步,提供了Process、Queue、Pipe、Lock等组件。
Process.PID中保存有PID,如果进程还没有start(),则PID为None。
window系统下,需要注意的是要想启动一个子进程,必须加上那句if __name__ == "main",进程相关的要写在这句下面。
创建进程的类:Process([group [, target [, name [, args [, kwargs]]]]]),target表示调用对象,args表示调用对象的位置参数元组。kwargs表示调用对象的字典。name为别名。group实质上不使用。
方法:is_alive()、join([timeout])、run()、start()、terminate()。其中,Process以start()启动某个进程。join()方法实现进程间的同步。
#__author: greg
#date: 2017/9/19 23:52
from multiprocessing import Process
import time
def f(name):
time.sleep(1)
print(\'hello\', name,time.ctime())
if __name__ == \'__main__\':
p_list=[]
for i in range(3):
p = Process(target=f, args=(\'alvin\',))
p_list.append(p)
p.start()
for i in p_list:
i.join()
print(\'end\')#一个主进程,三个子进程
# output:
# hello alvin Fri Nov 24 19:10:08 2017
# hello alvin Fri Nov 24 19:10:08 2017
# hello alvin Fri Nov 24 19:10:08 2017
# end
类式调用:
#__author: greg
#date: 2017/9/21 20:02
from multiprocessing import Process
import time
class MyProcess(Process):
def __init__(self):
super(MyProcess, self).__init__()
#self.name = name
def run(self):
time.sleep(1)
print (\'hello\', self.name,time.ctime())
if __name__ == \'__main__\':
p_list=[]
for i in range(3):
p = MyProcess()
p.start()
p_list.append(p)
for p in p_list:
p.join()
print(\'end\')
#output:
# hello MyProcess-1 Fri Nov 24 19:12:17 2017
# hello MyProcess-2 Fri Nov 24 19:12:17 2017
# hello MyProcess-3 Fri Nov 24 19:12:17 2017
# end
显示进程ID号:
#__author: greg
#date: 2017/9/21 20:16
from multiprocessing import Process
import os
import time
def info(title):
print(title)
print(\'module name:\', __name__)
print(\'parent process:\', os.getppid())#父进程号
print(\'process id:\', os.getpid())#进程号
def f(name):
info(\'\\033[31;1mfunction f\\033[0m\')
print(\'hello\', name)
if __name__ == \'__main__\':
info(\'\\033[32;1mmain process line\\033[0m\')
time.sleep(10)
p = Process(target=info, args=(\'bob\',))
p.start()
p.join()
#output:
# main process line
# module name: __main__
# parent process: 1548 pycharm的进程号
# process id: 8416 Python进程号
# bob
# module name: __mp_main__
# parent process: 8416 Python进程号
# process id: 5556 info进程号
构造方法:
Process([group [, target [, name [, args [, kwargs]]]]])
group: 线程组,目前还没有实现,库引用中提示必须是None;
target: 要执行的方法;
name: 进程名;
args/kwargs: 要传入方法的参数。
实例方法:
is_alive():返回进程是否在运行。
join([timeout]):阻塞当前上下文环境的进程程,直到调用此方法的进程终止或到达指定的timeout(可选参数)。
start():进程准备就绪,等待CPU调度
run():strat()调用run方法,如果实例进程时未制定传入target,这star执行t默认run()方法。
terminate():不管任务是否完成,立即停止工作进程
属性:
authkey
daemon:和线程的setDeamon功能一样
exitcode(进程在运行时为None、如果为–N,表示被信号N结束)
name:进程名字。
pid:进程号。
三 进程间通讯
不同进程间内存是不共享的,要想实现两个进程间的数据交换,可以用以下方法:
Queues 用来在多个进程间通信:
1. 阻塞模式
import queue
import time
q = queue.Queue(10) #创建一个队列
start=time.time()
for i in range(10):
q.put(\'A\')
time.sleep(0.5)
end=time.time()
print(end-start)
这是一段极其简单的代码(另有两个线程也在操作队列q),我期望每隔0.5秒写一个\'A\'到队列中,但总是不能如愿:
间隔时间有时会远远超过0.5秒。
原来,Queue.put()默认有 block = True 和 timeout两个参数。
源码:def put(self, item, block=True, timeout=None):
当 block = True 时,写入是阻塞式的,阻塞时间由 timeout确定。
当队列q被(其他线程)写满后,这段代码就会阻塞,直至其他线程取走数据。
Queue.put()方法加上 block=False 的参数,即可解决这个隐蔽的问题。
但要注意,非阻塞方式写队列,当队列满时会抛出 exception Queue.Full 的异常。
#__author: greg
#date: 2017/9/21 22:27
from multiprocessing import Process, Queue
def f(q,n):
q.put([42, n, \'hello\'])
print(\'subprocess id\',id(q))
if __name__ == \'__main__\':
q = Queue()
p_list=[]
print(\'process id\',id(q))
for i in range(3):
p = Process(target=f, args=(q,i))
p_list.append(p)
p.start()
print(q.get())
print(q.get())
print(q.get())
for i in p_list:
i.join()
# output
# process id 2284856854176
# subprocess id 2607348001872
# [42, 0, \'hello\']
# subprocess id 1712786975824
# [42, 2, \'hello\']
# subprocess id 2254764977120
# [42, 1, \'hello\']
Pipe常用来两个进程间进行通信,两个进程分别位于管道的两端
def f(conn):
conn.send([42, None, \'hello\'])
conn.close()
if __name__ == \'__main__\':
parent_conn, child_conn = Pipe()
p = Process(target=f, args=(child_conn,))
p.start()
print(parent_conn.recv()) # prints "[42, None, \'hello\']"
p.join()
Pipe方法返回(conn1, conn2)代表一个管道的两个端。Pipe方法有duplex参数,如果duplex参数为True(默认值),那么这个管道是全双工模式,也就是说conn1和conn2均可收发。duplex为False,conn1只负责接受消息,conn2只负责发送消息。
#__author: greg
#date: 2017/9/21 22:57
import multiprocessing
import random
import time,os
def proc_send(pipe,urls):
for url in urls:
print("Process(%s) send: %s" %(os.getpid(),url))
pipe.send(url)
time.sleep(random.random())
def proc_recv(pipe):
while True:
print("Process(%s) rev:%s" %(os.getpid(),pipe.recv()))
time.sleep(random.random())
if __name__=="__main__":
pipe=multiprocessing.Pipe()
p1=multiprocessing.Process(target=proc_send,args=(pipe[0],[\'url_\'+str(i)
for i in range(10)]))
p2=multiprocessing.Process(target=proc_recv,args=(pipe[1],))
p1.start()
p2.start()
p1.join()
p2.terminate()
Manager()返回的管理器对象控制一个服务器进程,该进程持有Python对象,并允许其他进程使用代理来操纵它们。
#__author: greg
#date: 2017/9/21 23:10
from multiprocessing import Process, Manager
def f(d, l,n):
d[n] = \'1\'
d[\'2\'] = 2
d[0.25] = None
l.append(n)
# print(l)
if __name__ == \'__main__\':
with Manager() as manager:
d = manager.dict()
l = manager.list(range(5))
p_list = []
for i in range(10):
p = Process(target=f, args=(d, l,i))
p.start()
p_list.append(p)
for res in p_list:
res.join()
print(d)
print(l)
四 进程同步
当多个进程需要访问共享资源的时候,Lock可以用来避免访问的冲突。
#__author: greg
#date: 2017/9/21 23:25
from multiprocessing import Process, Lock
def f(l, i):
l.acquire()
try:
print(\'hello world\', i)
finally:
l.release()
if __name__ == \'__main__\':
lock = Lock()
for num in range(10):
Process(target=f, args=(lock, num)).start()
五 进程池 Pool类
Pool可以提供指定数量的进程供用户使用,默认大小是CPU的核数。当有新的请求提交到Pool中时,如果池还没有满,那么就会创建一个新的进程来执行该请求
但如果池中的进程数已经达到规定的最大值,那么该请求就会等待,直到池中有进程结束,才会创建新的进程来处理它。
# -*- coding: utf-8 -*-
# 2017/11/24 20:15
from multiprocessing import Pool
import os, time, random
def run_task(name):
print(\'Task %s (pid = %s) is running...\' % (name, os.getpid()))
time.sleep(random.random() * 3)
print(\'Task %s end.\' % name)
if __name__==\'__main__\':
print(\'Current process %s.\' % os.getpid())
p = Pool(processes=3)
for i in range(5):
p.apply_async(run_task, args=(i,))
print(\'Waiting for all subprocesses done...\')
p.close()
p.join()
print(\'All subprocesses done.\')
"""
Current process 9788.
Waiting for all subprocesses done...
Task 0 (pid = 5916) is running...
Task 1 (pid = 3740) is running...
Task 2 (pid = 6964) is running...
Task 2 end.
Task 3 (pid = 6964) is running...
Task 1 end.
Task 4 (pid = 3740) is running...
Task 0 end.
Task 3 end.
Task 4 end.
All subprocesses done.
"""
每次最多运行3个进程,当一个任务结束了,新的任务依次添加进来,任务执行使用的进程依然是原来的进程,这一点通过进程的pid可以看出来。