【算法工程】使用python脚本实现对异步接口的压力测试

1. 异步接口 压力测试方案

1.1 测试目的

评估异步解析接口在高并发场景下的处理能力、稳定性和性能瓶颈。
测量系统吞吐量（TPS）、响应时间、错误率及资源使用效率，验证系统设计的合理性。

1.2 测试环境

类别	配置/说明
服务端	地址：具体的访问地址接口：/xxx/sumbit（任务提交）、`/xxx/status`（状态查询）
客户端	Python 3.7+、aiohttp 3.8+、asyncio 网络：稳定低延迟环境（避免网络波动干扰测试结果）
硬件配置	根据实际压力调整客户端机器配置（建议4核CPU、8GB内存以上）

1.3 测试工具

核心工具: 基于asyncio和aiohttp的异步Python脚本（见文末代码）。
特性支持:
- 异步并发请求，模拟真实高并发场景。
- 动态生成唯一请求ID，避免重复提交。
- 自动轮询任务状态，支持超时重试与异常处理。
- 实时计算TPS（每秒事务数）与任务完成数。

1.4 测试场景设计

场景	参数配置	目标
基准测试	总请求数=1000，并发数=20	验证系统基础性能及参数合理性
梯度压力测试	并发数=50/100/200（逐步提升）	定位系统最大吞吐量及性能拐点
稳定性测试	持续运行12小时，总请求数=动态生成	检测内存泄漏、资源竞争等长期问题

1.5 监控指标

指标	采集方式	分析目标
吞吐量	客户端脚本每分钟统计完成的任务数	评估系统吞吐量是否达标
响应时间	记录每个任务从提交到完成的耗时（代码内嵌）	分析平均/最大/最小响应时间分布
错误率	统计任务提交失败及状态查询超时的比例	验证接口容错能力与稳定性
服务端资源	独立监控（CPU、内存、磁盘IO、网络带宽）	定位系统瓶颈

1.6 测试步骤

1.6.1 环境准备

验证服务端接口可访问性：curl {server_ip}/xxx/heartbeat

安装Python依赖：pip install aiohttp uuid

1.6.2 参数配置

        server_ip = "xxx" # 替换为实际服务地址
        total_requests = 1000 # 总请求数
        max_concurrency = 20 # 并发协程数

1.6.3 执行测试

python async_stress_test.py

1.6.4 实时监控

观察控制台输出的TPS和总完成任务数。

使用独立工具（如Prometheus+Grafana）监控服务端资源。

1.6.5 结果收集

日志文件：记录任务提交、状态轮询、异常信息。

性能数据：TPS趋势图、响应时间分布表、错误率统计。

1.7 预期结果

1. 性能基线：在并发20时，TPS ≥ 15 tasks/min，平均响应时间 ≤ 30分钟（含状态轮询）。

2. 稳定性：无内存泄漏，错误率 < 1%，无服务崩溃或拒绝连接现象。

3. 扩展性：并发提升至200时，系统吞吐量应线性增长，无明显性能衰减。

1.8 风险应对

风险	应对措施
服务端过载导致宕机	分阶段增加并发，监控资源并设置熔断机制
客户端网络不稳定影响结果	使用内网或专用测试网络，多次测试取平均值
任务超时未正确处理	优化`check_status`重试逻辑，记录超时任务ID

2. 异步测试代码脚本

python">import asyncio
import aiohttp
import time
import uuid
import threading
from aiohttp import TCPConnector, ClientTimeout
from collections import dequetask_start_times = deque()
completed_tasks = 0def generate_payload():request_id = str(uuid.uuid4())return {"request_id": request_id,"param1": "xxx","param2": "xxx","param3": "xxx"}async def fetch(session: aiohttp.ClientSession, url: str, payload: dict):try:async with session.post(url, data=payload, timeout=ClientTimeout(total=1200)) as response:return await response.json()except Exception as e:print(f"请求异常: {e}")return {}async def check_status(session: aiohttp.ClientSession, status_url: str, request_id: str, max_retries=240):retry_count = 0while retry_count < max_retries:try:form_data = {'request_id': request_id}async with session.post(status_url, data=form_data, timeout=ClientTimeout(total=30)) as response:status = await response.json()message = status.get("message", "")if "xxx" in message:print(f"任务 {request_id} 成功，结果: {message}")return statuselif status.get("status") == "failure":print(f"任务 {request_id} 失败.")return statuselse:print(f"任务 {request_id} 处理中，继续等待...")await asyncio.sleep(5)retry_count += 1except aiohttp.ServerDisconnectedError:print(f"服务器断开连接，正在重试 ({retry_count+1}/{max_retries})")await asyncio.sleep(5)retry_count += 1except Exception as e:print(f"检查状态异常: {e}")await asyncio.sleep(5)retry_count += 1print(f"任务 {request_id} 超时未完成.")return {}async def worker(name: int, url: str, status_url: str, sem: asyncio.Semaphore, session: aiohttp.ClientSession):global completed_tasksasync with sem:payload = generate_payload()request_id = payload["request_id"]start_time = time.time()task_start_times.append(start_time)result = await fetch(session, url, payload)if result.get("status") == "success":print(f"任务 {name} 提交成功，request_id: {request_id}")status_result = await check_status(session, status_url, request_id)elapsed_time = time.time() - start_timeprint(f"任务 {name} 完成，耗时 {elapsed_time:.2f} 秒，状态: {status_result}")else:print(f"任务 {name} 提交失败，错误: {result.get('message')}")completed_tasks += 1  # 完成任务计数return resultasync def run_load_test(url: str, status_url: str, total_requests: int, max_concurrency: int):timeout = ClientTimeout(total=1200)connector = TCPConnector(limit_per_host=200)async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:sem = asyncio.Semaphore(max_concurrency)tasks = [asyncio.create_task(worker(i, url, status_url, sem, session)) for i in range(total_requests)]return await asyncio.gather(*tasks)def calculate_tps():global task_start_times, completed_taskswhile True:time.sleep(60)current_time = time.time()while task_start_times and current_time - task_start_times[0] > 60:task_start_times.popleft()tps = len(task_start_times)print(f"当前TPS: {tps} tasks/min")print(f"总完成任务数: {completed_tasks}")if __name__ == '__main__':server_ip = "your server ip"url = server_ip + "/xxx/submit"status_url = server_ip + "/xxx/status"total_requests = 1000max_concurrency = 20print(f"开始压力测试: 总请求数 {total_requests}, 最大并发 {max_concurrency}")start_time = time.time()tps_calculator = threading.Thread(target=calculate_tps)tps_calculator.daemon = Truetps_calculator.start()asyncio.run(run_load_test(url, status_url, total_requests, max_concurrency))total_time = time.time() - start_timeprint(f"压力测试完成: 总耗时 {total_time:.2f} 秒")