Python 异步编程完全指南

从零开始，一步步掌握 Python 异步编程

阶段一：前置知识储备

1.1 并发 vs 并行

生活类比：

并发

你在厨房一个人做饭：先切菜，等水烧开的时候去炒菜，水开了去关火，然后继续炒菜。任务交替执行，看起来像同时在做。

并行

你和朋友两个人一起做饭：你切菜，朋友炒菜，真正同时进行。

1
2
3
4
5
6
7
并发（单核）：
任务A: ████░░░░████░░░░
任务B: ░░░░████░░░░████

并行（多核）：
任务A: ████████████████
任务B: ████████████████

关键区别：

特性 并发 并行
本质 交替执行 同时执行
硬件要求 单核即可 需要多核
关注点 任务调度 任务执行
记住

并发是逻辑上的”同时”，并行是物理上的”同时”。单核可以实现并发，但只有多核才能实现并行。

1.2 阻塞 vs 非阻塞 / 同步 vs 异步

这两个概念经常被混淆，我们用生活例子来理解：

同步阻塞（傻等）：

1
2
你烧水 → 站在水壶旁边等 → 水开了 → 去切菜
问题：等待期间什么都没干！

同步非阻塞（轮询）：

1
2
你烧水 → 去切菜 → 每隔30秒看一眼水 → 水开了 → 处理
问题：要不断检查，很累！

异步非阻塞（回调通知）：

1
2
你烧水 → 水壶有响铃功能 → 放心去切菜 → 水开了壶自动响 → 去处理
完美：不浪费时间，也不用反复检查！
核心总结

同步是主动等待，异步是通知就绪；阻塞是线程挂起，非阻塞是线程可做其他事。

1.3 进程、线程与协程

用公司来类比：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
进程 = 一家公司
├── 拥有独立的资源（办公室、设备）
├── 创建/销毁成本高
└── 公司之间互不影响

线程 = 公司里的员工
├── 共享公司资源
├── 创建/销毁成本中等
└── 员工之间可能产生冲突（竞争资源）

协程 = 员工的待办清单
├── 不需要额外资源
├── 创建/销毁成本极低
└── 由员工自己决定什么时候切换任务

详细对比：

特性 进程 线程 协程
资源占用 重（独立内存空间） 中（共享内存） 极轻（几KB）
切换成本 高（需要内核调度） 中（需要内核调度） 低（用户态切换）
数据共享 复杂（需要IPC） 简单（共享变量） 简单（共享变量）
编程难度 难 中 易（写起来像同步）
适合场景 CPU密集型 I/O密集型（有限） I/O密集型（高并发）
什么时候选哪个？

CPU 密集型→多进程，I/O 少量连接→多线程，I/O 高并发→协程。

1.4 Python 的 GIL（全局解释器锁）

什么是 GIL？

GIL（Global Interpreter Lock）是 Python 解释器的一个锁，确保同一时刻只有一个线程执行 Python 字节码。

1
2
3
4
5
6
7
8
9
10
11
┌─────────────────────────────────────────┐
│            Python 解释器                 │
│  ┌─────────────────────────────────┐    │
│  │            GIL 锁               │    │
│  └─────────────────────────────────┘    │
│                                         │
│  线程1: ████████░░░░████████░░░░        │
│  线程2: ░░░░████████░░░░████████        │
│                                         │
│  同一时刻只有一个线程在执行！            │
└─────────────────────────────────────────┘
GIL

的影响 GIL 简化了 CPython 实现并提高单线程性能，但多线程对 CPU 密集型任务无效！

GIL 的影响：

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
# CPU 密集型任务：多线程反而更慢！
import threading
import time

def cpu_bound(n):
    """CPU密集型：计算"""
    total = 0
    for i in range(n):
        total += i * i
    return total

# 单线程
start = time.time()
cpu_bound(10**7)
cpu_bound(10**7)
print(f"单线程: {time.time() - start:.2f}s")

# 多线程（受GIL限制，可能更慢）
start = time.time()
t1 = threading.Thread(target=cpu_bound, args=(10**7,))
t2 = threading.Thread(target=cpu_bound, args=(10**7,))
t1.start()
t2.start()
t1.join()
t2.join()
print(f"多线程: {time.time() - start:.2f}s")

为什么异步是 Python 高并发的出路？

1
2
3
4
5
6
7
8
9
传统多线程（受GIL限制）：
- 线程切换有开销
- CPU密集型任务无法并行
- 大量线程消耗系统资源

异步协程（绕过GIL）：
- 协程切换无系统开销
- I/O等待时自动切换，充分利用等待时间
- 单线程可运行成千上万个协程
Python

3.13+ 新变化 Python 3.13 引入实验性的 free-threaded（no-GIL）模式，但仍处于实验阶段，生产环境建议继续使用异步方案。

阶段二：Python 异步基础语法

2.1 定义与运行协程

第一个协程程序：

1
2
3
4
5
6
7
8
9
10
import asyncio

