高负载TCP请求数据抓取函数的优化方案咨询

算力阿峰

2026-7-26

现有代码问题分析

DataFrame拼接效率极低：循环中每次用pd.concat都会生成新的DataFrame，频繁的内存拷贝会拖慢整体速度，还会占用大量内存。
异常捕获过于宽泛：except:会捕获所有异常，包括逻辑错误（比如KeyError），导致问题难以排查，应该只捕获网络请求、JSON解析相关的具体异常。
payload线程安全隐患：当前代码直接修改全局的payload字典，如果后续用多线程，多个线程同时修改PageNo会导致页码混乱，必须为每个线程创建独立的payload副本。
无重试机制：遇到临时网络波动、服务器超时等问题直接跳过，会丢失数据，也没有应对限流（如429状态码）的策略。
无请求节流：单线程循环请求已经可能给服务器造成压力，多线程如果并发过高，容易被目标网站封禁IP。

Session使用正确性验证

你当前的Session用法是正确的：同一个requests.Session实例会复用TCP连接（保持长连接），避免每次请求都进行三次握手，确实能减少TCP层面的开销。但要注意requests.Session不是线程安全的——如果多线程共用一个Session，可能会出现请求混淆、连接池异常等问题，所以每个线程必须创建独立的Session实例。

多线程优化实现方案

针对分页请求的场景，用concurrent.futures.ThreadPoolExecutor拆分页码范围是合理的，以下是改写后的代码示例：

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

def fetch_page(url, payload, page_no):
    # 每个线程创建独立的Session和重试策略
    session = requests.Session()
    # 配置重试机制：针对连接错误、超时、5xx错误自动重试
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,  # 重试间隔：1s, 2s, 4s...
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    # 创建当前页的独立payload副本，避免线程间干扰
    page_payload = payload.copy()
    page_payload["PageNo"] = page_no
    
    try:
        # 设置超时时间，避免请求挂起
        response = session.post(url, json=page_payload, timeout=10)
        response.raise_for_status()  # 触发HTTP错误（如404、500）的异常
        json_data = response.json()
        return pd.DataFrame(json_data["AllResults"])
    except Exception as e:
        print(f"Page {page_no} failed: {str(e)}")
        return pd.DataFrame()  # 返回空DataFrame，不影响后续合并

def get_data(url, payload, total_pages, max_workers=5):
    # 用列表存储每页的DataFrame，最后一次性合并
    df_list = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有页码的请求任务
        futures = {executor.submit(fetch_page, url, payload, page): page for page in range(1, total_pages+1)}
        
        # 处理完成的任务
        for future in as_completed(futures):
            page_df = future.result()
            if not page_df.empty:
                df_list.append(page_df)
    
    # 一次性合并所有DataFrame
    if df_list:
        return pd.concat(df_list, ignore_index=True)
    else:
        return pd.DataFrame()

# 使用示例
if __name__ == "__main__":
    payload = {
        "MoreAssestsType": None,
        "FillterRoomNum": 0,
        "GridDisplayType": 0,
        "ResultLable": "city",
        "ResultType": 1,
        "ObjectID": "5000",
        "ObjectIDType": "text",
        "ObjectKey": "UNIQ_ID",
        "X": 180428.31832654,
        "Y": 665726.5550939,
        "Gush": "",
        "Parcel": "",
        "showLotParcel": False,
        "showLotAddress": False,
        "OriginalSearchString": "city",
        "CurrentLavel": 2,
        "Navs": [],
        "QueryMapParams": {
            "QueryToRun": None,
            "QueryObjectID": "5000",
            "QueryObjectType": "number",
            "QueryObjectKey": "SETL_CODE",
            "QueryDescLayerID": "KSHTANN_SETL_AREA",
            "SpacialWhereClause": True,
        },
        "isHistorical": False,
        "PageNo": 1,
        "OrderByFilled": "DEALDATETIME",
        "OrderByDescending": True,
        "Distance": 0
    }
    result_df = get_data("你的目标URL", payload, total_pages=5200, max_workers=5)
    print(result_df.shape)

其他性能优化建议

控制并发数：max_workers不要设置太高（建议5-10），过高的并发会导致目标服务器限流，甚至IP被封。可以根据服务器返回的响应速度调整。
避免内存溢出：如果数据量极大，不要把所有DataFrame都存在内存里，可以分批写入数据库（比如用df.to_sql搭配chunksize参数），或者写入CSV文件。

模拟浏览器请求：在Session中添加User-Agent、Accept等请求头，避免被目标网站识别为爬虫：

session.headers.update({
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"
})

处理限流响应：如果遇到429状态码，除了重试，还可以在重试策略中增加更长的延迟，或者主动在单个请求失败后添加短暂休眠。
验证数据完整性：最后可以检查返回的总数据量是否符合预期（比如每页固定条数的话，总条数=每页条数×有效页数），避免遗漏数据。

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