Python多进程执行同一函数的常用方法与实战指南

多进程并发执行目标与方案对比

在Python中，受GIL限制，多线程无法有效利用多核CPU进行并行计算。使用multiprocessing模块可绕过该限制，通过操作系统级进程实现真正的并行处理。以下为执行同一函数的三种主流路径：

• Process类手动调度：适合进程数量少、需独立监控生命周期或传递不同参数的场景。
• Pool进程池批量分配：适合输入数据量大、任务同质化的批处理工作，内置负载均衡与资源回收。
• apply_async异步提交：适合主流程不能中断、需按需获取各节点结果的复杂流水线。

第1步：使用Process类创建独立进程

目的：精确控制每个子进程的启动时机与参数，适用于需要差异化配置的少量任务。

入口：导入multiprocessing.Process，通过target指定函数，args传递元组参数。

import multiprocessing
import time

def worker(task_name, duration):
    print(f"进程 {task_name} 开始执行，预计耗时 {duration} 秒")
    time.sleep(duration)
    print(f"进程 {task_name} 执行完成")
    return f"{task_name} 处理结果"

if __name__ == "__main__":
    processes = []
    tasks = [("任务A", 2), ("任务B", 3), ("任务C", 1)]
    
    # 创建并启动多个进程
    for name, duration in tasks:
        p = multiprocessing.Process(target=worker, args=(name, duration))
        processes.append(p)
        p.start()
        print(f"已启动进程: {name}")
    
    # 等待所有进程完成，防止主进程提前退出
    for p in processes:
        p.join()
    
    print("所有进程执行完毕")

完成状态：控制台按顺序打印启动信息，随后各进程休眠结束后依次打印完成日志，主进程最后输出汇总提示。

第2步：利用Pool进程池自动调度任务

目的：避免频繁创建销毁系统级进程的开销，由父进程自动将数据分块派发给固定数量的工作进程。

入口：multiprocessing.Pool(processes=cpu_count)，配合.map()实现阻塞同步返回，或.map_async()实现异步包装。

import multiprocessing
import os

def worker(x):
    print(f"进程 {os.getpid()} 正在处理数据: {x}")
    result = x * x
    return result

if __name__ == "__main__":
    # 创建进程池，线程数通常设为CPU物理核心数
    with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
        data = list(range(1, 9))
        
        # map 阻塞式，按输入顺序严格返回结果
        results_sync = pool.map(worker, data)
        print(f"map 同步结果: {results_sync}")
        
        # map_async 非阻塞包装器，内部仍会阻塞至结果就绪
        async_result = pool.map_async(worker, data)
        results_async = async_result.get(timeout=10)
        print(f"map_async 异步包装结果: {results_async}")

完成状态：进程池自动复用工作进程处理全部数据，主进程在调用.get()或.map()时暂停，直至所有分片计算结束并返回有序列表。

第3步：通过apply_async实现异步非阻塞提交

目的：任务提交与结果收集解耦，主线程可在等待期间执行其他逻辑，适合长耗时或依赖外部API的请求。

入口：pool.apply_async(func, args=(param,), callback=None, error_callback=None)。

import multiprocessing
import time
import random

def worker(task_id):
    sleep_time = random.uniform(0.5, 2.0)
    time.sleep(sleep_time)
    return f"任务{task_id} 完成，耗时{sleep_time:.2f}秒"

if __name__ == "__main__":
    with multiprocessing.Pool(4) as pool:
        async_results = []
        for i in range(6):
            # 提交任务，立即返回AsyncResult对象，不阻塞主循环
            result = pool.apply_async(worker, args=(i,))
            async_results.append(result)
        
        print("所有任务已提交，主进程可继续执行其他工作...")
        
        # 遍历收集结果，第一个调用.get()时阻塞直到对应进程产出
        for i, res in enumerate(async_results):
            print(f"[{i}] {res.get()}")

完成状态：6个任务几乎同时派发，各进程因随机睡眠时长不同而先后完成，主进程依次打印结果，整体耗时低于串行执行的总和。

进阶配置：进程间通信、异常处理与性能调优

安全的数据交换通道

进程内存隔离，推荐使用Queue或Manager传递结构化数据：

from multiprocessing import Queue

def queue_worker(q_input, q_output):
    task = q_input.get()
    processed = f"处理后: {task}"
    q_output.put(processed)

异常捕获与重试机制

异步任务若抛出未处理异常，会导致该槽位死锁。必须传入error_callback明确消费错误：

def on_error(err):
    print(f"子任务崩溃，异常类型: {type(err).__name__}, 详情: {err}")

result = pool.apply_async(worker, args=(data,), error_callback=on_error)
try:
    val = result.get(timeout=30)
except Exception as e:
    print(f"获取结果超时或失败: {e}")

chunksize 参数优化

默认情况下map每次向队列推送单个元素，高频IO场景下IPC开销显著。手动指定chunksize可减少进程间通信次数：

# 经验法则：chunksize 设为数据长度除以进程数的整数倍
avg_chunk = max(1, len(data) // multiprocessing.cpu_count())
results = pool.map(worker, data, chunksize=avg_chunk)

实战演练：并行网络请求下载

目的：验证多进程在处理I/O密集型高延迟任务时的吞吐提升效果。

import multiprocessing
import requests

def download_url(url):
    try:
        resp = requests.get(url, timeout=10)
        return f"{url} | 成功 | {len(resp.content)} bytes"
    except requests.RequestException as e:
        return f"{url} | 失败 | {e}"

if __name__ == "__main__":
    targets = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1"
    ]
    
    with multiprocessing.Pool(3) as pool:
        reports = pool.map(download_url, targets)
    
    for rpt in reports:
        print(rpt)

注意事项：执行此类脚本前请确保已安装第三方库pip install requests。实际生产环境中建议增加重试装饰器与连接池复用策略。

总结与选型建议

Python多进程并发执行同一函数的核心在于权衡控制粒度与自动化程度。纯手工维护Process适合调试与异构参数分发；Pool.map以牺牲一定灵活性为代价换取最简代码与稳定负载均衡；apply_async则填补了流水线异步编排的空白。合理搭配Queue通信、错误回调与chunksize分块，即可覆盖绝大多数CPU与I/O混合加速需求。以上就是Python多进程并发执行同一函数的详细内容，更多关于Python并发编程的技巧资料请关注本站其它相关文章！