Python加速含PUT请求for循环的多进程/并发实现方案

架构老林

2026-8-29

问题根因

之前的并发方案没有提速，核心是三个低级错误：

你写的testing2逻辑里，把future.result()阻塞等待逻辑放在了提交任务的for循环内部，每提交1个请求就立刻卡着等结果返回，本质和串行执行没有区别
max_workers=2设置的并发数太低，从日志看单个请求平均耗时400-600ms，2并发的理论耗时还是接近串行的一半，远达不到大幅提速的要求
跨线程共享同一个requests.Session存在线程安全隐患，容易出现请求头错乱、连接复用异常的问题
多进程完全不适合这个场景：HTTP请求属于典型IO密集型任务，多进程的进程创建、上下文切换开销远大于多线程，属于杀鸡用牛刀。

正确实现代码

直接替换原有newapproach.py内容即可，默认15并发，按单请求500ms计算，原60秒的任务可以压缩到4秒左右完成：

import requests
import json
import urllib3
import cpack_utility
from cpack_utility import classes
import concurrent.futures
import time
from functools import partial

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def single_permission_update(task_config, payload):
    """单条权限更新请求，每个线程独立创建session避免线程安全问题"""
    url, header, verifySSL = task_config
    session = requests.Session()
    session.headers.update(header)
    resp = session.put(url, verify=verifySSL, data=payload, timeout=10)
    return resp.json()

def update_role(data):
    logger = cpack_utility.logging.get_logger("role")
    mApi = classes.morphRequests_config()
    url, header, verifySSL = mApi.role()

    # 从读取的JSON中提取所有权限配置
    permissions_roleprivs = data["rolePermissions"]["roleprivs"]
    task_payloads = []
    for v in permissions_roleprivs.values():
        payload = json.dumps({
            "permissionCode": v["permissionCode"],
            "access": v["access"]
        })
        task_payloads.append(payload)

    start = time.time()
    max_workers = 15 # 可根据接口限流规则调整，一般内部接口开10-20不会触发限流
    success = 0
    fail = 0
    # 绑定固定请求参数，避免重复传值
    task_runner = partial(single_permission_update, (url, header, verifySSL))

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 第一步：把所有任务一次性提交到线程池
        futures = [executor.submit(task_runner, payload) for payload in task_payloads]
        # 第二步：所有任务提交完成后再统一等待结果，绝对不能边提交边等
        for future in concurrent.futures.as_completed(futures):
            try:
                result = future.result()
                logger.debug(result)
                if result.get("success"):
                    success += 1
                else:
                    fail += 1
                    logger.error(f"更新失败，返回: {result}")
            except Exception as e:
                fail += 1
                logger.error(f"请求异常: {str(e)}")

    cost = round(time.time() - start, 2)
    print(f"处理完成，总耗时: {cost}秒，成功{success}条，失败{fail}条")

def main(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    update_role(data)

if __name__ == '__main__':
    main("data.json")

关键说明

彻底修复了边提交任务边等待结果的逻辑错误：先全量提交任务，再通过as_completed遍历已完成的任务，真正实现并行请求
去掉了全局变量：把logger、接口配置的初始化放到对应函数内部，避免多线程/多进程场景下全局变量跨上下文传递的异常
线程隔离Session：每个工作线程独立创建requests Session，避免多线程共享同一个Session带来的连接异常、头信息错乱问题
增加了异常捕获和成功/失败统计，避免单个请求失败导致整个任务中断
不需要用多进程：IO密集型场景下Python的GIL会在网络等待时自动释放，多线程调度开销远低于多进程，性能完全足够，也不需要处理进程间数据传递的复杂问题
如果接口端有明确限流规则，把max_workers调到限流阈值以下即可，避免触发流控导致请求失败。

之前的asyncio后台方案没有提速，本质是默认线程池并发数太低，也没有做任务提交和等待的逻辑拆分，实际并发度和串行差不多。

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