# 用 async def 定义协程函数
async def say_hello():
    print("Hello")
    await asyncio.sleep(1)  # 模拟I/O操作，等待1秒
    print("World")

# 运行协程
asyncio.run(say_hello())

输出：

1
2
3
Hello
（等待1秒）
World

关键语法：

关键字 作用 说明
async def 定义协程函数 普通函数变成协程函数
await 等待异步操作 挂起当前协程，交出控制权
asyncio.run() 运行入口 创建事件循环并运行协程

await 的本质：

1
2
3
4
5
6
7
8
async def fetch_data():
    print("开始获取数据...")
    # 遇到 await，程序会：
    # 1. 挂起当前协程
    # 2. 去执行其他可执行的协程
    # 3. 等 I/O 完成后，回来继续执行
    data = await some_io_operation()
    print(f"获取到数据: {data}")
常见错误

在普通函数中用 await 会 SyntaxError；直接调用协程函数不会执行，返回协程对象。正确做法是用 asyncio.run()。

1
2
3
4
5
6
7
8
9
10
11
12
# ❌ 错误：在普通函数中使用 await
def wrong():
    await asyncio.sleep(1)  # SyntaxError!

# ❌ 错误：直接调用协程函数（不会执行）
async def hello():
    print("hello")

hello()  # 什么都不会发生！返回一个协程对象

# ✅ 正确：使用 asyncio.run() 运行
asyncio.run(hello())

2.2 任务调度

为什么需要 create_task？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# ❌ 顺序执行（没有并发）
async def main():
    await fetch_user()      # 等2秒
    await fetch_orders()    # 等3秒
    await fetch_products()  # 等2秒
# 总共需要 7 秒！

# ✅ 并发执行
async def main():
    # 创建任务（立即开始执行）
    task1 = asyncio.create_task(fetch_user())
    task2 = asyncio.create_task(fetch_orders())
    task3 = asyncio.create_task(fetch_products())
  
    # 等待所有任务完成
    await task1
    await task2
    await task3
# 总共只需要 3 秒（取决于最慢的任务）！
核心区别

直接 await 是顺序执行；create_task + await 是并发执行；gather 是批量并发的快捷方式。

asyncio.gather() - 批量并发：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import asyncio

async def fetch(url, delay):
    print(f"开始请求 {url}")
    await asyncio.sleep(delay)  # 模拟网络请求
    print(f"完成请求 {url}")
    return f"{url} 的数据"

async def main():
    # gather 会并发执行所有任务，并收集结果
    results = await asyncio.gather(
        fetch("api/user", 2),
        fetch("api/orders", 3),
        fetch("api/products", 1),
    )
  
    print(f"所有结果: {results}")

asyncio.run(main())

输出：

1
2
3
4
5
6
7
开始请求 api/user
开始请求 api/orders
开始请求 api/products
完成请求 api/products
完成请求 api/user
完成请求 api/orders
所有结果: ['api/user 的数据', 'api/orders 的数据', 'api/products 的数据']

2.3 休眠与时间控制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import asyncio

async def timer_demo():
    print("开始")
  
    # 异步休眠（不会阻塞事件循环）
    await asyncio.sleep(1)
    print("过了1秒")
  
    # 带返回值的休眠
    result = await asyncio.sleep(2, result="完成")
    print(f"结果: {result}")

async def countdown(n):
    """倒计时示例"""
    for i in range(n, 0, -1):
        print(f"倒计时: {i}")
        await asyncio.sleep(1)
    print("时间到！")

asyncio.run(countdown(5))
永远不要用

time.sleep() 在异步代码中使用 time.sleep() 会阻塞整个事件循环！始终使用 await asyncio.sleep()。

2.4 异步上下文管理器

什么是上下文管理器？

就是 with 语句的异步版本，常用于管理需要打开/关闭的资源（如数据库连接、文件）。

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
import asyncio

class AsyncDatabaseConnection:
    """模拟异步数据库连接"""
  
    async def __aenter__(self):
        """进入 with 块时调用"""
        print("连接数据库...")
        await asyncio.sleep(1)  # 模拟连接耗时
        print("连接成功")
        return self
  
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """离开 with 块时调用"""
        print("关闭数据库连接...")
        await asyncio.sleep(0.5)
        print("连接已关闭")
  
    async def query(self, sql):
        print(f"执行查询: {sql}")
        await asyncio.sleep(1)
        return [{"id": 1, "name": "张三"}]

async def main():
    # 使用 async with 自动管理连接生命周期
    async with AsyncDatabaseConnection() as db:
        result = await db.query("SELECT * FROM users")
        print(f"查询结果: {result}")
  
    # 离开 with 块后，连接自动关闭

asyncio.run(main())

实际应用（aiohttp）：

1
2
3
4
5
6
7
8
9
import aiohttp
import asyncio

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

asyncio.run(fetch_url("https://example.com"))
什么时候用

async with？ 文件读写、数据库连接、HTTP 会话等需要获取和释放资源的场景，都应该使用异步上下文管理器。

