如何高效获取2000个URL状态码并以字典存储？

你的同步requests.get方案串行处理每个请求，速度慢是必然的。下面提供两种高效优化方案，同时先修正你原代码里的一个小bug：异常分支里你把append写成了赋值，会覆盖之前的结果，应该改成result["anything_else"].append((url, str(e)))。

方案一：用ThreadPoolExecutor实现多线程请求 #

基于你熟悉的requests库，通过多线程并行处理IO密集型的网络请求，改动成本低，能显著提升速度。

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

def check_single_url(url):
    try:
        response = requests.get(url, timeout=10)  # 增加超时避免请求卡住拖慢整体
        return (url, response.status_code, None)
    except requests.exceptions.RequestException as e:
        return (url, None, str(e))

def check_urls_threaded(list_of_urls, max_workers=20):
    result = {"200": [], "404": [], "anything_else": []}
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有检测任务
        futures = {executor.submit(check_single_url, url): url for url in list_of_urls}
        
        # 实时处理已完成的任务
        for future in as_completed(futures):
            url, status_code, error = future.result()
            if status_code == 200:
                result["200"].append(url)
            elif status_code == 404:
                result["404"].append(url)
            else:
                if error:
                    result["anything_else"].append((url, error))
                else:
                    result["anything_else"].append((url, f"HTTP Error {status_code}"))
    
    return result

关键说明： #

• max_workers：线程数建议设为20-50，过多可能触发目标网站反爬限制，可根据实际网络环境调整
• 增加timeout参数：避免单个无响应请求卡住整个任务流程
• 用as_completed实时处理结果，无需等待所有请求完成，效率更高
• 列表的append操作是原子性的，无需额外加锁，线程安全

方案二：用aiohttp实现异步请求 #

异步IO是处理大量网络请求的最优解，比多线程资源占用更低、效率更高，需要用到aiohttp库（先通过pip install aiohttp安装）。

import aiohttp
import asyncio

async def check_single_url_async(session, url):
    try:
        async with session.get(url, timeout=10) as response:
            return (url, response.status, None)
    except Exception as e:
        return (url, None, str(e))

async def check_urls_async(list_of_urls):
    result = {"200": [], "404": [], "anything_else": []}
    
    # 复用客户端会话提升连接效率
    async with aiohttp.ClientSession() as session:
        # 创建所有异步任务
        tasks = [check_single_url_async(session, url) for url in list_of_urls]
        # 并发执行所有任务并收集结果
        results = await asyncio.gather(*tasks)
        
        # 归类处理结果
        for url, status_code, error in results:
            if status_code == 200:
                result["200"].append(url)
            elif status_code == 404:
                result["404"].append(url)
            else:
                if error:
                    result["anything_else"].append((url, error))
                else:
                    result["anything_else"].append((url, f"HTTP Error {status_code}"))
    
    return result

# 调用示例
if __name__ == "__main__":
    urls = ["https://example.com", ...]  # 替换为你的2000个URL列表
    final_result = asyncio.run(check_urls_async(urls))

关键说明： #

• 异步模式下单个线程可处理成百上千并发请求，资源占用远低于多线程
• ClientSession是异步请求的核心，复用它能减少TCP连接建立的开销
• asyncio.gather用于批量执行异步任务，统一收集结果

方案选择 #

• 若只想快速优化现有代码、不想学习新库，选ThreadPoolExecutor方案
• 若追求极致性能、处理超大量URL，选aiohttp异步方案

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