Python多进程问题：向pool.map传递多变量/全局变量

问题背景 #

运行多ticker循环的多进程代码时，compile()函数提示全局变量ticker未定义。单ticker代码可正常运行，但循环处理多个ticker时出错，尝试调整循环位置、直接传递多参数均未解决问题。

出错代码 #

import multiprocessing
from multiprocessing import Pool

global ticker
global lst

lst = ['BABA','MSFT','NVDA']

def compile(file_list):
    print(file_list)
    print(f'C: {ticker}')

def main():
    print(f'B: {ticker}')
    file_list = [1,2,3]
    with Pool(multiprocessing.cpu_count()-2) as pool:
        results_df = pool.map(compile, file_list)
    print(f'D: {ticker}')

if __name__ == '__main__':
    for ticker in lst:
        print(f'A: {ticker}')
        main()

报错输出 #

A: BABA
B: BABA
1
2
3

multiprocessing.pool.RemoteTraceback: 

Traceback (most recent call last):
  File "/Users/xxx/opt/anaconda3/lib/python3.9/multiprocessing/pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
  File "/Users/xxx/opt/anaconda3/lib/python3.9/multiprocessing/pool.py", line 48, in mapstar
    return list(map(*args))
  File "/Users/xxx/Desktop/OptionsData/example 2.py", line 12, in compile
    print(f'C: {ticker}')
NameError: name 'ticker' is not defined

可行的单ticker代码 #

import multiprocessing
from multiprocessing import Pool

def compile(file_list):
    print(file_list)
    print(f'C: {ticker}')

def main():
    print(f'B: {ticker}')
    file_list = [1,2,3]
    with Pool(multiprocessing.cpu_count()-2) as pool:
        results_df = pool.map(compile, file_list)
    print(f'D: {ticker}')

global ticker
ticker ='BABA'

if __name__ == '__main__':
    print(f'A: {ticker}')
    main()

单ticker代码输出 #

A: BABA
B: BABA
1
C: BABA
2
C: BABA
3
C: BABA
D: BABA
[Finished in 347ms]

解决方案 #

问题根源是多进程中，子进程拥有独立内存空间，不会继承主进程循环中动态赋值的全局变量，必须显式传递参数给子进程函数。以下是三种有效解决方法：

方法1：使用functools.partial绑定固定参数 #

通过partial将ticker绑定到compile函数，让pool.map只需传递file_list参数：

import multiprocessing
from multiprocessing import Pool
from functools import partial

lst = ['BABA','MSFT','NVDA']

def compile(ticker, file_item):
    print(file_item)
    print(f'C: {ticker}')

def main(ticker):
    print(f'B: {ticker}')
    file_list = [1,2,3]
    # 绑定ticker到compile函数，生成新的单参数函数
    bound_compile = partial(compile, ticker)
    with Pool(multiprocessing.cpu_count()-2) as pool:
        results_df = pool.map(bound_compile, file_list)
    print(f'D: {ticker}')

if __name__ == '__main__':
    for ticker in lst:
        print(f'A: {ticker}')
        main(ticker)

方法2：传递元组作为参数，修改compile函数拆解元组 #

将ticker和每个file_item打包成元组，compile函数通过拆解元组获取两个参数：

import multiprocessing
from multiprocessing import Pool

lst = ['BABA','MSFT','NVDA']

def compile(args):
    file_item, ticker = args
    print(file_item)
    print(f'C: {ticker}')

def main(ticker):
    print(f'B: {ticker}')
    file_list = [1,2,3]
    # 打包每个file_item和ticker成元组
    task_list = [(item, ticker) for item in file_list]
    with Pool(multiprocessing.cpu_count()-2) as pool:
        results_df = pool.map(compile, task_list)
    print(f'D: {ticker}')

if __name__ == '__main__':
    for ticker in lst:
        print(f'A: {ticker}')
        main(ticker)

方法3：使用pool.starmap直接传递多参数 #

starmap可自动将迭代对象中的元素拆解为函数的多个参数，适合函数需要多参数的场景：

import multiprocessing
from multiprocessing import Pool

lst = ['BABA','MSFT','NVDA']

def compile(file_item, ticker):
    print(file_item)
    print(f'C: {ticker}')

def main(ticker):
    print(f'B: {ticker}')
    file_list = [1,2,3]
    # 生成包含多参数的迭代对象
    task_list = [(item, ticker) for item in file_list]
    with Pool(multiprocessing.cpu_count()-2) as pool:
        results_df = pool.starmap(compile, task_list)
    print(f'D: {ticker}')

if __name__ == '__main__':
    for ticker in lst:
        print(f'A: {ticker}')
        main(ticker)

关键说明 #

• 多进程中，子进程与主进程内存空间独立，主进程的全局变量不会自动同步到子进程，必须显式传递参数。
• 避免依赖全局变量传递动态变化的参数，尤其是循环中修改的全局变量，子进程无法感知其变化。

内容的提问来源于stack exchange，提问作者andy_ruler_of_omicronpersei8