2.5 异步迭代器

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
import asyncio

class AsyncCounter:
    """异步计数器"""
  
    def __init__(self, start, end):
        self.start = start
        self.end = end
        self.current = start
  
    def __aiter__(self):
        return self
  
    async def __anext__(self):
        if self.current >= self.end:
            raise StopAsyncIteration
      
        await asyncio.sleep(0.5)  # 模拟异步操作
        value = self.current
        self.current += 1
        return value

async def main():
    # 使用 async for 遍历异步迭代器
    async for num in AsyncCounter(0, 5):
        print(f"数字: {num}")

asyncio.run(main())

输出：

1
2
3
4
5
6
数字: 0
（每0.5秒输出一个）
数字: 1
数字: 2
数字: 3
数字: 4

实际应用（异步读取流数据）：

1
2
3
4
async def read_stream():
    """模拟读取数据流"""
    async for chunk in async_stream_reader:
        process(chunk)

阶段三：深入 asyncio 核心机制

3.1 事件循环

事件循环是 asyncio 的心脏

什么是事件循环？

事件循环是 asyncio 的心脏，它负责：

  1. 监控哪些 I/O 操作完成了
  2. 调度哪些协程可以继续执行
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
事件循环工作流程：
┌─────────────────────────────────────────────────┐
│                   事件循环                        │
│                                                  │
│  ┌──────────────────────────────────────────┐   │
│  │  任务队列                                  │   │
│  │  ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐       │   │
│  │  │任务1│ │任务2│ │任务3│ │任务4│ ...   │   │
│  │  └─────┘ └─────┘ └─────┘ └─────┘       │   │
│  └──────────────────────────────────────────┘   │
│                      ↓                           │
│  ┌──────────────────────────────────────────┐   │
│  │  执行协程                                   │
│  │  遇到 await → 挂起，记录等待的 I/O         │   │
│  └──────────────────────────────────────────┘   │
│                      ↓                           │
│  ┌──────────────────────────────────────────┐   │
│  │  监控 I/O                                   │
│  │  select/poll/epoll 等待 I/O 就绪           │   │
│  │  I/O 完成 → 唤醒对应协程                   │   │
│  └──────────────────────────────────────────┘   │
│                      ↓                           │
│           循环回到任务队列                        │
└─────────────────────────────────────────────────┘

事件循环的使用（了解即可）：

1
2
3
4
5
6
7
8
9
10
11
12
import asyncio

# 现代 Python 推荐使用 asyncio.run()
# 它会自动创建和管理事件循环

# 底层 API（不推荐，但有助于理解）
async def main():
    loop = asyncio.get_running_loop()
    print(f"事件循环: {loop}")
    print(f"是否运行中: {loop.is_running()}")

asyncio.run(main())
理解事件循环

asyncio.run() 替你做了创建/运行/关闭事件循环，日常开发不需要直接操作它，但理解机制有助于排查问题。

3.2 Future 与 Task

Future - 未来的值：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import asyncio

async def demo_future():
    # Future 代表一个未来才会有结果的操作
    future = asyncio.get_running_loop().create_future()
  
    # 模拟1秒后设置结果
    async def set_result():
        await asyncio.sleep(1)
        future.set_result("任务完成了！")
  
    asyncio.create_task(set_result())
  
    # 等待结果
    result = await future
    print(result)

asyncio.run(demo_future())

Task - 协程的包装：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    return 42

async def main():
    # Task 是 Future 的子类
    # 它负责执行一个协程
  
    # 创建任务
    task = asyncio.create_task(my_coroutine())
  
    # task 本身是一个 Future
    print(f"是否完成: {task.done()}")  # False
  
    # 等待任务完成
    result = await task
    print(f"结果: {result}")  # 42
    print(f"是否完成: {task.done()}")  # True

asyncio.run(main())

Future vs Task：

Future

vs Task Future 是底层的”承诺”，需手动设置结果；Task 是 Future 子类，自动执行协程并设置结果，日常编程主要用 Task。

3.3 任务的精细控制

asyncio.wait() - 更灵活的并发控制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import asyncio

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

async def main():
    tasks = [
        task("A", 3),
        task("B", 1),
        task("C", 2),
    ]
  
    # 等待第一个完成
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
  
    print(f"已完成: {[t.result() for t in done]}")
    print(f"未完成: {len(pending)} 个任务")

asyncio.run(main())

return_when 选项：

选项 说明
FIRST_COMPLETED 任何一个完成就返回
FIRST_EXCEPTION 出现异常就返回
ALL_COMPLETED 全部完成才返回（默认）
Python

3.11+ 注意 asyncio.wait() 在 Python 3.11+ 中不再接受协程对象，需要传入 Task 对象。推荐使用 gather 或 TaskGroup。

asyncio.as_completed() - 按完成顺序获取结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import asyncio

async def task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} 完成"

async def main():
    tasks = [
        task("快任务", 1),
        task("中任务", 2),
        task("慢任务", 3),
    ]
  
    # 按完成顺序处理结果
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"收到结果: {result}")

