【FastAPI筑基-Day1】async/await 异步编程 + aiohttp 并发请求实战｜新手零基础版

专栏：FastAPI 从零到后端接口实战
标签：Python asyncio 异步IO aiohttp 网络爬虫 FastAPI前置基础

---

一、写在前面：为什么要学异步编程？

在 AI 时代，Python 后端几乎成了模型部署的首选——无论是用 FastAPI 封装一个 ChatGPT 对话接口，还是搭建 Stable Diffusion 的出图服务，都离不开 FastAPI。而 FastAPI 之所以能做到超高并发性能，底层依托的是 Python 原生 asyncio 异步框架。简单说：异步编程就是让程序在等待 IO（网络请求、文件读写）时不干等着，而是去干别的事，效率翻倍。

Day1 不用啃理论，三个目标立等可取：

1. 认清同步短板 —— 为什么同步代码慢？
2. 掌握异步语法 —— async/await 怎么写？
3. 上手并发请求 —— 用 aiohttp 同时抓多个网页

全程 可运行代码 + 直观效果对比，零基础照着敲就能跑通。

---

二、同步请求：一次只做一件事

2.1 运行逻辑

同步代码从上到下顺序执行。发起网络请求时，程序会阻塞等待服务器响应，一个请求结束，才会执行下一个请求。

请求1 ──等待──▶ 拿到结果 ──▶ 请求2 ──等待──▶ 拿到结果 ──▶ 请求3 ──等待──▶ 拿到结果
         ↑ 这段时间 CPU 在摸鱼

批量访问多个网站，总耗时 = 所有网站响应时间相加。

2.2 同步请求代码

import requests
import time


def sync_get_html(url):
    """同步获取网页内容"""
    response = requests.get(url)
    print(f"同步访问 {url}，页面字符总数：{len(response.text)}")


if __name__ == "__main__":
    start = time.time()

    sync_get_html("https://www.baidu.com")
    sync_get_html("https://www.qq.com")
    sync_get_html("https://www.163.com")

    end = time.time()
    print(f"同步执行总共花费：{round(end - start, 2)} 秒")

2.3 运行效果（示例）

同步访问 https://www.baidu.com，页面字符总数：2443
同步访问 https://www.qq.com，页面字符总数：327
同步访问 https://www.163.com，页面字符总数：146
同步执行总共花费：0.33 秒

三个网站一个接一个请求，总耗时是各自耗时之和。实际数值因网络环境和网站内容变化会有所不同。

---

三、异步核心语法：四个关键字搞定

3.1 关键字速查表

关键字	作用	注意事项
async def	定义一个异步函数（协程函数）	普通调用方式（不加 await）只能拿到协程对象，不会执行内部逻辑
await	等待一个 IO 任务完成，等待期间释放 CPU 去执行其他任务	只能写在 async def 修饰的函数内部
asyncio.run()	异步程序的统一入口，自动创建并销毁事件循环	Python 3.7+ 官方推荐写法
asyncio.gather()	接收多个协程对象，一次性并发执行	常用于批量任务调度

3.2 极简测试：感受异步

import asyncio
import time


async def test_task(task_id):
    """模拟一个异步任务"""
    print(f"[{time.strftime('%H:%M:%S')}] 任务 {task_id} 开始运行")
    await asyncio.sleep(2)  # 模拟网络等待耗时（非阻塞）
    print(f"[{time.strftime('%H:%M:%S')}] 任务 {task_id} 执行完成")


async def main():
    # 并发执行 3 个任务
    await asyncio.gather(
        test_task(1),
        test_task(2),
        test_task(3),
    )


if __name__ == "__main__":
    start = time.time()
    asyncio.run(main())
    print(f"总耗时：{round(time.time() - start, 2)} 秒")

3.3 运行效果

[17:13:31] 任务 1 开始运行
[17:13:31] 任务 2 开始运行
[17:13:31] 任务 3 开始运行
[17:13:33] 任务 1 执行完成
[17:13:33] 任务 2 执行完成
[17:13:33] 任务 3 执行完成
总耗时：2.01 秒

三个任务几乎同时开始，每个都等 2 秒，但总耗时只有 2 秒左右——这就是异步的魅力。

---

四、aiohttp：专为异步而生的 HTTP 库

requests 是同步库，内部会阻塞线程，无法在异步代码中高效使用。aiohttp 则是适配 asyncio 的异步 HTTP 客户端，和 await 配合天衣无缝。

安装

pip install aiohttp

固定写法：async with（异步上下文管理器）

