Python使用multiprocessing实现多进程完整指南

林舟

发布时间：2026-10-11| 2人浏览过

原创

本文详细解析Python中如何利用multiprocessing模块实现多进程并行计算。通过Process类的标准创建流程、核心方法调用以及数据隔离特性演示，帮助开发者快速搭建跨平台多任务处理框架，提升CPU密集型任务的执行效率。

一、多进程基础与核心概念

进程是操作系统分配资源与调度的基本单位。程序（如 .py 文件）仅是静态保存在硬盘中的代码集合，而将其运行后加载到内存中的动态实体即为进程。在 Python 环境中，若需同时处理多个 CPU 密集型任务，单线程顺序执行会导致明显阻塞。此时可通过 multiprocessing 模块开启多个独立进程，利用操作系统的任务调度或硬件多核特性，实现真正的并行计算或高效并发。

需注意两个易混淆概念：

并发（Concurrent）：任务数多于 CPU 核数时，系统通过时间片轮询机制交替执行，宏观上看似“一起”运行。
并行（Parallel）：当任务数小于或等于 CPU 物理核数时，不同核心可同时处理独立任务，属于真实的同步执行。

二、快速实现第一个多进程任务

第一步：定义目标函数并引入模块

多进程编程的核心入口是 multiprocessing.Process 类。首先需要导入该模块，并将需要异步执行的具体逻辑封装为独立函数。

from multiprocessing import Process
import time

def task_work(language):
    """定义需要在子进程中执行的计算逻辑"""
    for i in range(5):
        print(f"{language} coding | ", end='')
        time.sleep(1)

第二步：在主进程中实例化与启动子进程

脚本入口需使用 if __name__ == '__main__': 进行条件保护（Windows 系统下必需）。随后创建 Process 实例，传入目标函数 target 及位置参数 args（注意必须是元组格式，单元素时也需加逗号），最后调用 start() 激活子进程。主进程与子进程将各自独立调度运行。

if __name__ == '__main__':
    # 耗时基准测试（单进程串行）
    start_time = time.time()
    task_work('python')
    print("Single process cost:", time.time() - start_time)

    # 多进程并行测试
    multi_start = time.time()
    p = Process(target=task_work, args=('python', ))
    p.start()  # 触发子进程创建并开始执行
    
    # 主进程同步执行其他任务
    for i in range(5):
        print("main program | ", end='')
        time.sleep(1)
        
    print("\nMulti process cost:", time.time() - multi_start)

执行结果显示，双路并行耗时从约 10 秒压缩至 5 秒左右，验证了多任务重叠执行带来的效率提升。

三、Process类核心方法与属性配置

除了基础的创建流程，掌控进程生命周期与状态追踪依赖以下常用接口：

start()：创建子进程实例并触发 target 函数执行
join([timeout])：阻塞主进程直到子进程结束，可选传入超时阈值以防死锁
terminate()：强制立即终止子进程，无论内部任务是否完成
is_alive()：返回布尔值，实时判断进程是否仍在运行
.name：进程别名，默认格式为 Process-N（N 自增）
.pid：当前进程的系统级唯一标识符

获取父子进程 ID 需借助内置 os 模块：os.getpid() 返回当前进程号，os.getppid() 返回父进程号。建议在关键业务中结合日志打印这些标识以便调试跟踪。

四、验证进程间数据隔离性

与多线程共享全局变量不同，多进程的每个实例拥有独立的内存地址空间。全局变量在创建子进程时会复制一份副本，后续任何一方的修改均不会相互影响。

from multiprocessing import Process
import time

share_data = ['初始状态']

def worker_1():
    print('[P1] 初始:', share_data)
    for char in 'thon':
        share_data.append(char * 3)
        time.sleep(0.5)
        print(f'[P1] 追加 {char*3}, 当前: {share_data}')

def worker_2():
    print('[P2] 初始:', share_data)
    for char in 'thon':
        share_data.append(char)
        time.sleep(0.5)
        print(f'[P2] 追加 {char}, 当前: {share_data}')

if __name__ == '__main__':
    p1 = Process(target=worker_1)
    p2 = Process(target=worker_2)
    p1.start()
    p2.start()
    p1.join()
    p2.join()

运行结果表明，P1 向列表追加 'ttt', 'hhh' 等，P2 仅追加 't', 'h' 等，双方对 share_data 的写入完全隔离，印证了“进程间不共享全局变量”的设计原则。若实际业务需要跨进程交换数据或共享复杂对象，应改用 Queue、Pipe 或 Manager 提供的 IPC 机制。

五、常见排错与使用建议

Windows 报错 NameError: name 'Process' is not defined：未添加 if __name__ == '__main__': 防护导致子进程递归 spawn 自身，务必确保主逻辑在此条件块内。
进程卡死无响应：未及时调用 p.join() 且主进程提前退出，会导致子进程成为孤儿进程或异常中断。建议始终配对使用 start 与 join。
参数传递限制：args 仅支持可序列化的基础数据类型。若需传递字典或自定义类实例，请检查 Python 版本兼容性或使用 Pool 扩展。