asyncio.run(main())

输出：

1
2
3
收到结果: 快任务 完成
收到结果: 中任务 完成
收到结果: 慢任务 完成

asyncio.wait_for() - 设置超时：

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

async def slow_operation():
    await asyncio.sleep(10)  # 模拟很慢的操作
    return "完成"

async def main():
    try:
        # 设置3秒超时
        result = await asyncio.wait_for(slow_operation(), timeout=3)
    except asyncio.TimeoutError:
        print("操作超时！")

asyncio.run(main())

asyncio.shield() - 防止取消：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import asyncio

async def important_task():
    """这个任务不应该被取消"""
    print("重要任务开始")
    await asyncio.sleep(5)
    print("重要任务完成")
    return "重要结果"

async def main():
    # shield 保护任务不被外部取消
    task = asyncio.shield(important_task())
  
    # 即使我们取消它，实际任务也会继续执行
    task.cancel()
  
    try:
        await task
    except asyncio.CancelledError:
        print("任务被取消了")

asyncio.run(main())
shield

的典型场景 需要确保关键操作不被取消时使用（如资源清理、日志写入、事务提交）。shield 保护底层操作，外层 await 仍会抛 CancelledError。

3.4 Python 3.11+ 新特性

TaskGroup - 更安全的并发任务组：

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
import asyncio

async def fetch_user():
    await asyncio.sleep(1)
    return {"name": "张三"}

async def fetch_orders():
    await asyncio.sleep(2)
    return [{"id": 1, "item": "手机"}]

async def fetch_balance():
    await asyncio.sleep(1.5)
    return 10000

async def main():
    # TaskGroup 会自动管理任务的生命周期
    # 如果任何一个任务失败，会自动取消其他任务
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_user())
        task2 = tg.create_task(fetch_orders())
        task3 = tg.create_task(fetch_balance())
  
    # 所有任务完成后才能到这里
    print(f"用户: {task1.result()}")
    print(f"订单: {task2.result()}")
    print(f"余额: {task3.result()}")

asyncio.run(main())

TaskGroup vs gather：

TaskGroup

的优势 gather 中一个任务失败其他继续运行；TaskGroup 中一个任务失败自动取消其他任务。

asyncio.timeout() - 上下文管理器形式的超时：

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

async def main():
    # Python 3.11+ 推荐写法
    try:
        async with asyncio.timeout(3):
            await asyncio.sleep(10)  # 会超时
    except TimeoutError:
        print("超时了！")
  
    # 也可以用于一组操作
    async with asyncio.timeout(5):
        await fetch_user()
        await fetch_orders()  # 如果前面耗时太久，这里会超时

asyncio.run(main())
3.11+

超时推荐写法 使用 asyncio.timeout() 上下文管理器代替 wait_for()，代码更简洁，可包裹多行代码组成超时组。

阶段四：同步原语与多协程协作

4.1 为什么需要异步同步原语？

问题演示：

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

# ❌ 错误：使用 time.sleep 会阻塞整个事件循环！
async def bad_example():
    import time
    print("开始等待")
    time.sleep(5)  # 这会阻塞所有协程！
    print("等待结束")

# ✅ 正确：使用 asyncio.sleep
async def good_example():
    print("开始等待")
    await asyncio.sleep(5)  # 只阻塞当前协程
    print("等待结束")

为什么 threading.Lock 在异步中不能用？

1
2
3
4
5
6
7
8
import threading
import asyncio

lock = threading.Lock()

async def bad_with_threading_lock():
    with lock:  # ❌ 这会阻塞整个事件循环！
        await some_io_operation()
核心原则

在异步代码中，所有可能阻塞的操作都必须使用 asyncio 的异步版本，使用 threading 或 time 会阻塞整个事件循环。

4.2 异步锁：asyncio.Lock

使用场景： 多个协程需要修改同一个共享资源

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
import asyncio

# 共享资源
account_balance = 1000
lock = asyncio.Lock()

async def withdraw(amount):
    global account_balance
  
    # 获取锁（同一时刻只有一个协程能进入）
    async with lock:
        if account_balance >= amount:
            # 模拟一些处理时间
            await asyncio.sleep(0.1)
            account_balance -= amount
            print(f"取款 {amount}，余额: {account_balance}")
        else:
            print(f"余额不足，无法取款 {amount}")

async def main():
    # 并发执行多次取款
    tasks = [withdraw(300) for _ in range(5)]
    await asyncio.gather(*tasks)
    print(f"最终余额: {account_balance}")

asyncio.run(main())
没有锁的问题

不加锁可能导致并发竞态条件：两个协程同时检查余额都通过，结果超额取款。

4.3 异步事件：asyncio.Event

使用场景： 一个协程通知其他协程某事已发生

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
import asyncio

# 事件对象
server_ready = asyncio.Event()

