Python如何进行压力测试？
想模拟 100 个用户同时登陆，并同时访问我的一个接口，用余额买商品，大家有什么方法，好的脚本。
（之前因为搞活动，大量用户买东西，存进数据库有数据丢失，还有数据出错）
Python如何进行压力测试？
3 回复
Python压力测试实战指南
压力测试的核心是模拟高并发请求，Python最常用的工具是locust和requests。下面我直接给你一个完整的locust示例：
# locustfile.py
from locust import HttpUser, task, between
class WebsiteUser(HttpUser):
wait_time = between(1, 3) # 用户请求间隔1-3秒
@task(3) # 权重3，执行频率更高
def view_homepage(self):
self.client.get("/")
@task(1) # 权重1
def view_products(self):
self.client.get("/products")
@task(2)
def post_login(self):
self.client.post("/login", json={
"username": "test_user",
"password": "test_pass"
})
运行方式：
# 启动web界面（推荐）
locust -f locustfile.py --host=http://your-api.com
# 无界面模式直接压测
locust -f locustfile.py --host=http://your-api.com --headless -u 100 -r 10 -t 1m
参数说明：-u 100模拟100用户，-r 10每秒启动10用户，-t 1m持续1分钟。
如果你需要更底层的控制，用asyncio + aiohttp自己写：
import asyncio
import aiohttp
from datetime import datetime
async def send_request(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
url = "http://your-api.com/endpoint"
concurrency = 100 # 并发数
duration = 30 # 测试30秒
async with aiohttp.ClientSession() as session:
tasks = []
start = datetime.now()
while (datetime.now() - start).seconds < duration:
for _ in range(concurrency):
task = asyncio.create_task(send_request(session, url))
tasks.append(task)
await asyncio.sleep(0.1)
results = await asyncio.gather(*tasks)
print(f"总请求数: {len(results)}")
asyncio.run(main())
简单总结：用locust最快，要自定义就上asyncio。
