Python连接池实现：限制100连接分块处理DataFrame并提交API

核心思路 #

用requests的连接池配置结合线程池来控制并发连接数（最多100个），同时按10000条为一批处理数据，每批完成后提交到API并记录结果。不需要用asyncio，线程池的方式更适配requests的同步特性，实现起来更简单直接。

具体实现步骤 #

• 配置连接池：通过HTTPAdapter设置连接池的最大连接数为100，同时配置重试机制避免临时网络问题导致失败。
• 分批处理数据：按10000条切片遍历DataFrame，每批作为一个处理单元。
• 并发请求数据：用ThreadPoolExecutor启动最多100个线程，每个线程负责一个ID的请求与数据解析，将结果存入临时字典（避免直接在DataFrame上操作的线程安全问题）。
• 批量更新DataFrame：每批请求完成后，将临时字典中的结果批量更新到DataFrame对应行。
• 提交数据到API：将处理好的10000条数据转为字典格式提交，记录提交范围、成功状态以及失败的ID列表。

完整代码 #

import pandas as pd
import requests
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 初始化DataFrame（替换成你的170万条含ID的数据）
df = pd.DataFrame({'id': range(1, 1700001)})
df['Data'] = None  # 预留存储接口返回数据的列

# 配置核心参数
chunk_size = 10000
max_concurrent = 100
target_url_template = 'https://your-target-api.com/data/{id}'  # 替换成你的目标URL模板
submit_api_url = 'https://your-submit-api.com/save'  # 替换成你的提交API地址

# 配置requests连接池与重试策略
session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(
    max_retries=retry_strategy,
    pool_connections=max_concurrent,
    pool_maxsize=max_concurrent
)
session.mount('https://', adapter)
session.mount('http://', adapter)

# 记录所有批次的处理结果
processing_logs = []

def fetch_single_data(record_id):
    """单个ID的请求与解析逻辑"""
    try:
        response = session.get(target_url_template.format(id=record_id), timeout=10)
        response.raise_for_status()  # 主动抛出HTTP错误
        json_data = response.json()
        return (record_id, json_data.get('data'), None)  # 返回(ID, 有效数据, 错误信息)
    except Exception as e:
        return (record_id, None, str(e))

# 开始分批处理
total_rows = len(df)
for start_idx in range(0, total_rows, chunk_size):
    end_idx = min(start_idx + chunk_size, total_rows)
    batch_ids = df.iloc[start_idx:end_idx]['id'].tolist()
    batch_range = f"{start_idx+1}-{end_idx}"
    print(f"正在处理批次：{batch_range}")

    # 并发请求数据
    batch_result_map = {}
    failed_request_ids = []

    with ThreadPoolExecutor(max_workers=max_concurrent) as executor:
        task_futures = {executor.submit(fetch_single_data, rid): rid for rid in batch_ids}
        for future in as_completed(task_futures):
            rid, data, err = future.result()
            if err:
                failed_request_ids.append(rid)
                print(f"ID {rid} 请求失败：{err}")
            else:
                batch_result_map[rid] = data

    # 批量更新DataFrame
    df.loc[df['id'].isin(batch_result_map.keys()), 'Data'] = df['id'].map(batch_result_map)

    # 提交有效数据到API
    valid_submit_data = df.iloc[start_idx:end_idx].dropna(subset=['Data']).to_dict(orient='records')
    submit_status = "成功"
    try:
        submit_response = session.post(
            submit_api_url,
            data=json.dumps(valid_submit_data),
            headers={'Content-Type': 'application/json'},
            timeout=30
        )
        submit_response.raise_for_status()
    except Exception as e:
        submit_status = f"失败：{str(e)}"

    # 记录当前批次日志
    processing_logs.append({
        'batch_range': batch_range,
        'submit_status': submit_status,
        'failed_request_ids': failed_request_ids,
        'submitted_count': len(valid_submit_data)
    })

# 输出最终处理汇总
print("\n所有批次处理完成")
for log in processing_logs:
    print(f"批次 {log['batch_range']}: 提交状态{log['submit_status']}，失败请求ID共{len(log['failed_request_ids'])}个")

关键说明 #

• 连接池控制：pool_connections和pool_maxsize均设为100，严格匹配目标服务器的连接限制，避免触发限流。
• 效率优化：先将请求结果存入字典再批量更新DataFrame，比逐行调用df.loc效率提升数倍，同时规避线程安全风险。
• 错误追踪：详细记录每批的失败请求ID和提交状态，方便后续复盘与补处理。

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