async def server():
    """服务器启动"""
    print("服务器正在启动...")
    await asyncio.sleep(3)  # 模拟启动耗时
    print("服务器启动完成！")
  
    # 通知所有等待的协程
    server_ready.set()

async def client(name):
    """客户端连接"""
    print(f"客户端 {name} 等待服务器...")
  
    # 等待事件被设置
    await server_ready.wait()
  
    print(f"客户端 {name} 已连接")

async def main():
    # 并发运行服务器和客户端
    await asyncio.gather(
        server(),
        client("A"),
        client("B"),
        client("C"),
    )

asyncio.run(main())

输出：

1
2
3
4
5
6
7
8
服务器正在启动...
客户端 A 等待服务器...
客户端 B 等待服务器...
客户端 C 等待服务器...
服务器启动完成！
客户端 A 已连接
客户端 B 已连接
客户端 C 已连接

Event 的常用方法：

方法 说明
set() 设置事件（通知等待者）
clear() 清除事件（重置状态）
wait() 等待事件被设置
is_set() 检查事件是否已设置

4.4 异步条件：asyncio.Condition

使用场景： 等待某个条件成立

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
import asyncio

# 条件对象（自带锁）
condition = asyncio.Condition()
items = []  # 共享资源

async def producer():
    """生产者"""
    for i in range(5):
        async with condition:
            items.append(i)
            print(f"生产: {i}, 当前库存: {len(items)}")
          
            # 通知一个等待的消费者
            condition.notify()
      
        await asyncio.sleep(1)

async def consumer(name):
    """消费者"""
    while True:
        async with condition:
            # 等待条件：items 不为空
            await condition.wait_for(lambda: len(items) > 0)
          
            item = items.pop(0)
            print(f"消费者 {name} 消费: {item}")

async def main():
    await asyncio.gather(
        producer(),
        consumer("A"),
        consumer("B"),
    )

asyncio.run(main())

4.5 异步信号量：asyncio.Semaphore

限流利器：精确控制并发数量

核心应用： 限制并发数量（如限制同时请求数）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import asyncio

# 限制最多同时运行3个任务
semaphore = asyncio.Semaphore(3)

async def limited_task(task_id):
    """受限制的任务"""
    async with semaphore:  # 同时只有3个能进入
        print(f"任务 {task_id} 开始")
        await asyncio.sleep(2)  # 模拟耗时操作
        print(f"任务 {task_id} 完成")

async def main():
    # 创建10个任务，但同时只运行3个
    tasks = [limited_task(i) for i in range(10)]
    await asyncio.gather(*tasks)

asyncio.run(main())

爬虫限流示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import asyncio
import aiohttp

# 限制同时最多100个请求
semaphore = asyncio.Semaphore(100)

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

async def main():
    urls = [f"https://example.com/page/{i}" for i in range(1000)]
  
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        await asyncio.gather(*tasks)

asyncio.run(main())

4.6 异步队列：asyncio.Queue

生产者-消费者模式：

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
import asyncio
import random

async def producer(queue, name):
    """生产者：生产数据放入队列"""
    for i in range(5):
        item = f"{name}-产品-{i}"
        await asyncio.sleep(random.uniform(0.5, 1.5))  # 模拟生产耗时
        await queue.put(item)
        print(f"生产者 {name} 生产: {item}")

async def consumer(queue, name):
    """消费者：从队列取出数据处理"""
    while True:
        item = await queue.get()  # 队列为空时会等待
        print(f"消费者 {name} 处理: {item}")
        await asyncio.sleep(random.uniform(0.5, 1))  # 模拟处理耗时
        queue.task_done()  # 标记任务完成

async def main():
    # 创建队列（可设置最大容量）
    queue = asyncio.Queue(maxsize=10)
  
    # 创建生产者和消费者
    producers = [asyncio.create_task(producer(queue, f"P{i}")) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, f"C{i}")) for i in range(3)]
  
    # 等待所有生产者完成
    await asyncio.gather(*producers)
  
    # 等待队列被完全消费
    await queue.join()
  
    # 取消消费者（它们会一直等待）
    for c in consumers:
        c.cancel()

asyncio.run(main())

Queue 的常用方法：

方法 说明
put(item) 放入数据（队列满时会等待）
get() 取出数据（队列空时会等待）
task_done() 标记一个任务已完成
join() 等待所有任务完成
qsize() 当前队列大小
empty() 队列是否为空
full() 队列是否已满

阶段五：异步 I/O 与同步代码的桥接

5.1 在异步中运行同步阻塞代码

问题

有些库是同步的（如 requests、time.sleep），直接在异步中使用会阻塞事件循环。

解决方案：asyncio.to_thread()（Python 3.9+）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import asyncio
import time

def blocking_io():
    """同步阻塞的I/O操作"""
    print("开始阻塞操作...")
    time.sleep(5)  # 这是同步阻塞！
    print("阻塞操作完成")
    return "结果"

async def main():
    # 把同步函数放到线程池执行，不阻塞事件循环
    result = await asyncio.to_thread(blocking_io)
    print(f"结果: {result}")

