Python多进程队列使用异常:子进程结果无法正常存储
问题分析
1. 全局变量无法共享的原因
multiprocessing的每个子进程都拥有独立的内存空间,主进程的result_nums和子进程里的result_nums是完全独立的对象。子进程对result_nums的修改不会同步到主进程,所以最终主进程打印的长度为0。
2. Queue导致进程停滞的原因
multiprocessing.Queue底层依赖系统管道实现,当队列中的数据积累到一定程度(填满管道缓冲区),子进程调用queue.put(num)时会进入阻塞状态,等待主进程从队列取走数据。但你的代码中,主进程启动所有子进程后直接调用p.join()等待子进程结束,完全没有取数据的操作,导致子进程一直卡在put步骤无法完成任务,形成死锁——主进程等子进程结束,子进程等主进程取数据。
解决方法
方法一:改进Queue的使用,主进程主动取数据
在等待子进程的同时,主进程持续从队列中取出数据,避免队列满导致子进程阻塞:
from multiprocessing import Process, Queue def task(nums, queue): for num in nums: for _ in range(10000): pass queue.put(num) if __name__ == "__main__": num_packages = [] for _ in range(10): num_packages.append(list(range(50000))) queue = Queue() processes = [] for i in range(10): p = Process(target=task, args=(num_packages[i], queue)) processes.append(p) p.start() # 主进程循环取数据,直到所有子进程结束且队列空 res_list = [] while any(p.is_alive() for p in processes) or not queue.empty(): while not queue.empty(): res_list.append(queue.get_nowait()) # 等待所有子进程彻底结束 for p in processes: p.join() print(len(res_list)) # 输出500000
方法二:使用multiprocessing.Pool简化实现
Pool会自动管理进程池和结果收集,无需手动处理队列,代码更简洁易维护:
from multiprocessing import Pool def process_num(num): for _ in range(10000): pass return num if __name__ == "__main__": # 生成所有需要处理的数字 all_nums = [] for _ in range(10): all_nums.extend(range(50000)) # 创建进程池,默认使用CPU核心数,也可指定比如Pool(10) with Pool() as pool: # 若数据量过大,可改用imap并设置chunksize分批处理,减少内存占用 res_list = pool.map(process_num, all_nums) print(len(res_list)) # 输出500000
如果数据量极大,map一次性传入所有数据可能占用过多内存,可改用imap并设置chunksize参数优化:
with Pool() as pool: res_list = list(pool.imap(process_num, all_nums, chunksize=1000))
额外注意点
- 多进程间禁止用全局变量传递数据,必须通过
Queue、Pipe或Manager提供的共享对象实现数据交互。 - 使用
Queue时,主进程务必在子进程运行期间执行取数据操作,避免队列满导致阻塞。 - 批量任务优先使用进程池方案,代码可读性和维护性更强。
内容的提问来源于stack exchange,提问作者Preston Brown
相关产品推荐
相关产品推荐

