Karene's Blog
文章
合集Python 并发编程第 4 / 7 篇

asyncio 异步编程

约 22 分钟

高并发 I/O 的高效方案。协程的开销极低，单线程可以管理数万个并发连接。但 asyncio 需要整个生态的配合，用错库反而更慢。

1. 协程的本质

1.1 和线程的对比

对比维度线程协程
调度方式操作系统抢占式调度事件循环协作式调度（主动让出）
切换开销有系统调用开销（上下文切换）极低（用户态切换）
内存占用每个线程约 1MB 栈空间每个协程极少（KB 级别）
并发数量数千数万甚至数十万
适合场景I/O 密集 + 简单逻辑高并发 I/O（网络服务、爬虫）

1.2 协程的工作方式

普通函数：调用 → 执行 → 返回（一步到底）
协程函数：调用 → 执行到 await → 暂停，让出控制权 → 其他协程跑 → 回来继续

关键区别：协程在 await 时主动让出执行权，不浪费 CPU 在"等待"上。操作系统不需要介入调度——全部由事件循环在用户态完成。

2. 三件套：async def / await / asyncio.run()

2.1 定义和调用协程

import asyncio

async def fetch_data(name, delay):
    """async def 定义协程函数。"""
    print(f"开始获取 {name}")
    await asyncio.sleep(delay)   # await 是"等"的标志，让出控制权
    print(f"完成获取 {name}")
    return f"data: {name}"

2.2 串行 vs 并发

async def main():
    # 串行执行（总耗时 = 各任务之和 = 3 秒）
    r1 = await fetch_data("A", 1)
    r2 = await fetch_data("B", 1)
    r3 = await fetch_data("C", 1)

    # 并发执行（总耗时 = 最长任务 = 1 秒）
    results = await asyncio.gather(
        fetch_data("A", 1),
        fetch_data("B", 1),
        fetch_data("C", 1),
    )
    print(results)   # ['data: A', 'data: B', 'data: C']

# 启动入口
asyncio.run(main())   # 创建事件循环，运行完后关闭

2.3 关键规则

  • async def 定义协程函数，调用它不会立刻执行，返回协程对象
  • await 只能在 async def 内使用，表示"暂停当前协程，等结果回来再继续"
  • asyncio.run() 启动事件循环，整个程序只调用一次

3. 事件循环的工作原理

事件循环运行流程：
1. 获取就绪的任务（I/O 完成、定时器到期等）
2. 依次执行就绪任务，直到每个任务主动 await（让出控制权）
3. 轮询等待更多 I/O 事件
4. 重复 1-3，直到所有任务完成

简化理解：事件循环就是一个无限循环，不断检查"谁的 I/O 好了就执行谁"。没有线程切换的开销，一切都在单线程里协作完成。

# 推荐方式（Python 3.7+）
asyncio.run(main())   # 创建新事件循环，运行完后关闭

# 底层访问（框架开发、Jupyter 等特殊场景）
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

4. Task 和 Future

4.1 create_task

create_task 将协程立即提交给事件循环调度：

async def main():
    # create_task 立即调度，不阻塞
    task1 = asyncio.create_task(fetch_data("A", 1))
    task2 = asyncio.create_task(fetch_data("B", 0.5))

    # 等待各自完成
    result1 = await task1
    result2 = await task2

4.2 gather vs wait

# gather：等待全部完成，异常会传播（除非 return_exceptions=True）
results = await asyncio.gather(*tasks, return_exceptions=True)

# wait：可以按完成顺序处理，支持 FIRST_COMPLETED 等策略
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for task in done:
    print(task.result())
方法适用场景
gather等所有任务完成，收集结果
wait(FIRST_COMPLETED)先完成的先处理，类似多线程的 as_completed
wait(FIRST_EXCEPTION)遇到第一个异常就返回
wait(ALL_COMPLETED)等所有完成（默认行为）

4.3 超时控制

# Python 3.11+ 推荐方式
try:
    result = await asyncio.wait_for(fetch_data("C", 5), timeout=2.0)
except asyncio.TimeoutError:
    print("超时！")

4.4 取消任务

async def main():
    task = asyncio.create_task(long_running())
    await asyncio.sleep(1)
    task.cancel()   # 发送取消信号

    try:
        await task
    except asyncio.CancelledError:
        print("任务已取消")

5. 异步生态：哪些库支持异步

5.1 异步 vs 同步库对照

功能同步库异步库
HTTP 请求requestsaiohttp、httpx（AsyncClient）
文件 I/Oopen()aiofiles
数据库psycopg2、pymysqlasyncpg、aiomysql、aiosqlite
Redisredis-pyaioredis

5.2 正确的异步 HTTP 请求

import httpx

# 错误：requests 是同步库，会阻塞事件循环
async def bad_fetch(url):
    import requests
    return requests.get(url).text   # 这一行会阻塞所有协程！

# 正确：httpx 支持异步
async def good_fetch(url):
    async with httpx.AsyncClient() as client:
        r = await client.get(url)
        return r.text

# 也正确：aiohttp
import aiohttp

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

6. 阻塞代码怎么办：run_in_executor

当你必须调用同步库（如没有异步替代品）时，用 run_in_executor 桥接：

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

def blocking_io():
    """无法改为异步的阻塞操作。"""
    time.sleep(1)
    return "done"

async def main():
    loop = asyncio.get_running_loop()

    # 在线程池中运行阻塞函数，不阻塞事件循环
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

    # 使用自定义线程池控制并发数
    with ThreadPoolExecutor(max_workers=4) as pool:
        result = await loop.run_in_executor(pool, blocking_io)

7. 并发控制

7.1 Semaphore 限流

import asyncio

semaphore = asyncio.Semaphore(5)   # 最多 5 个并发

async def fetch_with_limit(url, client):
    async with semaphore:   # 超过 5 个协程会等待
        r = await client.get(url)
        return r.text

async def main():
    async with httpx.AsyncClient() as client:
        tasks = [fetch_with_limit(url, client) for url in urls]
        results = await asyncio.gather(*tasks)

7.2 Queue 生产者-消费者

import asyncio

async def producer(queue):
    for i in range(10):
        await queue.put(i)
        print(f"生产: {i}")
    await queue.put(None)   # 哨兵值

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)   # 传递给其他消费者
            break
        await asyncio.sleep(0.1)   # 模拟处理
        print(f"消费者 {name} 处理: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=3)   # 最多缓存 3 个
    await asyncio.gather(
        producer(queue),
        consumer(queue, "A"),
        consumer(queue, "B"),
    )

8. 什么时候用 asyncio vs 多线程

场景推荐原因
几个并发请求（< 10）ThreadPoolExecutor更简单，不需要异步生态
几十~几千并发asyncio性能优势明显
Web 框架（FastAPI/Sanic）异步是默认框架已经帮你管理事件循环
CPU 密集不要用 asyncio协程是单线程，无法利用多核
库不支持异步ThreadPoolExecutor + run_in_executor不要强行异步

9. 小结

知识点要记住的
async def / await协程的定义和等待，await 只能在 async def 内用
asyncio.run()启动入口，整个程序只调一次
gather / wait收集结果，wait 支持灵活的完成策略
异步生态必须用支持异步的库（aiohttp / httpx / aiofiles）
run_in_executor桥接同步和异步的桥梁
Semaphore / Queue并发控制和生产者-消费者

上一篇：python-multiprocessing，CPU 密集任务的解药。 下一篇：python-subprocess，当 Python 不够用时，调用外部工具。