34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
import asyncio
|
|
import httpx
|
|
|
|
URL = "http://127.0.0.1:8000/api/user_info"
|
|
|
|
# 并发量（你可以改 50 / 100）
|
|
MAX_CONCURRENCY = 100
|
|
TOTAL_REQUESTS = 2000
|
|
|
|
# 超时加长（关键）
|
|
timeout = httpx.Timeout(30, connect=30)
|
|
|
|
async def fetch(client: httpx.AsyncClient, idx: int):
|
|
try:
|
|
res = await client.get(URL, timeout=timeout)
|
|
print(f"请求 {idx:03d} | 状态码:{res.status_code} | 长度:{len(res.content)}")
|
|
except Exception as e:
|
|
# 这里会显示真实失败原因！
|
|
print(f"请求 {idx:03d} | 失败原因: {str(e)}")
|
|
|
|
async def run_pressure():
|
|
# 限制并发量（关键！）
|
|
limiter = asyncio.Semaphore(MAX_CONCURRENCY)
|
|
|
|
async with httpx.AsyncClient(limits=httpx.Limits(max_connections=MAX_CONCURRENCY)) as client:
|
|
tasks = []
|
|
for i in range(TOTAL_REQUESTS):
|
|
async with limiter:
|
|
tasks.append(fetch(client, i))
|
|
await asyncio.gather(*tasks)
|
|
|
|
if __name__ == "__main__":
|
|
print(f"🚀 开始压测：总 {TOTAL_REQUESTS} 请求，并发 {MAX_CONCURRENCY}")
|
|
asyncio.run(run_pressure()) |