wwt｜逻辑与碎片

首页 / 知识库 / 0基础入门-阅读资料 / 0基础-python入门到精通

第 20 章 异步编程

20.1 协程的演进

Python 异步编程经历了三个阶段：

# 阶段 1：yield 生成器（Python 2.5+）
def old_coroutine():
    result = yield "请求数据"
    print(f"收到: {result}")

# 阶段 2：yield from 委托（Python 3.3+）
def delegating():
    result = yield from sub_generator()
    return result

# 阶段 3：async/await 原生协程（Python 3.5+）— 现代方式
async def modern_coroutine():
    result = await some_async_operation()
    return result

20.2 事件循环与 asyncio

事件循环是异步编程的核心——它不断检查并执行就绪的协程：

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)   # 非阻塞等待
    print("World")

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

# 事件循环同时管理多个协程
async def task(name, delay):
    print(f"[{name}] 开始")
    await asyncio.sleep(delay)
    print(f"[{name}] 完成（{delay}s）")
    return name

async def main():
    # 并发执行三个任务
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3),
    )
    print(f"结果: {results}")

asyncio.run(main())
# [A] 开始
# [B] 开始
# [C] 开始
# [B] 完成（1s）
# [A] 完成（2s）
# [C] 完成（3s）
# 结果: ['A', 'B', 'C']
# 总耗时约 3 秒（而非 6 秒）

20.3 async def 与 await

import asyncio

# async def 定义协程函数
async def fetch_data(url: str) -> str:
    print(f"开始请求 {url}")
    await asyncio.sleep(1)   # 模拟网络请求
    return f"来自 {url} 的数据"

# 调用协程函数返回协程对象，不会立即执行
coro = fetch_data("https://example.com")
print(type(coro))  # <class 'coroutine'>

# 必须用 await 或事件循环来执行
async def main():
    result = await fetch_data("https://example.com")
    print(result)

asyncio.run(main())

# await 只能在 async 函数内部使用
# await 后面必须是可等待对象（awaitable）：协程、Task、Future

协程 vs 普通函数

import asyncio
import time

# 错误：在异步代码中使用阻塞调用
async def bad_example():
    time.sleep(3)   # 阻塞！会冻结整个事件循环
    return "done"

# 正确：使用异步版本
async def good_example():
    await asyncio.sleep(3)   # 非阻塞，事件循环可以处理其他任务
    return "done"

# 如果必须调用阻塞函数，放到线程池中
async def run_blocking():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, time.sleep, 3)
    # 或使用 asyncio.to_thread（Python 3.9+）
    result = await asyncio.to_thread(time.sleep, 3)

20.4 asyncio.gather()、asyncio.create_task()

create_task — 创建并调度任务

async def main():
    # create_task 立即调度协程运行
    task1 = asyncio.create_task(fetch_data("url1"))
    task2 = asyncio.create_task(fetch_data("url2"))
    
    # 此时 task1 和 task2 已经在后台运行
    print("任务已创建，做其他事情...")
    
    # await 获取结果
    result1 = await task1
    result2 = await task2
    print(result1, result2)

gather — 并发运行多个协程

async def main():
    # 并发执行，收集所有结果
    results = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3"),
    )
    print(results)  # 按传入顺序返回结果

    # return_exceptions=True 时异常作为结果返回而不是抛出
    results = await asyncio.gather(
        fetch_data("url1"),
        failing_task(),
        fetch_data("url3"),
        return_exceptions=True,
    )
    # results[1] 可能是一个 Exception 对象

wait — 更灵活的等待

async def main():
    tasks = [asyncio.create_task(fetch_data(f"url{i}")) for i in range(5)]
    
    # 等待第一个完成
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    for task in done:
        print(f"完成: {task.result()}")
    
    # 等待所有完成（带超时）
    done, pending = await asyncio.wait(tasks, timeout=5.0)
    for task in pending:
        task.cancel()  # 取消超时的任务

20.5 异步迭代器

class AsyncRange:
    def __init__(self, start, stop):
        self.start = start
        self.stop = stop
    
    def __aiter__(self):
        self.current = self.start
        return self
    
    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)  # 模拟异步操作
        value = self.current
        self.current += 1
        return value

async def main():
    async for i in AsyncRange(0, 5):
        print(i)

异步推导式：

async def main():
    # 异步列表推导
    results = [i async for i in AsyncRange(0, 5)]
    print(results)  # [0, 1, 2, 3, 4]
    
    # 带条件的异步推导
    evens = [i async for i in AsyncRange(0, 10) if i % 2 == 0]

20.6 异步上下文管理器