asyncio.run(main())
CPU

密集型任务不能用 to_thread to_thread 把任务放到线程池，但 GIL 导致 CPU 密集型任务仍然串行，需要用 ProcessPoolExecutor。

to_thread

的工作原理 to_thread 把同步函数放到线程池执行，对 I/O 密集型有效，但对 CPU 密集型无效（GIL 限制）。

loop.run_in_executor() - 更底层的控制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def blocking_task():
    import time
    time.sleep(3)
    return "完成"

async def main():
    loop = asyncio.get_running_loop()
  
    # 使用默认线程池
    result = await loop.run_in_executor(None, blocking_task)
    print(result)
  
    # 使用自定义线程池
    with ThreadPoolExecutor(max_workers=5) as pool:
        result = await loop.run_in_executor(pool, blocking_task)
  
    # 使用进程池（适合CPU密集型）
    with ProcessPoolExecutor(max_workers=4) as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_task)

asyncio.run(main())

5.2 在同步代码中调用异步代码

规则：千万不要嵌套事件循环！

如果已有事件循环在运行，不能再调用 asyncio.run()，否则会 RuntimeError。

跨线程调用异步代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import asyncio
import threading

async def async_task():
    await asyncio.sleep(1)
    return "异步结果"

def run_async_in_thread():
    """在另一个线程中运行异步代码"""
    # 创建新的事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
  
    try:
        result = loop.run_until_complete(async_task())
        print(f"结果: {result}")
    finally:
        loop.close()

# 在单独线程中运行
thread = threading.Thread(target=run_async_in_thread)
thread.start()
thread.join()

run_coroutine_threadsafe() - 跨线程安全调用：

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
import asyncio
import threading

# 全局事件循环（在主线程运行）
main_loop = None

async def async_task(data):
    await asyncio.sleep(1)
    return f"处理: {data}"

def background_thread():
    """后台线程"""
    # 从主线程的事件循环提交协程
    future = asyncio.run_coroutine_threadsafe(
        async_task("来自线程的数据"),
        main_loop
    )
  
    # 等待结果（阻塞当前线程）
    result = future.result(timeout=5)
    print(f"线程收到结果: {result}")

async def main():
    global main_loop
    main_loop = asyncio.get_running_loop()
  
    # 启动后台线程
    thread = threading.Thread(target=background_thread)
    thread.start()
  
    # 主线程继续做其他事
    await asyncio.sleep(3)
  
    thread.join()

asyncio.run(main())

5.3 CPU 密集型任务的处理

为什么 to_thread 不适合 CPU 密集型？

1
2
3
4
5
GIL 的限制：
- 同一时刻只有一个线程执行 Python 字节码
- to_thread 只是把任务放到线程池
- 线程之间还是要竞争 GIL
- 结果：CPU 密集型任务用多线程没有性能提升

必须使用 ProcessPoolExecutor（多进程）：

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
import asyncio
from concurrent.futures import ProcessPoolExecutor
import time

def cpu_heavy(n):
    """CPU密集型任务"""
    total = 0
    for i in range(n):
        total += i * i
    return total

async def main():
    # 创建进程池
    with ProcessPoolExecutor(max_workers=4) as pool:
        loop = asyncio.get_running_loop()
      
        # 在进程池中执行CPU密集型任务
        tasks = [
            loop.run_in_executor(pool, cpu_heavy, 10**7)
            for _ in range(4)
        ]
      
        results = await asyncio.gather(*tasks)
        print(f"结果: {results}")

if __name__ == "__main__":
    # 注意：多进程代码需要在 if __name__ == "__main__" 中运行
    start = time.time()
    asyncio.run(main())
    print(f"耗时: {time.time() - start:.2f}s")

完整的混合方案：

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
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

async def handle_request():
    """处理一个请求"""
  
    # 1. I/O 操作：直接用异步
    data = await fetch_from_database()
  
    # 2. 同步 I/O 操作：用线程池
    file_content = await asyncio.to_thread(read_file, "data.txt")
  
    # 3. CPU 密集型操作：用进程池
    processed = await loop.run_in_executor(process_pool, cpu_process, data)
  
    # 4. 继续异步 I/O
    await send_response(processed)

async def main():
    # 配置执行器
    thread_pool = ThreadPoolExecutor(max_workers=10)
    process_pool = ProcessPoolExecutor(max_workers=4)
  
    loop = asyncio.get_running_loop()
    loop.set_default_executor(thread_pool)
  
    # 处理请求
    await handle_request()

阶段六：实际框架应用

6.1 FastAPI 中的异步模式

FastAPI 天生支持异步：

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

app = FastAPI()

# 异步路由处理函数
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 可以直接使用 await
    user = await fetch_user_from_db(user_id)
    return user

# 同步路由也可以（FastAPI 会自动放到线程池）
@app.get("/sync-endpoint")
def sync_endpoint():
    # 这是同步函数，FastAPI 会自动处理
    return {"message": "sync"}
FastAPI

