快捷導(dǎo)航

fastapi與django異步的并發(fā)對(duì)比分析

更新時(shí)間：2023年03月15日 08:51:22 作者：Chise1

這篇文章主要介紹了fastapi與django異步的并發(fā)對(duì)比分析，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

概述

據(jù)說(shuō)fastapi是目前最快的異步框架，遂決定將其和django異步進(jìn)行并發(fā)比較。

先說(shuō)結(jié)果

fastapi的異步可以使整體運(yùn)行速度非常均衡，不會(huì)出現(xiàn)較大波動(dòng)，但是django會(huì)出現(xiàn)大量的波動(dòng)問(wèn)題，部分訪問(wèn)速度很快，但是部分訪問(wèn)速度很慢，甚至超時(shí)。

個(gè)人感覺(jué)是因?yàn)閐jango的數(shù)據(jù)庫(kù)訪問(wèn)是同步的，但是fastapi我是用的異步數(shù)據(jù)庫(kù)訪問(wèn)。（django不支持異步數(shù)據(jù)庫(kù)訪問(wèn)）

在長(zhǎng)時(shí)間并發(fā)情況下，fastapi的運(yùn)行狀態(tài)非常穩(wěn)定，但是django的運(yùn)行速度不行（另外我發(fā)現(xiàn)runserver運(yùn)行雖然慢，但是至少測(cè)試能成功，daphne并發(fā)量大了測(cè)試直接失敗。。）。（之后還會(huì)考慮pypy的測(cè)試和其他情況的）

測(cè)試環(huán)境

阿里云

服務(wù)器： 2 vCPU 4 GiB (I/O優(yōu)化) ecs.c5.large 4Mbps

數(shù)據(jù)庫(kù)：rds.mysql.s1.small

服務(wù)器和數(shù)據(jù)庫(kù)都在華北三

測(cè)試指令：

ab -n 10000 -c 1000 http://127.0.0.1:8002/get_update_info

在更大并發(fā)和更長(zhǎng)時(shí)間的情況下daphne測(cè)試報(bào)錯(cuò)，就算設(shè)置-k也報(bào)錯(cuò)，但是fastapi能夠很穩(wěn)定的響應(yīng)。

測(cè)試結(jié)果

示例代碼

#fastapi
# -*- encoding: utf-8 -*-
"""
@File    : test2.py
@Time    : 2020/3/13 14:21
@Author  : chise
@Email   : chise123@live.com
@Software: PyCharm
@info    :
"""
from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "mysql+pymysql://......"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)

engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)

app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


class D(BaseModel):
    id: int
    info: str
    data: str


@app.get('/get_update_info', response_model=D)
async def get_update_info():
    return await database.fetch_one("SELECT * FROM `publicadmin_updateinfo` LIMIT 1 ;")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app,)

django代碼如下

class UpdateInfo(models.Model):
    info = models.TextField(verbose_name="更新日志")
    data = models.TextField(verbose_name="主要內(nèi)容")

    # data = MDTextField(verbose_name="主要內(nèi)容")
    class Meta:
        verbose_name = "公告欄"
        verbose_name_plural = verbose_name
def get_update_info(request):
    """
    更新日志
    Args:
        request:
    Returns:
    """
    info: UpdateInfo = UpdateInfo.objects.first()
    return JsonResponse({"id": info.id, "msg": info.info, "main_info": info.data})