Pythonflask与fastapi性能测试方法介绍

Tanisha ·

更新时间:2024-11-10

· 466 次阅读

背景

apache ab介绍

测试计划

测试代码

测试结果

结论

背景

sy项目通过MQ接受业务系统的业务数据，通过运行开发者开发的python脚本执行业务系统与财务系统数据的一致性校验。

sy系统需要每天运行大量的python脚本。目前使用falsk日运行6W+次python脚本，由于性能存在瓶颈，需要引入

新的fastapi框架，来解决cpu、内存性能压榨不够及目前的性能瓶颈。本文目标给出两者的性能测试报告。

给出选择哪个框架的性能数据支撑。

apache ab介绍

apache ab性能测试

安装

yum -y install httpd-tools

部分参数说明

-n 执行的请求总数

-c 并发数, 同时执行的数量, c不能大于n
-p post请求指定的文件
-T header Content-type值，默认为 'text/plain'

测试get请求

ab -c 10 http://127.0.0.1:8081/cppla

测试post请求

ab -n 100 -c 10 -T 'application/json' -p httpjson.txt http://127.0.0.1:8081/cppla1

// httpjson.txt的内容
{"recordId": 123}

测试计划

模拟真实每次请求调用脚本，分别对每一个数量级的请求量进行测试。

请求总数每次并发数每次并发数每次并发数

100	10	100	1000
1000	10	100	1000
10000	10	100	1000
20000	10	100	1000
30000	10	100	1000
40000	10	100	1000
50000	10	100	1000
60000	10	100	1000
80000	10	100	1000

测试代码

处理post请求，延时3s返回结果。flask启动20个进程。fastapi启动一个进程。

## flask 代码
# coding: utf-8
from gevent import monkey
from gevent.pywsgi import WSGIServer
import requests
import datetime
import os
from multiprocessing import cpu_count, Process
from flask import Flask, jsonify,request
import json
import traceback
import importlib
from loguru import logger
import time
app = Flask(__name__)
# 执行run方法
@app.route("/cppla1", methods=['POST', 'GET'])
def cppla1():
    data = request.json
    time.sleep(3)
    return data
# 启动监听ip、端口
def run(MULTI_PROCESS):
    if MULTI_PROCESS == False:
        WSGIServer(('0.0.0.0', 8081), app).serve_forever()
    else:
        mulserver = WSGIServer(('0.0.0.0', 8081), app)
        mulserver.start()
        def server_forever():
            mulserver.start_accepting()
            mulserver._stop_event.wait()
       # for i in range(cpu_count()):
        for i in range(20):
            logger.info('启动进程第几个:{}', i)
            p = Process(target=server_forever)
            p.start()
if __name__ == "__main__":
    # 单进程 + 协程
    # run(False)
    # 多进程 + 协程
    log_init()
    run(True)

## fastapi
# coding: utf-8
# import web framework
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
# import base lib
import datetime
import os
import requests
import json
import traceback
import importlib
from loguru import logger
import time
app = FastAPI()
@app.post("/cppla1")
def function_benchmark(data:dict):
    time.sleep(3)
    return {"item": data}
# 启动监听ip、端口
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8081)

测试结果 框架类型请求总数每次并发数耗时(s)每次并发数耗时(s)每次并发数耗时(s)

fastapi	100	10	33.119	100	12.148	1000	ab命令不支持
flask	100	10	45.088	100	81.106	1000	ab命令不支持
fastapi	1000	10	304.057	100	78.283	1000	78.631
flask	1000	10	327.472	100	198.273	1000	303.442
fastapi	10000	10	x	100	754.296	1000	757.719
flask	10000	10	x	100	1550.119	1000	1970.427
fastapi	20000	10	x	100	x	1000	x
flask	20000	10	x	100	x	1000	x
fastapi	30000	10	x	100	x	1000	x
flask	30000	10	x	100	x	1000	x
fastapi	40000	10	x	100	x	1000	x
flask	40000	10	x	100	x	1000	x
fastapi	50000	10	x	100	x	1000	x
flask	50000	10	x	100	x	1000	x
fastapi	60000	10	x	100	x	1000	x
flask	60000	10	x	100	x	1000	x
fastapi	80000	10	x	100	x	1000	x
flask	80000	10	x	100	x	1000	x