Python使用multiprocessing并行调用API收集失败player_id的实现咨询

• Pool.map 传参逻辑完全错位
  pool.map(func, iterable) 的固定规则是：把iterable里的每一个单独元素作为入参传给func。你现在把send_player_result的形参定义成接收List[str]类型的列表，但实际传进去的是单个字符串格式的player_id，运行时直接触发类型错误。
• 失败收集逻辑不全
  你在函数里遍历到第一个失败的player_id就直接return，同个任务上下文里后续的player_id根本不会被处理，完全没法收集全所有失败的id。
• 运行时直接报错的语法问题
  你写的failed_player_ids = Set(responses)里的Set是typing模块提供的类型注解，不是Python内置的可运行构造器，跑的时候直接抛NameError，实际用来创建集合的是内置的小写set()。
• 隐藏的跨进程传参坑
  你代码里直接引用的player_api、user都是主进程的全局变量，在Windows系统、或者macOS默认的spawn启动模式下，子进程无法直接继承这些对象，会出现序列化报错、对象状态不一致的问题。

这个场景是典型的IO密集型任务（绝大多数时间在等待API返回，不是占满CPU做计算），并行收益极高，不受CPU核心数的限制。

核心改造思路 #

1. 把任务函数改成单元素处理逻辑：每个任务只接收一个player_id，调用API后失败就返回该id，成功就返回None，不在函数里写嵌套循环。
2. 解决全局对象传递问题：用Pool自带的initializer参数，在每个子进程启动时一次性初始化好需要用到的player_api、user对象，避免每个任务都重复序列化传输大对象。
3. 按API速率限制匹配进程数：你提到API上限是每分钟2000次调用，单次调用耗时5秒，单进程每分钟仅能完成12次调用，开20个进程的话每分钟仅能处理240次，远低于上限，运行非常安全。

示例可运行代码 #

from multiprocessing import Pool, current_process
from typing import List, Optional, Set

# 子进程启动时一次性执行的初始化逻辑，每个子进程只跑一次
def _init_worker(api_obj, user_obj):
    # 把需要的对象绑定到当前子进程的上下文
    current_process().player_api = api_obj
    current_process().user = user_obj

def _send_single_result(player_id: str) -> Optional[str]:
    # 直接从当前子进程上下文取初始化好的对象，不需要重复传参
    proc = current_process()
    success = proc.player_api.send_results(
        player_id,
        user=proc.user,
        send_health_results=True
    )
    return player_id if not success else None

if __name__ == "__main__":
    # 可根据实际API速率调整进程数，这里取20远低于2000次/分钟的上限
    PROCESS_NUM = 20
    # 替换为你实际的player_id列表、API实例、用户对象
    player_ids: List[str] = []
    global_player_api = None
    global_user = None

    with Pool(
        processes=PROCESS_NUM,
        initializer=_init_worker,
        initargs=(global_player_api, global_user)
    ) as pool:
        # 把每个player_id单独传给处理函数，拿到全量返回结果
        result_list = pool.map(_send_single_result, player_ids)

    # 过滤掉成功的None结果，剩下的就是所有失败的player_id
    failed_player_ids: Set[str] = set(res for res in result_list if res is not None)

• Windows、macOS（默认spawn启动模式）下，所有Pool相关代码必须放在if __name__ == "__main__":判断块内，不然会无限递归创建子进程，直接把设备卡死。
• 不要一开始就把进程数开太高，先从10个开始压测，观察API返回状态，避免触发速率限制被封禁。
• 如果需要处理的player_id量特别大（比如几十万条），可以把pool.map换成pool.imap_unordered，可以边处理边拿结果，不需要等所有任务跑完才返回完整列表，内存占用会低很多。

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