的异步处理 FastAPI 会自动将同步路由放到线程池执行，但推荐使用 async def 获得更好性能。

并发请求处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from fastapi import FastAPI
import asyncio

app = FastAPI()

async def fetch_user_data(user_id: int):
    await asyncio.sleep(1)  # 模拟数据库查询
    return {"id": user_id, "name": f"用户{user_id}"}

async def fetch_user_orders(user_id: int):
    await asyncio.sleep(1.5)  # 模拟外部API调用
    return [{"order_id": 1, "item": "手机"}]

@app.get("/user-dashboard/{user_id}")
async def user_dashboard(user_id: int):
    # 并发获取用户信息和订单（总共只需1.5秒）
    user, orders = await asyncio.gather(
        fetch_user_data(user_id),
        fetch_user_orders(user_id)
    )
    return {"user": user, "orders": orders}

后台任务：

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
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

async def send_email(email: str, message: str):
    """异步发送邮件"""
    await asyncio.sleep(2)  # 模拟发送
    print(f"邮件已发送到 {email}")

async def log_action(action: str):
    """异步记录日志"""
    await asyncio.sleep(0.5)
    print(f"日志: {action}")

@app.post("/order")
async def create_order(
    order_data: dict,
    background_tasks: BackgroundTasks
):
    # 处理订单
    order = await save_order(order_data)

    # 添加后台任务（不会阻塞响应）
    background_tasks.add_task(send_email, "user@example.com", "订单创建成功")
    background_tasks.add_task(log_action, f"订单 {order['id']} 已创建")

    return order

依赖注入中的异步：

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, Depends, HTTPException

app = FastAPI()

async def get_db():
    """异步数据库连接依赖"""
    db = await create_async_connection()
    try:
        yield db
    finally:
        await db.close()

async def get_current_user(token: str = Depends(get_token)):
    """异步验证用户"""
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401)
    return user

@app.get("/protected")
async def protected_route(user = Depends(get_current_user), db = Depends(get_db)):
    data = await db.fetch_user_data(user.id)
    return data

中间件中的异步：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    """异步中间件：记录请求耗时"""
    start_time = time.time()

    # 继续处理请求
    response = await call_next(request)

    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)

    return response

6.2 aiohttp 异步 HTTP 客户端

基本使用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import aiohttp
import asyncio

