如何用Python测量后端单位时间内可处理的GET请求数及响应时间

原代码的核心问题 #

• 计时逻辑错误：用datetime.datetime.now().microsecond计算耗时，只取了微秒部分，跨秒时会出现负数，且未计算秒级差异，导致结果严重不准。应该用time.perf_counter()，这是专门用于短时间计时的高精度函数。
• 未利用异步并发：原代码是串行执行异步请求，每次必须等上一个请求响应返回后才发送下一个，和同步代码的并发量没区别，完全浪费了aiohttp的异步能力。
• 请求计数定义模糊：原代码中serviced统计的是成功返回的请求数，因为只有当await resp.json()完成后才会递增，未返回的请求不会被计数。

需求完全可行，解决方案如下 #

要实现“指定时长内高并发发送请求+统计按时返回的请求数+精准计时每个请求”，可以通过异步任务池+精准计时+超时控制来实现：

修正后的代码示例 #

import asyncio
import time
import aiohttp

BASE_URL = "http://localhost:8000/test/"  # 替换成你的后端地址

async def single_request(session, url):
    # 记录请求发送前的时间点
    start_time = time.perf_counter()
    try:
        async with session.get(url) as resp:
            await resp.json()  # 等待响应内容解析完成
            # 计算耗时，转成微秒（可选，也可以保留秒）
            elapsed = (time.perf_counter() - start_time) * 1_000_000
            return True, elapsed
    except Exception as e:
        # 处理请求失败的情况（比如连接错误、超时）
        print(f"请求失败: {e}")
        return False, 0

async def scalability_test(seconds):
    total_serviced = 0
    total_response_time = 0.0
    start_test = time.perf_counter()
    active_tasks = set()
    sent_requests = 0  # 可选：统计发送的总请求数

    async with aiohttp.ClientSession() as session:
        while time.perf_counter() - start_test < seconds:
            sent_requests += 1
            # 生成请求URL（可以根据需要调整，比如用递增ID或者随机参数）
            request_url = f"{BASE_URL}{sent_requests}"
            # 创建异步任务并加入任务集合
            task = asyncio.create_task(single_request(session, request_url))
            active_tasks.add(task)
            # 当任务完成时，从集合中移除
            task.add_done_callback(lambda t: active_tasks.remove(t))
            
            # 可选：控制并发数，防止瞬间创建过多任务耗尽资源
            # if len(active_tasks) > 1000:
            #     await asyncio.sleep(0.001)

        # 若严格只统计指定时长内返回的请求，可跳过这段等待；若要等所有已发请求完成，保留这段
        # await asyncio.gather(*active_tasks, return_exceptions=True)
        
        # 统计已完成的有效请求
        for task in asyncio.all_tasks():
            if task.done() and task is not asyncio.current_task():
                result = task.result()
                if result[0]:
                    total_serviced += 1
                    total_response_time += result[1]

    # 输出统计结果
    print(f"{seconds}秒内发送的总请求数: {sent_requests}")
    print(f"{seconds}秒内成功处理的请求数: {total_serviced}")
    if total_serviced > 0:
        print(f"总响应耗时: {total_response_time:.2f}微秒")
        print(f"平均响应耗时: {total_response_time / total_serviced:.2f}微秒")

if __name__ == "__main__":
    asyncio.run(scalability_test(10))

关键改进点 #

1. 精准计时：用time.perf_counter()记录请求发起和响应完成的时间差，精度可达纳秒级别，完全避免了跨秒计时错误。
2. 真正的异步并发：通过asyncio.create_task()批量创建请求任务，同时发送多个请求，充分利用异步IO的优势模拟高并发场景。
3. 清晰的计数逻辑：新增sent_requests统计发送的总请求数，total_serviced统计在测试时长内成功返回的请求数，满足你对两类计数的明确需求。
4. 可选的并发控制：如果担心瞬间创建过多任务导致系统资源耗尽，可以添加并发数限制（代码中注释的部分），避免压垮本地后端。

额外说明 #

• 如果要严格限制测试总时长（不等待未返回的请求），直接去掉await asyncio.gather(*active_tasks...)这段代码即可。
• 可以拓展统计请求失败的数量，更全面地评估后端的稳定性。
• 本地测试时，注意后端的最大并发处理能力，避免因后端崩溃导致测试数据失真。

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