一、asyncio

1.1 概述

asyncio 是 Python 标准库中用于编写并发代码的模块，使用 async/await 语法。它是构建高性能异步 I/O、事件循环、协程和任务的基础库，特别适合处理 I/O 密集型和高并发的网络应用。

• 适用场景：网络请求、Web 服务器、爬虫、数据库连接、消息队列等 I/O 密集型任务
• 不适用场景：CPU 密集型计算(应使用 multiprocessing)

1.2 核心概念

1. 协程 (Coroutine)

协程是可以暂停和恢复执行的函数,使用 async def 定义。

[普通函数] 总耗时: 3.00s

import time

names = ["任务A", "任务B", "任务C"]

# ===== 普通函数版本:for 循环串行执行 =====
def task_sync(name):
    print(f"  {name} 开始")
    time.sleep(1)               # 阻塞 1 秒,整个线程被卡住
    print(f"  {name} 完成")

start = time.time()
for name in names:              # 一个接一个执行,共 3 秒
    task_sync(name)
print(f"[普通函数] 总耗时: {time.time() - start:.2f}s")


"""
  任务A 开始
  任务A 完成
  任务B 开始
  任务B 完成
  任务C 开始
  任务C 完成
[普通函数] 总耗时: 3.00s
"""

[协程函数] 总耗时: 1.01s

import time
import asyncio

names = ["任务A", "任务B", "任务C"]

# ===== 协程版本:for 循环创建任务,gather 并发执行 =====
async def task_async(name):
    print(f"  {name} 开始")
    await asyncio.sleep(1)      # 暂停 1 秒,让出控制权,其他协程趁机运行
    print(f"  {name} 完成")

async def main():
    start = time.time()
    tasks = [task_async(name) for name in names]   # for 循环生成协程列表
    await asyncio.gather(*tasks)                   # 解包后并发执行
    print(f"[协程函数] 总耗时: {time.time() - start:.2f}s")

asyncio.run(main())

"""
  任务A 开始
  任务B 开始
  任务C 开始
  任务A 完成
  任务B 完成
  任务C 完成
[协程函数] 总耗时: 1.01s
"""

async def say_hello(name, delay):
    print(f"  Hello, {name}! (将等待 {delay}s)")
    await asyncio.sleep(delay)
    print(f"  {name} 完成")
    return f"{name}-done"

# <class 'coroutine'> , 不执行函数体
print(type(say_hello("melong", 1)))

为什么协程更快

对比项	普通函数 (time.sleep)	协程 (await asyncio.sleep)
等待时的状态	整个线程被阻塞,啥也干不了	让出控制权,事件循环去跑别的协程
3 个任务并发	不可能,只能串行 → 3 秒	gather 同时调度 → 1 秒
输出顺序	严格按调用顺序:A完→B完→C完	三个"开始"先后打印,然后三个"完成"几乎同时打印

关键点:协程不会让单个 I/O 操作本身变快(等 1 秒还是 1 秒),但它能让"等待时间互相重叠"。3 个任务串行要 3 秒,并发只要 1 秒——这就是 asyncio 的核心价值。

延伸:如果换成 100 个网络请求,普通函数要 100×单次耗时,协程依然只需要约 1 个最慢请求的时间。这就是为什么爬虫和 API 网关都爱用 asyncio。

对比项	普通函数	协程函数
定义	def	async def
调用结果	立即执行函数体	返回协程对象,不执行
执行方式	直接调用	必须由事件循环驱动 (await / asyncio.run)
中途阻塞	整个线程被卡住	遇到 await 时让出控制权,事件循环可去做别的事

怎么个"暂停和恢复"法

关键在于 await 这个关键字,它是协程的"暂停点"。

async def hello():
    print("Hello")              # ① 执行
    await asyncio.sleep(1)      # ② 这里"暂停":告诉事件循环"我要等 1 秒,你先去忙别的"
    print("World")              # ③ 1 秒后,事件循环把控制权交还,继续执行

执行流程示意:

时间线:
0.0s  hello() 开始执行 -> 打印 "Hello"
      ↓
0.0s  遇到 await asyncio.sleep(1)
      ↓ 【暂停】hello 把自己"挂起",控制权交回事件循环
      ↓ 事件循环此时可以去运行其他协程(如果有的话)
      ↓
