如何用Python多进程加速Locust测试中批量ID的数据库入库校验？

问题根源分析 #

你之前用multiprocessing失败的核心原因之一是全局变量在多进程中无法共享更新——每个子进程都有独立的内存空间，修改子进程内的yes/no不会同步到主进程，最后统计结果必然错误。另外，单线程循环处理数百个网络请求会被IO等待拖慢效率，多进程/多线程才是正确的优化方向。

最优实现方案 #

针对这种IO密集型的批量请求场景，推荐使用concurrent.futures.ProcessPoolExecutor（多进程）或ThreadPoolExecutor（多线程，IO密集型更高效），两者API一致，切换成本极低。以下是具体实现：

1. 多进程实现代码

import requests
from concurrent.futures import ProcessPoolExecutor

def check_sausage(sausage):
    """单个香肠的校验逻辑，返回是否为香肠的布尔值"""
    try:
        response = requests.get(f"http://www.isitasausage.com/{sausage}")
        response.raise_for_status()  # 主动捕获HTTP请求错误（如404、500）
        return "Yes, this is a sausage" in response.text
    except Exception as e:
        print(f"校验{sausage}时出错: {str(e)}")
        return False  # 出错的情况统一归为"非香肠"

if __name__ == "__main__":
    all_sausages = ["cumberland", "lego", "bratwurst"]
    
    # 进程数建议设置为CPU核心数（可通过os.cpu_count()获取），IO密集型可适当增加
    with ProcessPoolExecutor(max_workers=4) as executor:
        # 批量提交任务，获取所有结果
        results = list(executor.map(check_sausage, all_sausages))
    
    # 统计结果，无需全局变量
    yes = sum(results)
    no = len(results) - yes
    
    print(f"数组中香肠的数量 = {yes}")
    print(f"数组中非香肠的数量 = {no}")

2. 更高效的IO密集型优化：多线程实现

因为网络请求属于IO密集型任务（大部分时间在等待响应），线程切换的开销远小于进程，用ThreadPoolExecutor能获得更高的效率，只需替换一行代码：

# 把ProcessPoolExecutor换成ThreadPoolExecutor即可
from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=10) as executor:  # 线程数可设为20-50，根据并发限制调整
    results = list(executor.map(check_sausage, all_sausages))

关键注意事项 #

• 必须加if __name__ == "__main__"：Windows系统下多进程启动时会重新导入模块，不加这句话会导致进程无限创建，直接崩溃。
• 避免全局变量统计：通过子进程/线程返回结果，主进程统一汇总，这是多进程/线程编程的标准做法。
• 控制并发数：不要把max_workers设得过大，否则可能触发目标服务器的限流机制，根据目标接口的并发限制调整。
• 异常处理：一定要捕获请求中的异常（如网络超时、HTTP错误），避免单个请求失败导致整个批量任务中断。

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