Python3多进程multiprocessing模块详解:Process、Pool、Queue与Pipe用法
本文详细介绍Python3中multiprocessing模块的核心组件,包括Process类的创建与同步、Pool进程池的apply_async与apply方法对比、Queue进程间通信的put/get机制及参数设置,以及Pipe双向通信的实现。通过代码示例解析各组件特性与适用场景,帮助开发者高效实现并发编程。
本文通过具体实例解析Python3中的multiprocessing模块,旨在为读者提供清晰的参考与学习素材。
多进程 Multiprocessing 模块
参考 multiprocessing 模块的官方说明文档。
Process 类
Process 类用于定义进程对象。在创建子进程时,只需传入执行函数及其参数,即可完成 Process 实例的构建。
star() 方法用于启动进程; join() 方法实现进程同步,等待所有子进程结束。 close() 方法用于阻止多余进程进入进程池 Pool,从而避免进程阻塞。
multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)
target 指定要调用的函数名。 args 为函数所需的参数,以 tuple 形式传入。
示例:
import multiprocessing
import os
def run_proc(name):
print('Child process {0} {1} Running '.format(name, os.getpid()))
if __name__ == '__main__':
print('Parent process {0} is Running'.format(os.getpid()))
for i in range(5):
p = multiprocessing.Process(target=run_proc, args=(str(i),))
print('process start')
p.start()
p.join()
print('Process close')
结果:
Parent process 809 is Running process start process start process start process start process start Child process 0 810 Running Child process 1 811 Running Child process 2 812 Running Child process 3 813 Running Child process 4 814 Running Process close
Pool
Pool 提供指定数量的进程供用户使用,默认数量等于 CPU 核数。当有新请求提交到 Pool 时,若进程池未满,则创建新进程执行;若进程池已满,请求将进入等待状态。
- 调用 Pool 对象的 join 方法会等待所有子进程执行完毕。 - 在调用 join 方法之前,必须先调用 close。 - 调用 close 后,不能再向进程池添加新的 Process。
pool.apply_async
apply_async 方法用于异步执行进程,允许多个进程同时进入进程池。
import multiprocessing
import os
import time
def run_task(name):
print('Task {0} pid {1} is running, parent id is {2}'.format(name, os.getpid(), os.getppid()))
time.sleep(1)
print('Task {0} end.'.format(name))
if __name__ == '__main__':
print('current process {0}'.format(os.getpid()))
p = multiprocessing.Pool(processes=3)
for i in range(6):
p.apply_async(run_task, args=(i,))
print('Waiting for all subprocesses done...')
p.close()
p.join()
print('All processes done!')
结果:
current process 921 Waiting for all subprocesses done... Task 0 pid 922 is running, parent id is 921 Task 1 pid 923 is running, parent id is 921 Task 2 pid 924 is running, parent id is 921 Task 0 end. Task 3 pid 922 is running, parent id is 921 Task 1 end. Task 4 pid 923 is running, parent id is 921 Task 2 end. Task 5 pid 924 is running, parent id is 921 Task 3 end. Task 4 end. Task 5 end. All processes done!
pool.apply
apply(func[, args[, kwds]])
该方法仅允许一个进程进入进程池,需等待当前进程结束后,下一个进程方可进入。
import multiprocessing
import os
import time
def run_task(name):
print('Task {0} pid {1} is running, parent id is {2}'.format(name, os.getpid(), os.getppid()))
time.sleep(1)
print('Task {0} end.'.format(name))
if __name__ == '__main__':
print('current process {0}'.format(os.getpid()))
p = multiprocessing.Pool(processes=3)
for i in range(6):
p.apply(run_task, args=(i,))
print('Waiting for all subprocesses done...')
p.close()
p.join()
print('All processes done!')
结果:
Task 0 pid 928 is running, parent id is 927 Task 0 end. Task 1 pid 929 is running, parent id is 927 Task 1 end. Task 2 pid 930 is running, parent id is 927 Task 2 end. Task 3 pid 928 is running, parent id is 927 Task 3 end. Task 4 pid 929 is running, parent id is 927 Task 4 end. Task 5 pid 930 is running, parent id is 927 Task 5 end. Waiting for all subprocesses done... All processes done!
Queue 进程间通信
Queue 用于在多个进程间进行通信。Queue 包含两个主要方法:get 和 put。
put 方法
Put 方法用于向队列插入数据,支持两个可选参数:blocked 和 timeout。 - blocked = True(默认值),timeout 为正数
该方法会阻塞 timeout 指定的时间,直到队列出现剩余空间。若超时,则抛出 Queue.Full 异常。
blocked = False
若 Queue 已满,则立即抛出 Queue.Full 异常。
get 方法
get 方法用于从队列中读取并删除一个元素。支持两个可选参数:blocked 和 timeout。 - blocked = False (默认值),timeout 为正值
在等待时间内若未获取到任何元素,则抛出 Queue.Empty 异常。
blocked = True
若 Queue 中有可用值,则立即返回该值;若 Queue 中没有任何元素,
from multiprocessing import Process, Queue
import os, time, random
# 写数据进程执行的代码:
def proc_write(q,urls):
print('Process(%s) is writing...' % os.getpid())
for url in urls:
q.put(url)
print('Put %s to queue...' % url)
time.sleep(random.random())
# 读数据进程执行的代码:
def proc_read(q):
print('Process(%s) is reading...' % os.getpid())
while True:
url = q.get(True)
print('Get %s from queue.' % url)
if __name__=='__main__':
# 父进程创建Queue,并传给各个子进程:
q = Queue()
proc_writer1 = Process(target=proc_write, args=(q,['url_1', 'url_2', 'url_3']))
proc_writer2 = Process(target=proc_write, args=(q,['url_4','url_5','url_6']))
proc_reader = Process(target=proc_read, args=(q,))
# 启动子进程proc_writer,写入:
proc_writer1.start()
proc_writer2.start()
# 启动子进程proc_reader,读取:
proc_reader.start()
# 等待proc_writer结束:
proc_writer1.join()
proc_writer2.join()
# proc_reader进程里是死循环,无法等待其结束,只能强行终止:
proc_reader.terminate()
结果:
Process(1083) is writing... Put url_1 to queue... Process(1084) is writing... Put url_4 to queue... Process(1085) is reading... Get url_1 from queue. Get url_4 from queue. Put url_5 to queue... Get url_5 from queue. Put url_2 to queue... Get url_2 from queue. Put url_6 to queue... Get url_6 from queue. Put url_3 to queue... Get url_3 from queue.
Pipe 进程间通信
Pipe 通常用于两个进程间的通信,两个进程分别位于管道的两端。
multiprocessing.Pipe([duplex])
示例一和示例二源自网络上的其他案例,本文在尝试理解的基础上增加了注释。网络上的示例通常将示例一和示例二放在一起,此处分开查看更易于理解。
示例一:
from multiprocessing import Process, Pipe
def send(pipe):
pipe.send(['spam'] + [42, 'egg']) # send 传输一个列表
pipe.close()
if __name__ == '__main__':
(con1, con2) = Pipe() # 创建两个 Pipe 实例
sender = Process(target=send, args=(con1, )) # 函数的参数,args 一定是实例化之后的 Pip 变量,不能直接写 args=(Pip(),)
sender.start() # Process 类启动进程
print("con2 got: %s" % con2.recv()) # 管道的另一端 con2 从send收到消息
con2.close() # 关闭管道
结果:
con2 got: ['spam', 42, 'egg']
示例二:
from multiprocessing import Process, Pipe
def talk(pipe):
pipe.send(dict(name='Bob', spam=42)) # 传输一个字典
reply = pipe.recv() # 接收传输的数据
print('talker got:', reply)
if __name__ == '__main__':
(parentEnd, childEnd) = Pipe() # 创建两个 Pipe() 实例,也可以改成 conf1, conf2
child = Process(target=talk, args=(childEnd,)) # 创建一个 Process 进程,名称为 child
child.start() # 启动进程
print('parent got:', parentEnd.recv()) # parentEnd 是一个 Pip() 管道,可以接收 child Process 进程传输的数据
parentEnd.send({x * 2 for x in 'spam'}) # parentEnd 是一个 Pip() 管道,可以使用 send 方法来传输数据
child.join() # 传输的数据被 talk 函数内的 pip 管道接收,并赋值给 reply
print('parent exit')
结果:
parent got: {'name': 'Bob', 'spam': 42} talker got: {'ss', 'aa', 'pp', 'mm'} parent exit
更多关于Python相关内容感兴趣的读者可查看本站专题:《Python进程与线程操作技巧总结》、《Python Socket编程技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》及《Python文件与目录操作技巧汇总》
希望本文所述对大家Python程序设计有所帮助。
本站声明
本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。