1.0s  sleep(1) 计时完成,事件循环【恢复】hello 的执行
      ↓
1.0s  打印 "World",hello() 结束

核心机制:协程通过 await 主动"让出" CPU,由事件循环统一调度。这就是单线程也能处理上万并发连接的原因——遇到 I/O 等待时,不傻等,而是切换去处理别的协程。

类比:普通函数像"打电话"——你打过去对方没接,你就一直拿着话筒等;协程像"发微信"——发完消息后你可以去做别的事,对方回复了系统再通知你。

async vs await:什么时候用哪个

async 用来"声明"(标记这是异步的)，await 用来"等待"(真正暂停拿结果)。

async 的三种用法 —— 都是"定义/声明"

写法	用途	例子
async def 函数()	声明一个协程函数	async def fetch(): ...
async with 对象	使用异步上下文管理器(对象有 __aenter__/__aexit__)	async with aiohttp.ClientSession() as s:
async for 元素 in 可迭代对象	异步迭代(对象有 __aiter__/__anext__)	async for line in stream:

记忆点:async 修饰的是"结构"(函数定义、with 块、for 循环),它本身不会让代码暂停,只是告诉 Python “这块东西支持异步”。

await 只有一种用法 —— “等待一个可等待对象”

result = await 可等待对象

可等待对象包括三种:

• 协程 (coroutine):await fetch_data()
• Task:await task1
• Future:await future

记忆点:await 是真正的暂停点,程序运行到这里会挂起协程,等异步操作完成后再恢复。

三条铁律

• 想定义异步函数 → async def(没有 async,函数就是普通同步函数)
• 想等异步结果 → await(没有 await,协程不会被执行,只会得到一个协程对象)
• await 只能写在 async def 内部(写在普通函数里会报 SyntaxError)

一段代码看懂所有用法

import asyncio
import aiohttp                    # 第三方异步 HTTP 库

async def fetch_one(session, url):       # ① async def — 声明协程函数
    async with session.get(url) as resp: # ② async with — 进入异步上下文(连接管理)
        return await resp.text()         # ③ await — 等待响应文本(I/O 暂停点)

async def main():                        # ① async def
    async with aiohttp.ClientSession() as session:  # ② async with
        urls = ["https://a.com", "https://b.com"]

        # 单个 await:等一个协程
        html = await fetch_one(session, urls[0])    # ③ await

        # 用 await + gather 并发等多个协程
        htmls = await asyncio.gather(               # ③ await
            *(fetch_one(session, u) for u in urls)
        )

        # async for:异步遍历(假设 stream 是异步可迭代对象)
        # async for chunk in stream:                # ④ async for
        #     print(chunk)

asyncio.run(main())   # 启动事件循环执行顶层协程

常见错误对照

# ❌ 错误 1:用 def 定义协程
def fetch():               # 这是普通函数!
    await asyncio.sleep(1) # SyntaxError: 'await' outside async function

# ✅ 改成 async def
async def fetch():
    await asyncio.sleep(1)


# ❌ 错误 2:忘记 await
async def main():
    asyncio.sleep(1)       # 警告:coroutine was never awaited
                           # 协程对象被创建但从未执行

# ✅ 加上 await
async def main():
    await asyncio.sleep(1)


# ❌ 错误 3:对非可等待对象 await
async def main():
    await 42               # TypeError: object int can't be used in 'await' expression

# ✅ 只能 await 协程 / Task / Future
async def main():
    await asyncio.sleep(0.1)

一张速查表

场景	该用什么
我要定义一个能暂停的函数	async def
我要调用一个协程并拿结果	await coro()
我要进入一个异步资源(连接/文件)并自动关闭	async with
我要异步遍历数据流(WebSocket、流式响应)	async for
我在普通函数里想拿协程结果	用 asyncio.run(coro()) —— 此时不需要 await

一句话总结:
async 给"东西"贴异步标签(函数/with/for),
await 真正去等结果。
声明用 async,等待用 await。

2. 事件循环 (Event Loop)