class AsyncDBConnection:
    async def __aenter__(self):
        print("打开数据库连接")
        await asyncio.sleep(0.1)  # 模拟连接
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("关闭数据库连接")
        await asyncio.sleep(0.1)
        return False

async def main():
    async with AsyncDBConnection() as db:
        print("使用数据库")

用 contextlib 简化：

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_resource(name):
    print(f"获取资源 {name}")
    await asyncio.sleep(0.1)
    try:
        yield name
    finally:
        print(f"释放资源 {name}")
        await asyncio.sleep(0.1)

async def main():
    async with managed_resource("DB") as resource:
        print(f"使用 {resource}")

20.7 异步生成器

async def async_counter(start, stop):
    for i in range(start, stop):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for value in async_counter(0, 5):
        print(value)

# 异步生成器表达式
async def main():
    gen = (i * 2 async for i in async_counter(0, 5))
    async for value in gen:
        print(value)

实用场景：异步流式读取：

async def read_chunks(file_path, chunk_size=1024):
    """异步逐块读取文件"""
    import aiofiles
    async with aiofiles.open(file_path, "rb") as f:
        while chunk := await f.read(chunk_size):
            yield chunk

async def process_file(path):
    total = 0
    async for chunk in read_chunks(path):
        total += len(chunk)
    print(f"文件大小: {total} 字节")

20.8 asyncio 同步原语

异步版本的锁、信号量等，用于协程间同步：

import asyncio

# Lock — 异步锁
lock = asyncio.Lock()

async def safe_update(shared_data, key, value):
    async with lock:
        await asyncio.sleep(0.1)  # 模拟异步操作
        shared_data[key] = value

# Semaphore — 限制并发数
semaphore = asyncio.Semaphore(3)  # 最多 3 个并发

async def rate_limited_request(url):
    async with semaphore:
        print(f"请求 {url}")
        await asyncio.sleep(1)
        return f"{url} 的响应"

async def main():
    urls = [f"https://api.example.com/{i}" for i in range(10)]
    results = await asyncio.gather(*[rate_limited_request(u) for u in urls])
    # 每次最多 3 个并发请求

# Event — 异步事件
event = asyncio.Event()

async def waiter(name):
    print(f"{name} 等待中...")
    await event.wait()
    print(f"{name} 收到信号！")

async def trigger():
    await asyncio.sleep(2)
    print("触发事件")
    event.set()

async def main():
    await asyncio.gather(
        waiter("A"), waiter("B"), waiter("C"),
        trigger(),
    )

# Queue — 异步队列
queue = asyncio.Queue(maxsize=10)

async def producer():
    for i in range(5):
        await queue.put(f"item-{i}")
        await asyncio.sleep(0.1)

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

20.9 TaskGroup 与异常组

Python 3.11+ 引入 TaskGroup，提供结构化并发：

import asyncio

async def fetch(url):
    await asyncio.sleep(1)
    if "bad" in url:
        raise ValueError(f"请求失败: {url}")
    return f"{url} 的数据"

# TaskGroup — 结构化并发（Python 3.11+）
async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch("https://api.com/1"))
        task2 = tg.create_task(fetch("https://api.com/2"))
        task3 = tg.create_task(fetch("https://api.com/3"))
    
    # 离开 with 块时，所有任务已完成
    print(task1.result())
    print(task2.result())
    print(task3.result())

asyncio.run(main())

TaskGroup 的异常处理

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch("https://api.com/ok"))
            tg.create_task(fetch("https://bad.com/fail"))
            tg.create_task(fetch("https://api.com/ok2"))
    except* ValueError as eg:
        # 任何任务失败，其他任务会被取消
        for e in eg.exceptions:
            print(f"错误: {e}")
    except* ConnectionError as eg:
        for e in eg.exceptions:
            print(f"连接错误: {e}")

TaskGroup vs gather

gatherTaskGroup
异常处理一个失败全部取消（默认）使用异常组 except*
取消传播手动处理自动取消所有任务
结构化否是（保证退出时所有任务完成）
版本要求Python 3.4+Python 3.11+

超时控制

async def main():
    # asyncio.timeout（Python 3.11+）
    try:
        async with asyncio.timeout(5.0):
            result = await slow_operation()
    except TimeoutError:
        print("操作超时")
    
    # asyncio.wait_for（旧方式）
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
    except asyncio.TimeoutError:
        print("操作超时")

本章小结：异步编程是处理 I/O 密集型任务的最佳方案。核心概念：async/await 定义和等待协程，create_task 调度并发执行，gather 或 TaskGroup 收集结果。关键原则：永远不要在异步代码中调用阻塞函数（用 to_thread 包装），用异步同步原语（Lock/Semaphore）控制并发。