如何用Python3高效发送10万次Web请求并批量检测URL状态码？

先聊聊你当前代码的核心问题：无限制创建线程会让操作系统陷入疯狂的上下文切换，不仅不会提速，反而会拖垮系统；另外每个请求都新建PyCurl实例，初始化的开销累积起来也是不小的时间成本。结合你是同域名多子目录的场景，我们可以通过复用连接、控制并发数来最大化效率，下面是具体方案：

方案1：优化PyCurl + 线程池（推荐，兼容现有技术栈） #

线程池可以控制并发数（建议根据服务器承载能力设为100-200），同时复用PyCurl实例和TCP连接，彻底解决前期准备耗时久的问题：

import pycurl
import certifi
from concurrent.futures import ThreadPoolExecutor

# 提前创建可复用的PyCurl实例，避免重复初始化
def create_reusable_curl():
    curl = pycurl.Curl()
    curl.setopt(pycurl.CAINFO, certifi.where())
    curl.setopt(pycurl.WRITEFUNCTION, lambda x: None)
    curl.setopt(pycurl.CONNECTTIMEOUT, 5)
    # 开启TCP长连接，复用同域名连接
    curl.setopt(pycurl.TCP_KEEPALIVE, 1)
    curl.setopt(pycurl.TCP_KEEPIDLE, 30)
    curl.setopt(pycurl.TCP_KEEPINTVL, 10)
    return curl

def check_single_url(url, curl):
    try:
        curl.setopt(pycurl.URL, url)
        curl.perform()
        status = curl.getinfo(pycurl.HTTP_CODE)
        return f"Requests | URL: {url} | Status Code: {status}"
    except pycurl.error as e:
        return f"Requests | URL: {url} | Error: {str(e)}"

if __name__ == "__main__":
    # 迭代读取URL，避免大文件占用过多内存
    with open("urllist.txt") as f:
        urls = [line.strip() for line in f if line.strip()]
    
    # 设置线程池大小，同域名下并发过高易被限流，100-200是安全范围
    max_workers = 150
    # 为每个线程分配一个独立的PyCurl实例
    curl_instances = [create_reusable_curl() for _ in range(max_workers)]
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 轮询分配curl实例，确保每个实例被复用
        results = executor.map(
            lambda idx: check_single_url(urls[idx], curl_instances[idx % max_workers]),
            range(len(urls))
        )
        
        # 建议批量输出结果，减少控制台IO开销
        for res in results:
            print(res)
    
    # 最后统一关闭所有curl实例
    for curl in curl_instances:
        curl.close()

方案2：异步IO（aiohttp）—— 无线程切换开销的极致效率 #

如果你的环境支持异步，aiohttp的异步请求+连接池会更高效，因为异步模型没有线程上下文切换的额外开销：

import aiohttp
import asyncio

async def check_url(session, url):
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
            return f"URL: {url} | Status Code: {resp.status}"
    except Exception as e:
        return f"URL: {url} | Error: {str(e)}"

async def main():
    with open("urllist.txt") as f:
        urls = [line.strip() for line in f if line.strip()]
    
    # 配置连接池，复用TCP连接
    connector = aiohttp.TCPConnector(
        limit=150,  # 同时保持的连接数
        keepalive_timeout=30,
        force_close=False
    )
    
    async with aiohttp.ClientSession(connector=connector) as session:
        # 批量创建异步任务
        tasks = [check_url(session, url) for url in urls]
        # 并发执行所有任务
        results = await asyncio.gather(*tasks)
        
        for res in results:
            print(res)

if __name__ == "__main__":
    asyncio.run(main())

关键优化点总结 #

• 控制并发数：不要无限制创建线程/任务，同域名下并发过高容易被服务器限流，100-200是比较安全的范围。
• 复用连接：开启HTTP Keep-Alive，复用TCP连接，避免每次请求都经历三次握手的开销，这对同域名请求的提速效果极其明显。
• 复用请求实例：不管是PyCurl还是aiohttp的Session，提前创建好实例复用，能大幅减少初始化的时间成本。
• 减少不必要IO：频繁的控制台打印会拖慢速度，建议先把结果存入列表，最后批量写入文件或打印。

内容的提问来源于stack exchange，提问作者神瀬来未