如何将现有多线程代码转换为Multiprocessing Pool以测试性能？

转换后的代码示例 #

import multiprocessing

user_dict = {
    'users': [
        {
            'username': 'test1',
            'password': 'opop1',
        },
        {
            'username': 'test2',
            'password': 'opop2',
        }
    ]
}

def task(some_object, some_other_object, another_object, username, password):
    # 原有的登录、API调用、写入JSON逻辑保持不变
    pass

if __name__ == '__main__':
    # 准备每个任务的参数列表，将固定对象和用户凭证打包成元组
    tasks_args = []
    for user in user_dict['users']:
        args_tuple = (
            some_object,
            some_other_object,
            another_object,
            user['username'],
            user['password']
        )
        tasks_args.append(args_tuple)
    
    # 创建进程池，默认使用CPU核心数，也可手动指定processes参数
    with multiprocessing.Pool() as pool:
        # 用starmap分发任务，自动将元组内元素作为参数传入task
        pool.starmap(task, tasks_args)

关键说明 #

• 参数传递逻辑：进程池的starmap方法专门适配多参数任务，会把列表中的每个元组拆解为独立参数传入目标函数，完全匹配原线程调用的参数结构。
• 序列化要求：进程间内存不共享，所有传递给子进程的对象（some_object等）必须能被Python的pickle模块序列化。如果是自定义类实例，要确保类定义能被子进程访问，且没有不可序列化的属性（比如文件句柄、活跃网络连接）。
• 进程池管理：用with语句创建进程池会自动处理close()和join()操作，无需手动调用，避免资源泄漏。
• 性能测试提示：你的场景（登录、API请求、文件写入）属于IO密集型任务，多线程在这类场景通常更高效（线程切换开销远低于进程）。转用进程池后如果性能没有提升，属于正常现象，可根据实际测试结果选择合适的并发方式。

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