异步编程与 asyncio

asyncio 提供协作式并发：单线程内通过事件循环调度多个任务，在 I/O 等待期间切换执行其他任务。适合大量网络连接、文件操作等 I/O 密集型场景。

事件循环原理

事件循环是 asyncio 的核心，它持续监控和调度异步任务：

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)  # 模拟异步操作
    print("World")

# asyncio.run() 创建事件循环并运行
asyncio.run(say_hello())

执行过程：

1. asyncio.run() 创建事件循环
2. 事件循环启动 say_hello() 协程
3. 遇到 await asyncio.sleep(1)，协程暂停，控制权返回事件循环
4. 1秒后，事件循环恢复 say_hello()
5. 协程完成，事件循环关闭

与线程的区别：

• 线程：抢占式调度，操作系统决定何时切换
• asyncio：协作式调度，代码通过 await 主动让出控制权

async/await 语法

定义协程

async def fetch_data():
    """async def 定义协程函数"""
    await asyncio.sleep(1)  # await 只能在 async 函数内使用
    return "数据"

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

# 需要事件循环执行
result = asyncio.run(coro)
print(result)  # 数据

await 表达式

await 暂停当前协程，等待另一个协程完成：

async def step1():
    print("步骤1 开始")
    await asyncio.sleep(1)
    print("步骤1 完成")
    return "结果1"

async def step2():
    print("步骤2 开始")
    await asyncio.sleep(1)
    print("步骤2 完成")
    return "结果2"

async def main():
    # 顺序执行（串行）
    result1 = await step1()
    result2 = await step2()
    print(f"结果: {result1}, {result2}")

asyncio.run(main())

输出：

步骤1 开始
（等待1秒）
步骤1 完成
步骤2 开始
（等待1秒）
步骤2 完成
结果: 结果1, 结果2

总耗时约2秒（串行）。

并发执行任务

gather：等待多个协程

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

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

asyncio.run(main())

输出：

任务A 开始
任务B 开始
任务C 开始
（等待1秒）
任务B 完成
（等待0.5秒）
任务C 完成
（等待0.5秒）
任务A 完成
全部结果: ['任务A 的结果', '任务B 的结果', '任务C 的结果']

总耗时约2秒（并发），三个任务几乎同时开始。

create_task：创建任务

async def background_task():
    await asyncio.sleep(2)
    print("后台任务完成")

async def main():
    # 创建任务立即开始执行
    task = asyncio.create_task(background_task())
    
    print("主协程继续执行")
    await asyncio.sleep(1)
    print("主协程工作中")
    
    # 等待后台任务
    await task

asyncio.run(main())

输出：

主协程继续执行
（等待1秒）
主协程工作中
（等待1秒）
后台任务完成

TaskGroup：任务组（Python 3.11+）

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

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(worker("A", 1))
        task2 = tg.create_task(worker("B", 2))
        task3 = tg.create_task(worker("C", 1.5))
    
    # 退出 async with 时自动等待所有任务
    print(f"结果: {task1.result()}, {task2.result()}, {task3.result()}")

asyncio.run(main())

TaskGroup 的优势：

• 任一任务失败时取消所有任务
• 自动清理资源
• 结构化并发

异步迭代器

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

async def main():
    async for num in AsyncRange(5):
        print(num)

asyncio.run(main())

异步上下文管理器

class AsyncResource:
    async def __aenter__(self):
        print("获取资源")
        await asyncio.sleep(0.5)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("释放资源")
        await asyncio.sleep(0.5)

async def main():
    async with AsyncResource() as resource:
        print("使用资源")

asyncio.run(main())

异步生成器

async def async_generator():
    for i in range(3):
        await asyncio.sleep(1)
        yield i

async def main():
    async for value in async_generator():
        print(f"收到: {value}")

asyncio.run(main())

实战案例：异步 HTTP 请求

需要安装 aiohttp：

import asyncio
import aiohttp

async def fetch_url(session, url):
    """异步获取 URL"""
    async with session.get(url) as response:
        return await response.text()

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

asyncio.run(main())

与线程集成

阻塞式 I/O 需要在线程池中运行：

import asyncio
import time

def blocking_io():
    """阻塞式函数"""
    time.sleep(2)
    return "完成"

async def main():
    loop = asyncio.get_running_loop()
    
    # 在线程池中运行阻塞函数
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

asyncio.run(main())

常见模式

超时控制

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

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2)
    except asyncio.TimeoutError:
        print("操作超时")

asyncio.run(main())

等待多个任务

async def task1():
    await asyncio.sleep(1)
    return "任务1"

async def task2():
    await asyncio.sleep(2)
    return "任务2"

async def main():
    # 等待第一个完成
    done, pending = await asyncio.wait(
        [task1(), task2()],
        return_when=asyncio.FIRST_COMPLETED
    )
    
    print(f"完成: {[t.result() for t in done]}")
    
    # 取消未完成的任务
    for task in pending:
        task.cancel()

asyncio.run(main())

小结

• asyncio 用单线程实现高并发，适合 I/O 密集型任务
• async/await 是协作式并发的核心语法
• gather/create_task 并发执行多个协程
• TaskGroup 提供结构化并发（Python 3.11+）
• 异步迭代器/上下文管理器 扩展异步能力
• run_in_executor 集成阻塞式代码

下一节学习 concurrent.futures，它提供统一的高层并发接口。

---

练习：答案