事件循环是 asyncio 的核心调度器,负责调度和执行协程。可以把它想象成一个"无限循环的工头",不断从任务队列里挑选可以运行的协程,推进它们的执行。

loop = asyncio.get_event_loop()        # ① 获取(或创建)当前线程的事件循环对象
loop.run_until_complete(hello())       # ② 把协程交给事件循环,直到它跑完才返回
loop.close()                           # ③ 关闭事件循环,释放资源

┌─────────────────────────────────────────┐
│           事件循环 Event Loop             │
│  ┌───────────────────────────────────┐  │
│  │ 1. 从就绪队列取出一个可运行的协程    │  │
│  │ 2. 运行协程,直到遇到 await         │  │
│  │ 3. 把协程挂起,登记它在等什么(I/O)  │  │
│  │ 4. 检查 I/O 是否就绪、定时器是否到期 │  │
│  │ 5. 把就绪的协程放回就绪队列          │  │
│  │ 6. 回到步骤 1                      │  │
│  └───────────────────────────────────┘  │
└─────────────────────────────────────────┘

Python 3.7+ 推荐使用 asyncio.run(),它会自动帮你完成"创建 → 运行 → 关闭"三步,避免忘记 close() 导致的资源泄漏。

# 旧写法(不推荐,容易忘记 close)
loop = asyncio.get_event_loop()
loop.run_until_complete(hello())
loop.close()

# 新写法(推荐)— 一行搞定
asyncio.run(hello())

3. 任务 (Task)

Task 是对协程的封装,用于在事件循环中并发执行多个协程。
用 create_task 把协程包装成 Task 后,它会被立即提交给事件循环开始运行,然后你可以继续做别的事:

import asyncio
import time

# say_hello 是一个简单协程:打印问候语,等待 delay 秒,再打印完成提示
async def say_hello(name, delay):
    print(f"  Hello, {name}! (将等待 {delay}s)")
    await asyncio.sleep(delay)
    print(f"  {name} 完成")
    return f"{name}-done"

# ✅ 并发执行,总耗时 max(1, 2) = 2 秒
async def mai2n():
    start = time.time()
    task1 = asyncio.create_task(say_hello("Alice", 1))  # 立即开始跑
    task2 = asyncio.create_task(say_hello("Bob", 2))  # 立即开始跑
    await task1  # 等 task1 完成(此时 task2 已经在并发运行)
    await task2  # 等 task2 完成
    print(f"总耗时: {time.time() - start:.2f}s")

asyncio.run(main2())
"""
  Hello, Alice! (将等待 1s)
  Hello, Bob! (将等待 2s)
  Alice 完成
  Bob 完成
总耗时: 2.02s
"""

# ❌ 串行执行,总耗时 1 + 2 = 3 秒
async def main2():
    start = time.time()
    await say_hello("Alice", 1)    # 等 1 秒
    await say_hello("Bob", 2)      # 再等 2 秒
    print(f"总耗时: {time.time() - start:.2f}s")
asyncio.run(main2())
"""
  Hello, Alice! (将等待 1s)
  Alice 完成
  Hello, Bob! (将等待 2s)
  Bob 完成
总耗时: 3.02s
"""

关键区别

写法	行为	总耗时
await coro()	立即执行并等待完成,串行	各协程耗时之和
asyncio.create_task(coro())	立即调度,后台并发运行,稍后用 await task 拿结果	最长协程的耗时

一句话总结:await 是"等",create_task 是"先开火再等"。要并发,就用 Task。

4. Future

Future 表示一个尚未完成的异步操作的最终结果,是比 Task 更底层的概念。

• Task 实际上是 Future 的子类,专门用于包装协程
• Future 通常由底层库(如网络库、数据库驱动)创建并返回,你只需要 await 它即可
• 日常应用代码中很少需要直接创建 Future

# 简单理解:Future 像一个"占位符",代表"将来会有的结果"
future = loop.create_future()         # 底层库可能这样创建
# ... 某个回调里:
future.set_result("数据已就绪")        # 设置结果
# 你的代码里:
result = await future                 # 拿到 "数据已就绪"

新手记忆口诀:
协程是"任务的描述",
Task 是"在跑的任务",
Future 是"等结果的凭证",
事件循环是"调度一切的工头"。

---

1.3 常用 API

