
去年我们有个定时任务,需要对一批用户数据做清洗和转换。因为数据量不小,我决定用多进程来加速。
代码大概是这样的:
import multiprocessing as mp
# 全局配置,所有子进程共用
config = {"mode": "fast", "retry": 3}
counter = 0
def process_data(data):
global counter
# 根据config处理数据
result = do_something(data, config)
counter += 1 # 统计处理了多少条
return result
if __name__ == "__main__":
with mp.Pool(4) as pool:
results = pool.map(process_data, big_data_list)
print(f"一共处理了{counter}条数据")跑完之后,counter输出是0。
我盯着屏幕看了十分钟——counter明明是全局变量,所有子进程都在累加它,为什么最后是0?
更诡异的是,我加了个打印语句,在每个子进程里打印counter的值,发现每个子进程里的counter都在正常累加。但主进程里的counter始终是0。
全局变量不“全局”了。
后来我才明白:这个锅不该Python背,是操作系统的机制决定的。每个进程有自己独立的内存空间,所谓的“全局变量”只是在各自进程内部“全局”,跨进程根本看不见。
先搞清楚一个基本概念:进程是操作系统分配资源的最小单位。
当你创建一个子进程时,操作系统会为它分配独立的内存空间。这个空间里包含了父进程内存的一份拷贝——注意,是拷贝,不是引用。
这就好比你把一份文件复印了4份,分别发给4个人。每个人都在自己的复印件上写写画画,但你手里的原件一个字都没变。
多线程和多进程的区别就在这里:
Python官方文档也明确说了:global关键字只控制变量在模块内部的可见性,跟跨进程共享没有半点关系。
开头那个counter的例子就是典型。每个子进程都有一份counter的拷贝,各自累加各自的。等所有子进程跑完,主进程的counter还是初始值0。
有人可能会想:那我在子进程里用global counter声明一下呢?
没用。global只解决同一个进程内不同作用域的访问问题。跨进程,global管不着。
在Linux/macOS下,默认用fork方式启动子进程,子进程会复制父进程的整个内存空间,全局变量还能读到。
但在Windows下,只能用spawn方式——启动一个全新的Python解释器,重新执行所有代码。子进程里的全局变量是重新初始化后的值,父进程里后续的修改,子进程根本不知道。
这就导致一个更隐蔽的问题:你在Windows下写的多进程代码,部署到Linux上可能行为不同;反过来也一样。
用Pool.map的时候,每个子进程处理完任务就退出了。你在子进程里对全局变量的任何修改,随着子进程的结束一起消失。
from multiprocessing import Pool
total = 0
def worker(x):
global total
total += x
return x * 2
if __name__ == "__main__":
with Pool(4) as pool:
pool.map(worker, [1,2,3,4])
print(total) # 还是0每个worker都活在独立的进程里,它修改的total是它自己那份,主进程的total从来没变过。
Python的multiprocessing模块提供了几种解决方案。
不要依赖全局变量,把数据通过参数显式传递。
def process_data(data, config):
# 用传进来的config,不用全局的
return do_something(data, config)
if __name__ == "__main__":
CONFIG = {"mode": "fast"}
with mp.Pool(4) as pool:
# 把config作为参数传给每个worker
results = pool.map(lambda x: process_data(x, CONFIG), big_data_list)这是最干净的方式,没有隐式依赖,没有跨进程共享的烦恼。
multiprocessing.Value和Array(共享内存)如果确实需要在多个进程间共享数据,可以用Value(单个值)或Array(数组)。
from multiprocessing import Pool, Value
counter = Value('i', 0) # 'i'表示整数
def worker(x):
with counter.get_lock(): # 加锁防止竞争
counter.value += x
if __name__ == "__main__":
with Pool(4) as pool:
pool.map(worker, [1,2,3,4])
print(counter.value) # 10Value和Array使用的是共享内存,多个进程可以直接读写同一块内存区域。注意要加锁,否则多个进程同时修改会出问题。
Manager(共享复杂对象)如果需要共享列表、字典等复杂数据结构,用Manager。
from multiprocessing import Pool, Manager
def worker(shared_list, x):
shared_list.append(x)
if __name__ == "__main__":
manager = Manager()
shared_list = manager.list()
with Pool(4) as pool:
pool.map(lambda x: worker(shared_list, x), [1,2,3,4])
print(list(shared_list)) # [1,2,3,4]Manager会启动一个独立的“管理器进程”来协调所有共享数据的访问。优点是支持多种数据类型,缺点是比共享内存慢,因为每次读写都要走进程间通信。
Queue(任务分发和结果收集)如果你只是想分发任务、收集结果,Queue是最合适的选择。
from multiprocessing import Process, Queue
def worker(q):
while True:
item = q.get()
if item is None:
break
# 处理item
q.put(item * 2)
if __name__ == "__main__":
q = Queue()
# 放任务、启动进程、取结果...场景 | 推荐方案 | 原理 |
|---|---|---|
子进程只需要读取父进程的数据 | 参数传递 | 拷贝一份,各用各的 |
子进程需要修改共享的简单数据 | Value / Array | 共享内存 |
子进程需要修改共享的复杂数据 | Manager | 管理器进程协调 |
任务分发 / 结果收集 | Queue | 进程间消息传递 |
我那个计数器最后怎么解决的?用了Value:
from multiprocessing import Pool, Value
counter = Value('i', 0)
def process_data(data, config):
global counter
with counter.get_lock():
counter.value += 1
return do_something(data, config)这次终于能正确统计处理条数了。
记住:多进程不共享全局变量,这不是Python的bug,是操作系统的设计。想跨进程共享数据,别指望global,用对工具才是正解。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。