Python中如何对Django Restful API进行并发性能测试

RT

我有台机器配置还是不错（ 48 核 CPU、128GB 内存），将一个 Django 的 Web 应用搭建在这上面了，里面有几个 api 接口用于数据库插入和更新操作，然后使用的 celery （ 200worker ）去消费 mq 的任务，就会出现并发调用 api 接口情况，结果发现很多请求提示 connect time out 情况如下。

error <br>Max retries exceeded with url: /api/member/test1/insertorupdate (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x3b1f750>: Failed to establish a new connection: [Errno 110] Connection timed out',))",),) <br>

感觉应该是 django restful api 不支持如此高并发导致的，所以想请教 V2EX 的大神，我该如何测试和提升并发量，多谢！
Python中如何对Django Restful API进行并发性能测试

wuwangju 1楼

你不是已经测试了吗？

sinazl 2楼作者

用locust吧，这玩意儿测并发最直接。下面是个完整例子，测一个假设的/api/items/端点。

先装包：pip install locust

然后建个locustfile.py：

from locust import HttpUser, task, between

class ApiUser(HttpUser):
    wait_time = between(1, 3)  # 用户执行任务间隔1-3秒
    
    @task
    def get_items(self):
        # GET请求示例
        self.client.get("/api/items/")
    
    @task(3)  # 权重为3，执行频率更高
    def create_item(self):
        # POST请求示例
        payload = {"name": "test_item", "price": 100}
        headers = {"Content-Type": "application/json"}
        self.client.post("/api/items/", json=payload, headers=headers)
    
    @task(2)
    def update_item(self):
        # PUT请求示例，假设item_id为1
        payload = {"price": 150}
        self.client.put("/api/items/1/", json=payload)

运行测试：locust -f locustfile.py --host=http://your-api-host

浏览器打开http://localhost:8089，设置并发用户数和每秒启动速率就能开测。locust会实时显示RPS、响应时间和失败率。

用locust直接写脚本测就完事了。