API	说明
asyncio.run(coro)	运行最顶层协程,自动管理事件循环
asyncio.create_task(coro)	将协程包装为 Task 并立即调度
asyncio.gather(*coros)	并发运行多个协程,收集结果
asyncio.wait(tasks)	等待多个任务完成,可设置超时
asyncio.sleep(seconds)	异步休眠(非阻塞)
asyncio.wait_for(coro, timeout)	给协程设置超时
asyncio.shield(coro)	保护协程免受取消
asyncio.Queue	异步队列
asyncio.Lock / Semaphore / Event	异步同步原语

---

1.4 典型示例

示例 1:并发执行多个任务

import asyncio
import time

async def fetch(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 完成")
    return f"{name} 的结果"

async def main():
    start = time.time()
    results = await asyncio.gather(
        fetch("任务A", 2),
        fetch("任务B", 3),
        fetch("任务C", 1),
    )
    print(f"所有结果: {results}")
    print(f"总耗时: {time.time() - start:.2f}s")

asyncio.run(main())

输出说明:三个任务并发执行,总耗时约 3 秒(最长任务的时间),而非 6 秒。

示例 2:超时控制

async def long_task():
    await asyncio.sleep(10)
    return "完成"

async def main():
    try:
        result = await asyncio.wait_for(long_task(), timeout=2)
    except asyncio.TimeoutError:
        print("任务超时")

asyncio.run(main())

示例 3:异步 HTTP 请求(配合 aiohttp)

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://www.python.org",
        "https://www.github.com",
        "https://www.google.com",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, html in zip(urls, results):
            print(f"{url}: {len(html)} 字节")

asyncio.run(main())

示例 4:异步队列(生产者-消费者模型)

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        print(f"生产: {i}")
        await asyncio.sleep(0.5)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"消费: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    prod = asyncio.create_task(producer(queue))
    cons = asyncio.create_task(consumer(queue))
    await prod
    await queue.join()
    cons.cancel()

asyncio.run(main())

1.5 同步原语

asyncio 提供了与 threading 类似的同步原语,但用于协程之间。

# 锁
lock = asyncio.Lock()
async with lock:
    # 临界区
    pass

# 信号量(限制并发数)
sem = asyncio.Semaphore(10)
async with sem:
    await do_something()

# 事件
event = asyncio.Event()
event.set()
await event.wait()

---

1.6 协程 vs 线程 vs 进程

特性	协程 (asyncio)	线程 (threading)	进程 (multiprocessing)
切换开销	极低	中等	高
内存占用	极小	较小	较大
并发数量	数千~数万	数百	数十
GIL 影响	不受影响(单线程)	受影响	不受影响
适用场景	I/O 密集	I/O 密集	CPU 密集
调试难度	中等	较难(竞态)	较难(IPC)

---

1.7 常见陷阱与最佳实践

1. 不要在协程中调用阻塞函数

# 错误:time.sleep 是阻塞的,会卡住整个事件循环
async def bad():
    time.sleep(1)

# 正确:使用 asyncio.sleep
async def good():
    await asyncio.sleep(1)

2. 注意异常处理

async def main():
    try:
        await asyncio.gather(task1(), task2())
    except Exception as e:
        print(f"出错: {e}")

# 或者使用 return_exceptions=True 收集所有异常
results = await asyncio.gather(task1(), task2(), return_exceptions=True)

3. 不要忘记 await

# 错误:协程未被等待,不会执行
async def bad():
    asyncio.sleep(1)  # 警告:coroutine was never awaited

# 正确
async def good():
    await asyncio.sleep(1)

4. 优雅地取消任务

task = asyncio.create_task(long_running())
await asyncio.sleep(1)
task.cancel()
try:
    await task
except asyncio.CancelledError:
    print("任务已取消")

1.8 常用第三方异步库

库	用途
aiohttp	异步 HTTP 客户端/服务器
aiofiles	异步文件操作
aiomysql / asyncpg	异步数据库驱动
aioredis	异步 Redis 客户端
httpx	同步/异步 HTTP 客户端
FastAPI	基于 asyncio 的现代 Web 框架
Tornado	异步 Web 框架
Sanic	高性能异步 Web 框架