aiohttp与requests基准测试是否可靠？求社区评估测试方案

嘿，先给你点个赞——做技术选型前搞基准测试是非常靠谱的做法！不过从你给出的代码片段来看，目前的方案还有不少可以优化的地方，我帮你梳理下：

现有代码的明显问题 #

首先你的aiohttp代码没写完，而且从已有的部分来看，还没做到和requests的配置对齐，这会直接影响测试的公平性：

• requests用了Session（复用连接池），但aiohttp如果没正确使用ClientSession而是每次创建新的，会额外增加连接建立的开销，结果就偏了
• 同步的requests是单线程单请求，而异步aiohttp默认可以并发多个请求，两者的并发模型不对等，没法直接比

优化后的公平测试方案建议 #

要做到公平对比，得从配置对齐、并发模型匹配、测试流程规范这几个方面入手：

1. 统一连接池与基础配置 #

• 给requests的Session设置明确的连接池大小（比如adapter = requests.adapters.HTTPAdapter(pool_connections=10, pool_maxsize=10)）
• 给aiohttp的ClientSession配置对应的Connector，比如connector=aiohttp.TCPConnector(limit=10)，和requests的连接池大小保持一致
• 两者都关闭自动重连，或者统一重连次数，避免差异

2. 匹配并发模型 #

• 如果要测单请求性能：那两边都单次执行请求，统计单次耗时（但这种场景下异步优势不明显）
• 如果要测高并发场景：requests需要用线程池（比如concurrent.futures.ThreadPoolExecutor）来模拟并发，线程数要和aiohttp的并发数一致；aiohttp用asyncio.gather控制并发量，避免无限制并发压垮目标服务器

3. 规范测试流程 #

• 预热环节：测试正式开始前，先跑3-5次请求，让连接池建立、JIT编译（如果有）完成，避免冷启动影响结果
• 多次取平均：单次测试结果波动大，建议跑10-20轮测试，取平均耗时、成功率等指标
• 多维度统计：别只看总耗时，还要统计响应时间的分布（p50/p95/p99）、请求成功率、错误率，这些更能反映真实性能

4. 完整测试代码示例 #

给你补一个对齐后的测试代码参考：

import asyncio
import aiohttp
import requests
import time
from concurrent.futures import ThreadPoolExecutor
from statistics import mean, median

TEST_URL = "https://a-domain-i-can-use.tld"
TEST_TIMES = 20
CONCURRENCY = 10  # 并发数，两边保持一致

# ---------------------- Requests 测试（线程池模拟并发）----------------------
def requests_single_fetch(session):
    try:
        with session.get(TEST_URL, timeout=5) as resp:
            resp.raise_for_status()
            return len(resp.text)
    except Exception as e:
        print(f"Requests 请求失败: {e}")
        return None

def run_requests_test():
    # 配置连接池
    session = requests.Session()
    adapter = requests.adapters.HTTPAdapter(pool_connections=CONCURRENCY, pool_maxsize=CONCURRENCY)
    session.mount("https://", adapter)
    
    # 预热
    for _ in range(3):
        requests_single_fetch(session)
    
    total_times = []
    success_count = 0
    
    for _ in range(TEST_TIMES):
        start = time.perf_counter()
        with ThreadPoolExecutor(max_workers=CONCURRENCY) as executor:
            results = list(executor.map(lambda _: requests_single_fetch(session), range(CONCURRENCY)))
        success_count += sum(1 for res in results if res is not None)
        total_times.append(time.perf_counter() - start)
    
    session.close()
    return {
        "总平均耗时": mean(total_times),
        "中位数耗时": median(total_times),
        "总成功率": success_count / (TEST_TIMES * CONCURRENCY)
    }

# ---------------------- aiohttp 测试 ----------------------
async def aio_single_fetch(session):
    try:
        async with session.get(TEST_URL, timeout=5) as resp:
            resp.raise_for_status()
            return len(await resp.text())
    except Exception as e:
        print(f"aiohttp 请求失败: {e}")
        return None

async def run_aiohttp_test():
    # 配置连接池
    connector = aiohttp.TCPConnector(limit=CONCURRENCY)
    async with aiohttp.ClientSession(connector=connector) as session:
        # 预热
        for _ in range(3):
            await aio_single_fetch(session)
        
        total_times = []
        success_count = 0
        
        for _ in range(TEST_TIMES):
            start = time.perf_counter()
            tasks = [aio_single_fetch(session) for _ in range(CONCURRENCY)]
            results = await asyncio.gather(*tasks)
            success_count += sum(1 for res in results if res is not None)
            total_times.append(time.perf_counter() - start)
    
    return {
        "总平均耗时": mean(total_times),
        "中位数耗时": median(total_times),
        "总成功率": success_count / (TEST_TIMES * CONCURRENCY)
    }

# ---------------------- 执行测试并输出结果 ----------------------
if __name__ == "__main__":
    print("=== Requests 测试结果 ===")
    requests_result = run_requests_test()
    for k, v in requests_result.items():
        print(f"{k}: {v:.4f}")
    
    print("\n=== aiohttp 测试结果 ===")
    aio_result = asyncio.run(run_aiohttp_test())
    for k, v in aio_result.items():
        print(f"{k}: {v:.4f}")

额外注意事项 #

• 测试环境要稳定：尽量在同一台机器、相同网络环境下测试，关闭其他占用带宽/CPU的进程
• 目标服务器状态：确保测试期间目标服务器没有其他负载，避免外部因素干扰结果
• 超时设置：两边统一超时时间，避免因超时逻辑不同导致结果偏差

内容的提问来源于stack exchange，提问作者Loïc