async with aiohttp.ClientSession() as session:
    async with session.get(url) as response:
        html = await response.text()

async with 会自动创建和释放网络会话/连接，不会出现连接泄露，正式项目统一使用这种写法。

---

五、入门练习：单个网址异步请求

先跑通最简单的异步请求，感受一下 async/await 的完整写法。

import asyncio
import aiohttp


async def fetch_one_url(url):
    """异步请求单个网址"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as res:
            html = await res.text()          # 读取网页内容属于 IO 操作，必须加 await
            print(f"访问地址：{url}，网页长度：{len(html)}")


async def main():
    await fetch_one_url("https://www.baidu.com")


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

运行结果：

访问地址：https://www.baidu.com，网页长度：28918

---

六、核心实战：多站点并发抓取

利用 asyncio.gather() 批量调度任务，所有请求近乎同时发出，整体耗时约等于最慢那个网站的响应时间。同时加上以下工程化处理：

• 异常捕获：单个网址访问失败不会导致整个程序崩溃
• 超时控制：防止某个请求卡死影响整体

import asyncio
import aiohttp
import time


async def crawl_url(session, url):
    """异步爬取单个 URL，带异常处理和超时控制"""
    try:
        # 设置 10 秒超时，防止某个网站卡死
        timeout = aiohttp.ClientTimeout(total=10)
        async with session.get(url, timeout=timeout) as resp:
            html_content = await resp.text()
            print(f"✅ 请求成功｜{url} 数据大小：{len(html_content)} 字符")
    except asyncio.TimeoutError:
        print(f"⏰ 请求超时｜{url}，超过 10 秒未响应")
    except Exception as error:
        print(f"❌ 请求失败｜{url}，错误详情：{error}")


async def batch_crawl(url_list):
    """批量并发调度，共享同一个 ClientSession 以复用连接"""
    async with aiohttp.ClientSession() as session:
        # 为每个 url 创建一个协程任务，传入共享 session
        tasks = [crawl_url(session, url) for url in url_list]
        # * 解包列表，并发执行所有协程
        await asyncio.gather(*tasks)


if __name__ == "__main__":
    target_urls = [
        "https://www.baidu.com",
        "https://www.qq.com",
        "https://www.163.com",
    ]

    start = time.time()
    asyncio.run(batch_crawl(target_urls))
    total_time = round(time.time() - start, 2)
    print(f"\n🎯 所有任务执行结束，总耗时：{total_time} 秒")

运行效果对比

=== 同步方式 ===
同步访问 https://www.baidu.com，页面字符总数：2443
同步访问 https://www.qq.com，页面字符总数：327
同步访问 https://www.163.com，页面字符总数：146
同步执行总共花费：0.33 秒

=== 异步方式 ===
✅ 请求成功｜https://www.163.com 数据大小：146 字符
✅ 请求成功｜https://www.baidu.com 数据大小：28918 字符
✅ 请求成功｜https://www.qq.com 数据大小：118550 字符
🎯 所有任务执行结束，总耗时：0.16 秒

三个请求同时发出，总耗时 ≈ 最慢那个网站的响应时间，异步耗时约为同步的 一半。需要说明的是：requests 和 aiohttp 发送的默认请求头不同，同一网站返回的内容长度可能不同，这是正常现象；本例中同步与异步在同一环境下先后运行，数据具有可比性。

---

七、同步 vs 异步：一张表看懂核心区别

维度	同步（requests）	异步（aiohttp）
执行方式	串行排队，一个接一个	并发执行，同时发出多个请求
IO 等待时	CPU 闲置，干等	切换去执行其他任务
总耗时	所有请求耗时之和	约等于最慢请求的耗时
适用场景	少量请求、简单脚本	爬虫、高并发后端接口
代码复杂度	简单直观	需要理解 async/await，稍复杂
底层库	requests	aiohttp

---

八、Day1 学习检查清单

知识点	是否掌握？
理解同步代码的阻塞问题	□
会用 async def 定义异步函数	□
会用 await 等待异步 IO 操作	□
会用 asyncio.run() 启动异步程序	□
会用 asyncio.gather() 并发执行多个任务	□
理解 async with 的作用	□
能独立编写 aiohttp 异步网络请求代码	□
明白异步 IO 对 FastAPI 高并发的作用	□

---

九、下期预告

Day2：自定义彩色日志输出 + Python 装饰器原理

标准化日志打印，方便后续代码调试排查问题。会涉及：

• logging 模块配置
• 终端彩色输出（ANSI 颜色码）
• 装饰器原理与实战（日志装饰器、计时装饰器）

---