Python多进程优化Meraki交换机端口状态获取报错排查

错误原因 #

Python多进程拥有独立的内存空间，子进程无法直接访问父函数的局部变量。你写的switchsts函数里用到的switches、dashboard、yesterday_midnight都是getSwitchStatus的局部变量，子进程完全看不到这些变量，所以触发NameError。

修正方案 #

方案1：手动传递参数+队列收集结果（单进程示例，仅演示参数传递） #

如果要保留Process手动管理进程，必须把所有需要的参数传给子进程，并用队列收集返回结果：

from multiprocessing import Process, Queue
import datetime
from datetime import timedelta, time
import meraki

def switchsts(queue, dashboard, switches, yesterday_midnight):
    print("Inside switchsts")
    statuses = []
    for dic in switches:
        statuses.append(dashboard.switch.getDeviceSwitchPortsStatuses(dic['serial'], t0=yesterday_midnight))
    queue.put(statuses)

def getSwitchStatus(dashboard: meraki.DashboardAPI, switches): 
    print("Testing if switches is accessible")
    print("Switches type", type(switches))
    print("Switches", switches[0])

    yesterday_midnight = datetime.combine(datetime.today(), time.min) - timedelta(days=1)
    result_queue = Queue()
    # 把所有需要的参数传入子进程
    p = Process(target=switchsts, args=(result_queue, dashboard, switches, yesterday_midnight))
    p.start()
    p.join()
    # 从队列获取结果
    statuses = result_queue.get()
    print(statuses)
    return statuses

方案2：用进程池实现真正的并行处理（推荐） #

单个进程起不到提速效果，用multiprocessing.Pool可以批量分发任务到多个进程，自动处理结果收集：

from multiprocessing import Pool
import datetime
from datetime import timedelta, time
import meraki

# 单个交换机的任务函数，接收打包后的参数
def fetch_switch_port_status(task_args):
    dashboard, switch_serial, time_stamp = task_args
    return dashboard.switch.getDeviceSwitchPortsStatuses(switch_serial, t0=time_stamp)

def getSwitchStatus(dashboard: meraki.DashboardAPI, switches): 
    print("Testing if switches is accessible")
    print("Switches type", type(switches))
    print("Switches", switches[0])

    yesterday_midnight = datetime.combine(datetime.today(), time.min) - timedelta(days=1)
    # 打包每个交换机的任务参数
    task_list = [(dashboard, dic['serial'], yesterday_midnight) for dic in switches]
    
    # 创建进程池，可手动指定进程数（比如processes=8，避免API限流）
    with Pool(processes=8) as pool:
        # 批量执行任务，收集结果
        statuses = pool.map(fetch_switch_port_status, task_list)
    
    print(statuses)
    return statuses

注意事项 #

• Meraki API存在请求频率限制，并行进程数不要设置过高，建议先查官方文档的限流规则，再调整processes参数
• 进程池的map方法会自动按顺序返回结果，和输入的task_list顺序对应，无需担心结果混乱

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