Python多进程实现批量POST请求：1万司机数据1分钟更新方案求助

当前通过单循环遍历字典列表逐个发送POST请求更新司机状态，IO等待导致效率低下，目标是1分钟内完成1万条数据更新，以下是基于多进程分块处理的优化方案：

核心思路 #

利用concurrent.futures.ProcessPoolExecutor实现多进程并行处理，将1万条司机数据分块后分配给不同进程，同时发起HTTP请求，通过并行IO操作提升整体效率。

完整代码实现 #

import requests
from concurrent.futures import ProcessPoolExecutor
import math

# 全局复用请求头，所有进程共享
headers = {
    "accept": "application/json",
    "content-type": "application/json",
    "authorization": "#########"
}
# 基础URL模板
post_url = "https://api.xyz.com/v1/test/drivers/{}/daily_status/duty_status"

def update_single_driver(driver_info):
    """单个司机状态更新的核心函数，供进程调用"""
    try:
        # 拼接目标URL
        target_url = post_url.format(driver_info['driver_id'])
        # 构造请求体
        payload = {
            "status": driver_info['status_type'],
            "remark": driver_info['remark'],
            "time": driver_info['time']
        }
        # 发送POST请求
        response = requests.post(target_url, json=payload, headers=headers)
        # 返回处理结果（可根据需求调整）
        if response.status_code in (200, 201):
            return f"司机{driver_info['driver_id']}状态更新成功"
        else:
            return f"司机{driver_info['driver_id']}更新失败，状态码：{response.status_code}"
    except Exception as e:
        return f"司机{driver_info['driver_id']}更新出错：{str(e)}"

def batch_update_drivers(details, process_count=8):
    """批量更新入口函数，负责数据分块与多进程调度"""
    # 计算每个进程处理的数据块大小
    chunk_size = math.ceil(len(details) / process_count)
    # 将数据列表拆分为多个子块
    data_chunks = [details[i:i+chunk_size] for i in range(0, len(details), chunk_size)]
    
    # 启动多进程池执行任务
    with ProcessPoolExecutor(max_workers=process_count) as executor:
        # 每个进程处理一个数据块
        chunk_results = executor.map(lambda chunk: [update_single_driver(info) for info in chunk], data_chunks)
    
    # 收集所有处理结果
    all_results = []
    for result in chunk_results:
        all_results.extend(result)
    return all_results

# 主程序入口（Windows系统必须加此判断，避免进程创建异常）
if __name__ == "__main__":
    # 替换为你的实际1万条司机数据列表
    driver_details = [
        {"driver_id": "123", "status_type": "ON_DUTY", "remark": "testing", "time": 1670009505000}
        for _ in range(10000)
    ]
    # 执行批量更新
    update_results = batch_update_drivers(driver_details)
    # 可按需打印或保存结果（示例打印前10条）
    for res in update_results[:10]:
        print(res)

关键注意事项 #

• 进程数设置：建议设置为CPU核心数的2-4倍（比如8核CPU开8-16个进程），过多进程会增大系统调度开销，还可能触发后端API限流。
• API限流适配：提前确认后端API的并发/QPS限制，若有限制，可在update_single_driver函数中添加time.sleep(0.01)之类的延迟，避免被封禁。
• 失败重试：对请求失败的场景，可引入重试逻辑（比如用tenacity库），提升任务成功率。
• 数据分块调整：分块大小可根据实际情况微调，比如每块100-500条数据，平衡进程负载。

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