Python多进程编程实战：从Process到Pool详解
本文深入解析Python中多进程编程的核心模块multiprocessing。对比多线程与协程在CPU密集型场景的局限，详细演示如何通过Process类创建单个或批量子进程，并重点讲解利用Pool进程池实现异步非阻塞任务调用的最佳实践，包含apply_async与get方法获取执行结果的具体代码示例。
在Python中，多线程主要适用于IO密集型任务。面对CPU密集型场景时，多线程无法充分利用多核CPU的性能；而协程虽然基于线程实现，同样难以发挥多核处理器的全部优势。
针对计算密集型应用，采用多进程是更优选择。Python的multiprocessing模块在API设计上高度模仿threading模块，支持通过进程池机制批量创建和管理子进程。
• 创建单个Process进程（基于函数）
只需实例化Process类，将目标函数通过target参数传入，这与threading模块的使用方式相似。若函数需要参数，可通过args元组传递。
import os
from multiprocessing import Process
# 子进程要执行的代码
def task(name):
print('run child process %s (%s)...' % (name, os.getpid()))
if __name__ == '__main__':
print('parent process %s.' % os.getpid())
p = Process(target=task, args=('test',))
p.start()
p.join()
print('process end.')
• 创建单个Process进程（基于类）
通过继承Process类并重写run方法来定义进程行为，这一机制与threading模块中自定义线程的方式基本一致。
import multiprocessing
import os
from multiprocessing import current_process
class Worker(multiprocessing.Process):
def run(self):
name = current_process().name # 获取当前进程的名称
print('run child process <%s> (%s)' % (name, os.getpid()))
print('In %s' % self.name)
return
if __name__ == '__main__':
print('parent process %s.' % os.getpid())
p = Worker()
p.start()
p.join()
print('process end.')
* 停止进程
调用terminate()可强制结束子进程，但此做法并不推荐。因为强制终止可能导致子进程资源无法正确释放，且在不确定子线程运行状态时强行结束，极易引发数据不一致或资源泄漏等不可预知的问题。
import multiprocessing
import time
def worker():
print('starting worker')
time.sleep(0.1)
print('finished worker')
if __name__ == '__main__':
p = multiprocessing.Process(target=worker)
print('执行前:', p.is_alive())
p.start()
print('执行中:', p.is_alive())
p.terminate() # 发送停止号
print('停止:', p.is_alive())
p.join()
print('等待完成:', p.is_alive())
• 直接创建多个Process进程
import multiprocessing
def worker(num):
print(f'Worker:%s %s', num)
return
if __name__ == '__main__':
jobs = []
for i in range(5):
p = multiprocessing.Process(target=worker, args=(i,))
jobs.append(p)
p.start()
• 使用进程池创建多个进程
在进行系统管理时，尤其是涉及大量文件目录操作或多台主机的远程控制，并行处理能显著节省时间。当目标对象数量较少时，可直接利用multiprocessing中的Process动态生成多个进程；若涉及数十甚至上百个任务，手动限制进程数量过于繁琐，此时使用进程池（Pool）则更为高效。
Pool允许用户指定进程数量。当新请求提交至Pool且池未满时，系统会立即创建新进程执行任务；若进程数已达上限，新请求将进入等待队列，直到有现有进程结束并释放资源后，再创建新进程处理该请求。
import os
import random
import time
from multiprocessing import Pool
from time import ctime
def task(name):
print('start task %s (%s)...' % (name, os.getpid()))
start = time.time()
time.sleep(random.random() * 3)
print('end task %s runs %0.2f seconds.' % (name, (time.time() - start)))
if __name__ == '__main__':
print('parent process %s.' % os.getpid())
p = Pool() # 初始化进程池
for i in range(5):
p.apply_async(task, args=(i,)) # 追加任务 apply_async 是异步非阻塞的，就是不用等待当前进程执行完毕，随时根据系统调度来进行进程切换。
p.close()
p.join() # 等待所有结果执行完毕，会等待所有子进程执行完毕，调用join()之前必须先调用close()
print(f'all done at: {ctime()}')
若需获取每个进程的执行结果，可调用返回对象的get方法。具体实现代码如下：
import os
import random
import time
from multiprocessing import Pool, current_process
from time import ctime
def task(name):
print('start task %s (%s)...' % (name, os.getpid()))
start = time.time()
time.sleep(random.random() * 3)
print('end task %s runs %0.2f seconds.' % (name, (time.time() - start)))
return current_process().name + 'done'
if __name__ == '__main__':
print('parent process %s.' % os.getpid())
result = []
p = Pool() # 初始化进程池
for i in range(5):
result.append(p.apply_async(task, args=(i,))) # 追加任务 apply_async 是异步非阻塞的，就是不用等待当前进程执行完毕，随时根据系统调度来进行进程切换。
p.close()
p.join() # 等待所有结果执行完毕
for res in result:
print(res.get()) # get()函数得出每个返回结果的值
print(f'all done at: {ctime()}')
总结
以上介绍了Python多进程的核心实现代码，希望能对大家有所帮助。如有疑问欢迎留言，小编会及时回复。感谢对本站的支持！
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