async def fetch_url(url: str) -> str:
    """异步获取网页内容"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# 批量请求
async def fetch_multiple(urls: list[str]):
    tasks = [fetch_url(url) for url in urls]
    return await asyncio.gather(*tasks)

asyncio.run(fetch_multiple([
    "https://example.com",
    "https://python.org",
]))

并发控制（限制请求数）：

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
import aiohttp
import asyncio

async def fetch_with_limit(
    semaphore: asyncio.Semaphore,
    session: aiohttp.ClientSession,
    url: str
):
    """带并发限制的请求"""
    async with semaphore:  # 限制并发数
        async with session.get(url) as response:
            return await response.text()

async def crawl(urls: list[str], max_concurrent: int = 10):
    """爬虫：控制并发数量"""
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(semaphore, session, url)
            for url in urls
        ]
        return await asyncio.gather(*tasks)

asyncio.run(crawl([f"https://example.com/page/{i}" for i in range(100)]))

请求超时处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import aiohttp
import asyncio

async def fetch_with_timeout(url: str, timeout_seconds: float = 5):
    """带超时的请求"""
    timeout = aiohttp.ClientTimeout(total=timeout_seconds)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            async with session.get(url) as response:
                return await response.json()
        except asyncio.TimeoutError:
            print(f"请求超时: {url}")
            return None
        except aiohttp.ClientError as e:
            print(f"请求失败: {e}")
            return None

asyncio.run(fetch_with_timeout("https://api.example.com/data"))

POST 请求与 JSON：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import aiohttp
import asyncio

async def create_user(name: str, email: str):
    """异步POST请求"""
    async with aiohttp.ClientSession() as session:
        data = {"name": name, "email": email}

        async with session.post(
            "https://api.example.com/users",
            json=data,
            headers={"Authorization": "Bearer token123"}
        ) as response:
            if response.status == 201:
                return await response.json()
            else:
                raise Exception(f"创建失败: {response.status}")

asyncio.run(create_user("张三", "zhangsan@example.com"))

连接池配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import aiohttp
import asyncio

async def main():
    # 配置连接池
    connector = aiohttp.TCPConnector(
        limit=100,           # 总连接数限制
        limit_per_host=10,   # 每个主机的连接数限制
        ttl_dns_cache=300,   # DNS缓存时间
    )

    async with aiohttp.ClientSession(connector=connector) as session:
        # 所有请求共享连接池
        tasks = [session.get(f"https://api.example.com/{i}") for i in range(50)]
        responses = await asyncio.gather(*tasks)

asyncio.run(main())

6.3 异步数据库操作

SQLAlchemy 异步模式：

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
39
40
41
42
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, select

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

# 创建异步引擎（注意使用 asyncpg 驱动）
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    echo=True,
)

# 创建异步会话工厂
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def create_user(name: str, email: str):
    """创建用户"""
    async with async_session() as session:
        user = User(name=name, email=email)
        session.add(user)
        await session.commit()
        return user

async def get_user(user_id: int):
    """查询用户"""
    async with async_session() as session:
        result = await session.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

async def get_all_users():
    """查询所有用户"""
    async with async_session() as session:
        result = await session.execute(select(User))
        return result.scalars().all()

asyncpg（高性能 PostgreSQL）：

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
import asyncpg
import asyncio

async def main():
    # 创建连接池
    pool = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/db",
        min_size=5,
        max_size=20
    )

    # 使用连接池
    async with pool.acquire() as conn:
        # 执行查询
        users = await conn.fetch("SELECT * FROM users WHERE age > $1", 18)

        # 执行插入
        await conn.execute(
            "INSERT INTO users(name, email) VALUES($1, $2)",
            "张三", "zhangsan@example.com"
        )

        # 事务
        async with conn.transaction():
            await conn.execute("UPDATE accounts SET balance = balance - $1 WHERE id = $2", 100, 1)
            await conn.execute("UPDATE accounts SET balance = balance + $1 WHERE id = $2", 100, 2)

    await pool.close()

asyncio.run(main())

aioredis（异步 Redis）：

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
import redis.asyncio as redis
import asyncio

async def main():
    # 创建连接池
    r = redis.Redis(host='localhost', port=6379, decode_responses=True)

    # 基本操作
    await r.set("key", "value")
    value = await r.get("key")

    # 列表操作
    await r.lpush("mylist", "a", "b", "c")
    items = await r.lrange("mylist", 0, -1)

    # 发布订阅
    pubsub = r.pubsub()
    await pubsub.subscribe("channel1")

    # 管道（批量操作）
    async with r.pipeline() as pipe:
        pipe.set("key1", "val1")
        pipe.set("key2", "val2")
        await pipe.execute()

    await r.close()

asyncio.run(main())

6.4 完整的 FastAPI + 异步数据库示例

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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, sessionmaker
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, select
from contextlib import asynccontextmanager
from pydantic import BaseModel

# 数据库配置
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

# 模型
class UserDB(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

# Pydantic 模型
class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        from_attributes = True

# 生命周期管理
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时创建表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # 关闭时清理
    await engine.close()

app = FastAPI(lifespan=lifespan)

# 依赖注入
async def get_db():
    async with async_session() as session:
        yield session

# 路由
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    db_user = UserDB(name=user.name, email=user.email)
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

@app.get("/users/{user_id}", response_model=UserResponse)
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(UserDB).where(UserDB.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

@app.get("/users", response_model=list[UserResponse])
async def read_users(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(UserDB).offset(skip).limit(limit))
    return result.scalars().all()

# 并发请求示例
@app.get("/user-stats/{user_id}")
async def user_stats(user_id: int, db: AsyncSession = Depends(get_db)):
    import asyncio

    async def get_user():
        result = await db.execute(select(UserDB).where(UserDB.id == user_id))
        return result.scalar_one_or_none()

    async def get_orders():
        # 假设有订单表
        await asyncio.sleep(1)  # 模拟查询
        return [{"id": 1, "total": 100}]

    async def get_logs():
        # 假设有日志表
        await asyncio.sleep(0.5)  # 模拟查询
        return [{"action": "login"}]

    # 并发获取数据
    user, orders, logs = await asyncio.gather(
        get_user(),
        get_orders(),
        get_logs()
    )

    return {
        "user": user,
        "orders": orders,
        "logs": logs
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

总结

核心要点回顾

1
2
3
4
5
6
1. 异步的本质：I/O 等待时不阻塞，去做其他事
2. 协程：用 async def 定义，用 await 等待
3. 事件循环：调度协程执行的心脏
4. 并发执行：用 create_task 或 gather
5. 同步原语：Lock、Event、Semaphore、Queue 等
6. 桥接同步代码：to_thread（I/O）、ProcessPoolExecutor（CPU）

什么时候用异步？

适合异步

Web 服务器、爬虫、数据库操作、API 调用、文件 I/O

不适合异步

简单脚本、CPU 密集型计算（用多进程）、已有大量同步代码的项目（改造成本高）

学习建议

先掌握基础语法：async/await、create_task、gather
理解事件循环的工作原理
多写代码，实践出真知
遇到问题先查文档：https://docs.python.org/3/library/asyncio.html
Python asyncio 官方文档https://docs.python.org/3/library/asyncio.html

祝你学习愉快！异步编程是 Python 高并发的利器，掌握它会让你的代码能力上一个新台阶。

许可协议

本文采用 署名-非商业性使用-相同方式共享 4.0 国际 许可协议，转载请注明出处。

快来参与讨论吧~

本站由 楠瓜 使用 Stellar 1.33.1 主题创建。
风起于青萍之末，浪成于微澜之间。