初识FASTAPI以及异步编程

一.异步编程1.为什么要异步?Web服务大多都在等待数据库网络文件读写采用异步模型可以极大提升并发能力。2.核心语法async/awaitasync def定义一个协程函数调用它返回一个协程对象不会立即执行函数体。await只能在async def内部使用表示挂起当前协程等待一个异步操作完成在此期间事件循环可以去执行其他任务。import asyncio async def greet(): print(Hello...) # 异步等待2秒不阻塞事件循环 await asyncio.sleep(2) print(...World!) # asyncio.run() 负责创建事件循环并运行协程 asyncio.run(greet())1.进程、线程、协程的对比1.1. 基础定义1.进程process进程是操作系统资源分配的最小单位。每个进程独立拥有内存、文件句柄、CPU 时间片、堆、全局变量进程间完全隔离。切换代价极高切换时要刷新页表、缓存、保存完整上下文。例子打开的 Chrome、微信、Python 解释器都是独立进程。通信方式管道、socket、共享内存、消息队列成本高。2. 线程Thread进程内CPU 调度最小单位依附进程存在。同一进程内所有线程共享进程资源全局内存、文件、网络每个线程独有栈、寄存器。切换代价远小于进程只切换少量寄存器不用换内存空间。线程安全问题共享资源会竞争需要锁mutex。分类内核线程操作系统调度Python threading、Java Thread用户线程用户态调度协程本质GIL 问题Python 特有CPython 全局解释锁同一时刻一个进程只有 1 个线程执行 CPU 运算多线程适合 IO 密集不适合计算密集。不依赖操作系统内核切换只保存少量局部上下文切换成本极低。单线程内并发手动交出执行权await/yield无线程切换开销、无锁竞争。不需要操作系统分配栈内存占用极小可同时创建几十万协程。适合高并发 IO 场景网络请求、数据库、文件读写。实现Python asyncio、Go goroutine、Lua 协程。2.2使用场景总结CPU 密集运算→ 多进程 利用多核规避 Python GIL代价是资源消耗大。普通 IO 并发少量任务→ 多线程 编码简单上手快并发量几千以内够用。高并发 IO爬虫、后端接口、网关→ 协程 百万级并发内存占用极低吞吐量最高。3.事件循环Event Loop—— 异步的大脑事件循环是一个无限循环负责调度和执行所有的异步任务。工作流程维护一个就绪队列存放可立即执行的协程/Task。维护一个等待队列存放正在等待 I/O 或定时器的协程。每次循环从就绪队列取出一个任务执行直到遇到await。将该任务放入等待队列并注册回调如定时器到时、网络数据到达。检查等待队列将已就绪的任务移回就绪队列。重复以上步骤直到所有任务完成。示例使用asyncio.create_task()将协程注册到事件循环import asyncio async def task(name: str, delay: int): print(f任务 {name} 开始等待 {delay} 秒) await asyncio.sleep(delay) print(f任务 {name} 完成) return f{name} 的结果 async def main(): # 创建 Task 对象它们会被立即注册到当前运行的事件循环中 t1 asyncio.create_task(task(A, 2)) t2 asyncio.create_task(task(B, 1)) print(主协程任务已创建等待它们完成...) # 等待两个任务结束并获取返回值 result1 await t1 result2 await t2 print(f主协程收到 {result1} 和 {result2}) asyncio.run(main())运行结果任务 A 开始等待 2 秒 任务 B 开始等待 1 秒 主协程任务已创建等待它们完成... 任务 B 完成 任务 A 完成 主协程收到 A 的结果 和 B 的结果关键点create_task使得协程在后台并发运行无需等待前一个完成。总耗时约为 2秒最长任务的时间而非 3 秒。1.4 并发执行asyncio.gathergather同时运行多个可等待对象并返回一个包含所有结果的列表语法更简洁。import asyncio import time async def download(file: str, duration: int): print(f开始下载 {file}预计 {duration} 秒) await asyncio.sleep(duration) print(f{file} 下载完成) return f{file} 内容 async def main(): start time.time() results await asyncio.gather( download(A.mp4, 3), download(B.pdf, 2), download(C.txt, 1) ) end time.time() print(f总耗时: {end - start:.2f} 秒) print(f结果: {results}) asyncio.run(main())输出开始下载 A.mp4预计 3 秒 开始下载 B.pdf预计 2 秒 开始下载 C.txt预计 1 秒 C.txt 下载完成 B.pdf 下载完成 A.mp4 下载完成 总耗时: 3.00 秒 结果: [A.mp4 内容, B.pdf 内容, C.txt 内容]1.5 异步队列asyncio.Queue生产者-消费者模式asyncio.Queue是协程间安全通信的队列常用于解耦生产者和消费者。常用方法方法说明queue asyncio.Queue(maxsizeN)创建队列maxsize为最大容量await queue.put(item)放入数据若队列满则挂起等待await queue.get()取出数据若队列空则挂起等待queue.task_done()标记一个取出的任务已处理完成await queue.join()阻塞直到队列中所有项目都被task_done()完整示例爬虫生产者 存储消费者import asyncio import random queue asyncio.Queue(maxsize5) # 容量为5的队列 async def producer(name: str): 模拟爬虫抓取数据 for i in range(4): await asyncio.sleep(random.uniform(0.3, 1.0)) # 模拟抓取耗时 item f{name}-页面{i} await queue.put(item) print(f[生产者 {name}] 生产: {item}队列大小: {queue.qsize()}) # 生产结束放入结束标记毒丸 await queue.put(None) print(f[生产者 {name}] 结束) async def consumer(): 模拟数据存储 active_producers 2 # 假设有2个生产者 while active_producers 0: item await queue.get() if item is None: active_producers - 1 print(f[消费者] 收到结束信号剩余生产者: {active_producers}) else: await asyncio.sleep(0.2) # 模拟数据库写入 print(f[消费者] 保存数据: {item}) queue.task_done() print([消费者] 所有数据处理完毕) async def main(): p1 asyncio.create_task(producer(站点A)) p2 asyncio.create_task(producer(站点B)) c asyncio.create_task(consumer()) await asyncio.gather(p1, p2) await queue.join() # 等待队列中所有任务被处理 c.cancel() print(主程序全部完成) asyncio.run(main())1.6 同步与异步对比总结维度同步多线程异步协程执行模型多线程并行受 GIL 限制单线程协作式调度无 GIL 竞争资源开销线程切换开销大内存占用高协程切换几乎无开销内存占用极小适用场景CPU 密集型计算I/O 密集型网络、数据库、文件编程复杂度需处理锁、竞态条件逻辑线性但需理解await挂起点Python 实现threading,multiprocessingasyncio2.初识FATAPI2.1 环境准备与第一个应用安装依赖pip install fastapi uvicorn[standard]创建 main.py 文件# 步骤1从 fastapi 包中导入 FastAPI 类 # FastAPI 是一个类我们用它来创建 Web 应用的核心对象 from fastapi import FastAPI # 步骤2创建 FastAPI 类的实例 # 这个 app 对象就是整个 Web 应用后续所有的路由、配置都挂载在它上面 # 括号里的参数是可选的用于给自动生成的 API 文档添加标题、描述和版本号 app FastAPI( title我的 FastAPI 应用, # 文档标题 descriptionDay01 入门示例, # 文档描述 version0.1.0 # 应用版本 ) # 步骤3定义路由 —— 告诉服务器“当有人访问某个 URL 时执行哪个函数” # app.get(/) 是一个装饰器Decorator # 它表示当客户端使用 GET 方法访问根路径 / 时触发下面这个函数 app.get(/) async def read_root(): 这是一个异步的视图函数也叫路径操作函数。 为什么要用 async def - 因为 Web 服务经常需要等待数据库查询、网络请求等 I/O 操作。 - 使用异步函数可以让服务器在等待期间去处理其他请求提高并发能力。 返回值 - 直接返回一个 Python 字典FastAPI 会自动把它转换成 JSON 格式的字符串 并设置响应头的 Content-Type 为 application/json。 return {message: Hello, FastAPI World!}启动服务器在终端中执行以下命令uvicorn main:app --reload 或者 uvicorn main:app --host 127.0.0.1 --port 8000 --reload命令详解uvicornASGI 服务器程序专门用于运行异步 Web 应用。main:appmain指的是main.py文件名app指的是文件里那个app FastAPI()的变量名。中间用冒号分隔。--reload开启热重载模式。当你修改代码并保存后服务器会自动重启开发时必须加上。验证结果打开浏览器访问http://127.0.0.1:8000→ 你会看到 JSON 响应{message:Hello, FastAPI World!}http://127.0.0.1:8000/docs→ 自动生成的 Swagger UI 交互式文档。你可以在这里直接测试接口http://127.0.0.1:8000/redoc→ 另一种风格的文档。为什么 FastAPI 能自动生成文档因为我们在定义函数时使用了 Python 的类型提示Type Hints并且 FastAPI 内部利用这些类型信息生成了 OpenAPI 规范文档再渲染成漂亮的界面。