flask与fastAPI巩固

作者: Hafsun
Icon发表于2026-06-10
Icon更新于2026-06-14
Icon总字数: 2.6k字
Icon阅读时长: 8分钟
Icon访问量:
Icon开发
Icon

异步

并发模型对比

框架 并发方式 原理 适合场景
Flask 多线程/多进程 每个请求新建线程 CPU密集型、阻塞IO
FastAPI 异步事件循环（单线程） await 切换任务 高并发IO操作

flask实现并发大概流程：
假如app.py集成了路由，然后路由下定义了业务逻辑函数，那么当多个请求到来时，各个请求互不影响，app.py会为各个请求创建线程，每个线程各自执行。所以假如业务逻辑函数实现需要3秒，当三个请求同时来到时，总处理时间也是3秒。

对于高并发以及实时场景，更主流的框架是faskAPI，后面简称fast。所以我重点学习fast

下面是app.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get('/sleep')
async def sleep():
    print("任务开始")
    await asyncio.sleep(3)  # 模拟IO等待
    print("任务结束")
    return 'ok'

if __name__ == '__main__':
    import uvicorn  //要用uvicorn启动fast项目，但是官方推荐在命令行用   						//uvicorn app:app --reload 启动
    uvicorn.run(app, host="127.0.0.1", port=8000)

下面是test_fastapi_client.py，用于模拟用户请求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import asyncio
import time
import httpx//支持异步的客户端，与asyncio.gather配合来同时发请求

URL = "http://127.0.0.1:8000/sleep"

async def fetch(client, request_id):
    """异步发起 GET 请求并打印耗时"""
    start_time = time.time()
    response = await client.get(URL)
    end_time = time.time()

    print(f"[请求 {request_id}] 状态码: {response.status_code}, "
          f"响应内容: {response.text}, "
          f"耗时: {end_time - start_time:.2f}秒")

async def main():
    # 使用 httpx.AsyncClient 保持长连接，提升并发性能
    async with httpx.AsyncClient() as client:
        tasks = []
        total_start = time.time()

        # 创建 3 个并发任务
        for i in range(1, 4):
            tasks.append(fetch(client, i))

        # 使用 gather 并发执行所有任务
        await asyncio.gather(*tasks)

        total_end = time.time()
        print("-" * 40)
        print(f"总耗时: {total_end - total_start:.2f}秒")

if __name__ == "__main__":
    asyncio.run(main())

运行后终端打印分别如下：
image-20260610201556812image-20260610201610262

为什么是2，1，3呢？因为逻辑是先创建任务（把任务加到人物列表中），然后并发执行，执行先后顺序多少会有细微不同（随机情况）

为什么要用httpx.AsyncClient 保持长连接

这里涉及到通信问题。首先要说明flask和fast都是用的http协议通信。

在使用普通的同步 HTTP 客户端（如 requests）时，每次发起请求都会经历：建立 TCP 连接 -> 发送数据 -> 接收数据 -> 关闭连接。如果并发发 3 个请求，就会重复这个过程 3 次。

而在 FastAPI 的异步世界中，我们使用 httpx.AsyncClient 作为上下文管理器（async with httpx.AsyncClient() as client:），起到连接复用的功能：它在后台维持底层的 TCP 连接不中断。当连续发出多个请求时，它会直接复用已经建立好的通道，省去了反复握手、建立连接的开销。

fast这个异步模型，基于事件循环来工作。当遇到await操作时，当前协程会主动让出事件循环的控制权，使其立马去处理其它的http请求，等await操作完成了再回来继续执行下一步。这里可以理解为主厨同时炒多个菜，然后到处走动

注意点：

虽然这里例子中fast用的是http协议，但是它也原生支持websocket协议，用于实时通话等场景：

1
2
3
4
5
6
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

fast有个优点就是可以自动生成接口文档，假如项目运行在localhost:9999，那么localhost:9999/doc就是文档界面。

TCP协议

这就是著名的三次握手、四次挥手协议

我们可以把它想象成两个人（客户端 A 和服务器 B）在嘈杂的环境中打电话：

  • 第一次握手（A -> B）： A 拨通电话说：“喂，你能听到我说话吗？”（这在技术上叫发送 SYN 报文，表示请求建立连接）。
  • 第二次握手（B -> A）： B 听到了，回复道：“我能听到你说话！那你能听到我说话吗？”（技术上叫返回 SYN + ACK，表示同意并反问确认）。
  • 第三次握手（A -> B）： A 回答道：“我也能听到你说话！”（技术上叫发送 ACK，表示最终确认）。

既然有开始，就有结束。当数据传输完毕，不需要再聊的时候，就需要断开连接。因为 TCP 是双向的（既能发又能收），所以关闭连接比建立连接多了一步，被称为四次挥手：

  • A 说：“我说完了。”（FIN）
  • B 回：“好的，我知道了。”（ACK）
  • B 接着说：“我也说完了。”（FIN）
  • A 回：“好的，那就挂了吧。”（ACK）

协议套娃

TCP协议是HTTP协议和Websocket协议的底层协议。这里联系到协议套娃：

在计算机里，当你通过浏览器访问一个网页时，数据是这样一层层往下传的：

  • 应用层 (HTTP)： 浏览器说：“我要请求 /sleep 接口！”（写好信件内容）
  • 传输层 (TCP)： 操作系统接过信件，把它塞进信封，贴上收发件人地址，并准备发起“三次握手”确认通道。（安排物流运输）
  • 网络层 (IP)： 邮政局看着信封上的地址，决定这封信要经过哪几个城市的转运中心才能到达对方电脑。（规划跨城市运输路线）
  • 物理/链路层 (网线/Wi-Fi)： 最终变成电信号或无线电波，通过网线和路由器发送出去。（货车开上高速公路）

