Python 多进程模块全面指南

一、为什么需要多进程Python 由于GIL全局解释器锁的限制多线程无法充分利用多核 CPU。多进程可以✅ 绕过 GIL 限制实现真正的并行计算✅ 适合 CPU 密集型任务科学计算、图像处理等✅ 提高程序整体吞吐量二、核心组件概览组件作用Process创建单个进程Pool进程池批量管理进程Queue进程间通信队列Pipe双向管道通信Value/Array共享内存Manager共享服务器支持多种数据结构Lock/Semaphore进程同步三、Process 类详解3.1 基本使用frommultiprocessingimportProcessimportosimporttimedefworker(name):子进程执行的任务print(f子进程{name}启动PID:{os.getpid()})time.sleep(2)print(f子进程{name}结束)if__name____main__:# Windows 必须加这行print(f主进程 PID:{os.getpid()})# 创建进程pProcess(targetworker,args(A,))p.start()# 启动进程p.join()# 等待进程结束print(主进程结束)3.2 常用属性和方法pProcess(targetfunc,nameWorker-1,daemonTrue)# 属性p.name# 进程名p.pid# 进程IDp.is_alive()# 是否存活p.daemon# 是否为守护进程# 方法p.start()# 启动进程p.join()# 阻塞等待p.terminate()# 强制终止3.3 继承 Process 类classMyProcess(Process):def__init__(self,name):super().__init__()self.namenamedefrun(self):# 必须实现 run 方法print(f自定义进程{self.name}运行中)if__name____main__:pMyProcess(Test)p.start()p.join()四、进程池 Pool4.1 基本用法frommultiprocessingimportPoolimporttimedeftask(x):time.sleep(1)returnx*xif__name____main__:# 创建包含4个进程的进程池withPool(processes4)aspool:# map: 阻塞式返回结果列表resultpool.map(task,range(10))print(fmap 结果:{result})# imap: 惰性求值返回迭代器forresinpool.imap(task,range(5)):print(fimap 结果:{res})# apply_async: 异步执行async_resultpool.apply_async(task,(20,))print(fasync 结果:{async_result.get()})4.2 Pool 常用方法对比方法特点适用场景map()阻塞有序需要全部结果imap()惰性有序大数据集节省内存imap_unordered()惰性无序不关心顺序apply()同步单任务调试apply_async()异步回调处理4.3 回调函数defcallback_func(result):print(f任务完成结果:{result})deferror_callback(err):print(f任务出错:{err})withPool(3)aspool:pool.apply_async(task,args(5,),callbackcallback_func,error_callbackerror_callback)pool.close()# 关闭进程池不再接受新任务pool.join()# 等待所有进程完成五、进程间通信5.1 Queue最常用frommultiprocessingimportProcess,Queuedefproducer(q):foriinrange(5):q.put(f产品{i})print(f生产: 产品{i})defconsumer(q):whileTrue:itemq.get()ifitemisNone:# 结束信号breakprint(f消费:{item})if__name____main__:qQueue(maxsize10)# 设置队列大小p1Process(targetproducer,args(q,))p2Process(targetconsumer,args(q,))p1.start()p2.start()p1.join()q.put(None)# 发送结束信号p2.join()5.2 Pipe双向通信frommultiprocessingimportProcess,Pipedefsender(conn):conn.send([Hello,World])conn.close()defreceiver(conn):msgconn.recv()print(f收到消息:{msg})if__name____main__:parent_conn,child_connPipe()p1Process(targetsender,args(child_conn,))p2Process(targetreceiver,args(parent_conn,))p1.start()p2.start()p1.join()p2.join()5.3 共享内存Value/ArrayfrommultiprocessingimportProcess,Value,Arraydefincrement(counter,arr):withcounter.get_lock():# 自动加锁counter.value1foriinrange(len(arr)):arr[i]1if__name____main__:# i 表示整数d 表示双精度浮点数counterValue(i,0)arrArray(d,[1.0,2.0,3.0])processes[Process(targetincrement,args(counter,arr))for_inrange(3)]forpinprocesses:p.start()forpinprocesses:p.join()print(fCounter:{counter.value})# 3print(fArray:{list(arr)})# [4.0, 5.0, 6.0]5.4 Manager最灵活frommultiprocessingimportProcess,Managerdefworker(d,l,v):d[key]valuel.append(1)v.value1if__name____main__:withManager()asmanager:shared_dictmanager.dict()shared_listmanager.list()shared_valmanager.Value(i,0)processes[Process(targetworker,args(shared_dict,shared_list,shared_val))for_inrange(3)]forpinprocesses:p.start()forpinprocesses:p.join()print(dict(shared_dict))# {key: value}print(list(shared_list))# [1, 1, 1]print(shared_val.value)# 3六、进程同步6.1 Lock互斥锁frommultiprocessingimportProcess,Lockimportsysdefsafe_print(lock,text):withlock:# 自动获取和释放锁sys.stdout.write(text\n)if__name____main__:lockLock()processes[Process(targetsafe_print,args(lock,f进程{i}))foriinrange(5)]forpinprocesses:p.start()forpinprocesses:p.join()6.2 Semaphore信号量frommultiprocessingimportProcess,Semaphoreimporttimedefaccess_resource(sem,worker_id):sem.acquire()try:print(f工人{worker_id}获得资源)time.sleep(2)finally:sem.release()if__name____main__:semSemaphore(3)# 最多3个进程同时访问processes[Process(targetaccess_resource,args(sem,i))foriinrange(10)]forpinprocesses:p.start()七、实际应用案例案例1并行文件处理frommultiprocessingimportPool,cpu_countimportosdefprocess_file(filepath):处理单个文件sizeos.path.getsize(filepath)# 模拟处理returnfilepath,sizedefbatch_process_files(file_list):并行处理文件列表withPool(min(cpu_count(),len(file_list)))aspool:resultspool.map(process_file,file_list)returndict(results)if__name____main__:files[file1.txt,file2.txt,file3.txt]sizesbatch_process_files(files)print(sizes)案例2生产者-消费者模式frommultiprocessingimportProcess,Queue,JoinableQueueimporttimeclassProducer(Process):def__init__(self,queue,items):super().__init__()self.queuequeue self.itemsitemsdefrun(self):foriteminself.items:self.queue.put(item)print(f生产:{item})time.sleep(0.5)self.queue.put(None)# 结束信号classConsumer(Process):def__init__(self,queue):super().__init__()self.queuequeuedefrun(self):whileTrue:itemself.queue.get()ifitemisNone:breakprint(f消费:{item})self.queue.task_done()if__name____main__:queueJoinableQueue()itemsrange(5)producerProducer(queue,items)consumerConsumer(queue)producer.start()consumer.start()producer.join()queue.join()# 等待队列中所有任务完成consumer.join()八、常见陷阱与最佳实践❌ 常见错误# 错误1Windows 下忘记 if __name__Process(targetfunc).start()# 会导致无限递归# 错误2在子进程中修改全局变量global_var0deffunc():globalglobal_var global_var1# 不会影响主进程的变量# 错误3使用不可序列化的对象pool.map(lambdax:x1,[open(file.txt)])# 文件对象不可序列化✅ 最佳实践# 1. 使用 if __name__ __main__if__name____main__:main()# 2. 控制进程数量cpu_numos.cpu_count()pool_sizemin(cpu_num,8)# 通常不超过CPU核心数# 3. 及时清理资源withPool()aspool:# 自动调用 close() 和 join()pass# 4. 使用 imap 处理大数据forresultinpool.imap(task,huge_list,chunksize100):process_result(result)# 5. 异常处理try:resultpool.apply_async(risky_task,(data,)).get(timeout10)exceptTimeoutError:print(任务超时)exceptExceptionase:print(f任务失败:{e})九、性能对比建议场景推荐方案CPU 密集型多进程 (Process/Pool)IO 密集型多线程 (threading) 或 asyncio少量任务Process批量任务Pool简单通信Queue复杂共享数据Manager高性能计算multiprocessing numpy十、总结Process适合少量独立任务Pool适合批量并行处理Queue/Pipe用于进程通信Manager提供灵活的共享数据Lock/Semaphore确保数据安全注意 Windows 下的__main__保护进程数通常不超过 CPU 核心数