当数据到达服务器时，再反过来一层层拆开（解封装），最后把 HTTP 的内容交给 FastAPI 或 Flask 处理。

ORM（对象关系映射）

对于fast,官方给的orm是sqlalchemy（之前有过学习），今天学习Tortoise,受Djiango启发。

fast的ORM操作，主要参考

1、创建模型

以选课系统为例：创建个模型类文件 models.py，每个类对应一张表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
#导入tortoise

from tortoise.models import Model
from tortoise import fields


#创建班级类
class Clas(Model):
    name = fields.CharField(max_length=255, description='班级名称')


#创建老师类
class Teacher(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255, description='姓名')
    tno = fields.IntField(description='账号')
    pwd = fields.CharField(max_length=255, description='密码')


#课程表
class Course(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255, description='课程名')
    teacher = fields.ForeignKeyField('models.Teacher', related_name='courses', description='课程讲师')



#创建学生类
class Student(Model):
    id = fields.IntField(pk=True)
    sno = fields.IntField(description='学号')
    #description 在接口文档有个显示
    pwd = fields.CharField(max_length=255, description='密码')
    name = fields.CharField(max_length=255, description='姓名')
    # 一对多，反向查询时使用related_name
    clas = fields.ForeignKeyField('models.Clas', related_name='students')
    # 多对多
    courses = fields.ManyToManyField('models.Course', related_name='students',description='学生选课表')

2、创建数据库连接配置文件

创建settings.py 配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
TORTOISE_ORM = {
    'connections': {
        'default': {
            # 'engine': 'tortoise.backends.asyncpg',  PostgreSQL
            'engine': 'tortoise.backends.mysql',  # MySQL or Mariadb
            'credentials': {
                'host': '10.10.0.52',
                'port': '3306',
                'user': 'root',
                'password': 'Jingxxxxxxxx',
                'database': 'fastapi',
                'minsize': 1,
                'maxsize': 5,
                'charset': 'utf8mb4',
                "echo": True
            }
        },
    },
    'apps': {
        'models': {
            #这个models就是自己配置的models.py位置

            'models': ['models'],
            'default_connection': 'default',

        }
    },
    'use_tz': False,
    'timezone': 'Asia/Shanghai'
}

3、启动项目

app.py启动项目

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from fastapi import FastAPI  # FastAPI 是一个为你的 API 提供了所有功能的 Python 类。
import uvicorn
from tortoise.contrib.fastapi import register_tortoise
from settings import TORTOISE_ORM

#创建应用程序，app是应用程序名
app = FastAPI()  # 这个实例将是创建你所有 API 的主要交互对象。这个 app 同样在如下命令中被 uvicorn 所引用

# 该方法会在fastapi启动时触发，内部通过传递进去的app对象，监听服务启动和终止事件
# 当检测到启动事件时，会初始化Tortoise对象，如果generate_schemas为True则还会进行数据库迁移
# 当检测到终止事件时，会关闭连接
register_tortoise(
    app,
    #数据库配置信息
    config=TORTOISE_ORM,
    # generate_schemas=True,  # 如果数据库为空，则自动生成对应表单，生产环境不要开
    # add_exception_handlers=True,  # 生产环境不要开，会泄露调试信息
)

if __name__ == '__main__':
    #注意，run的第一个参数 必须是文件名:应用程序名
    uvicorn.run("quickstart:app", port=8080,  reload=True, workers=1)

正常启动后便连接到了数据库

4、根据模型类创建数据库表

这里依靠ORM迁移工具aerich，结合tortoise异步orm框架使用。

迁移工具可以理解为核心功能是生成和执行数据库迁移脚本（SQL），实现数据库的增删改（没有查，并非少打字）功能，然后异步可以理解为aerich和tortoise是高度绑定的，它们共用配置文件，当执行查询等操作时，这些I/O操作都是非阻塞的。

回到操作步骤，这一步骤分多个小步骤，大概是初始化配置、然后迁移操作、回退操作（类似git）等，都是用命令实现，命令语句和大多工具的用法差不多。具体可参考原文。

查询功能

在分发的路由中的路由函数中，把sql的查询语句写成查询函数：

all查询：
image-20260611154900364

循环打印：

image-20260611154923544

静态页面

就是把查询到的信息，以你喜欢的方式，更好看的呈现在浏览器上。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#导入模板的包
from fastapi.templating import Jinja2Templates

# 实例化Jinja2对象，并将文件夹路径设置为以templates命名的文件夹
templates = Jinja2Templates(directory="templates")

接口：
@api_student.get("/index")
async def show_student(request:Request):
    students = await Student.all()
    return templates.TemplateResponse(
        'index.html', #第一个参数放模板文件
        {
            'request': request,  # 注意，返回模板响应时，必须有request键值对，且值为Request请求对象
            'students':students

        }, #context上下文对象，是个字典
    )

image-20260611154944909

启动程序后，浏览器访问：
image-20260611155001817

Icon
致谢名单
本作品由 Hafsun 于 2026-06-10 19:24:11 发布
作品地址：flask与fastAPI巩固
除特别声明外，本站作品均采用 CC BY-NC-SA 4.0 许可协议，转载请注明来自 欢迎回家
Logo
上一篇肖佬经验之谈下一篇做一张行测卷