Python asyncio 异步编程实战：从入门到精通

为什么需要异步编程？

在传统的同步编程模型中，程序按照顺序执行，每个操作都会阻塞直到完成。当我们需要处理大量 I/O 密集型任务（如网络请求、文件读写、数据库查询）时，这种模型会导致程序效率低下，大量时间浪费在等待上。

异步编程（Asynchronous Programming）是一种并发编程范式，允许程序在等待 I/O 操作时切换到其他任务，从而充分利用 CPU 资源。Python 的 asyncio 库从 3.4 版本开始引入，到 3.7 版本成熟，已成为 Python 异步编程的标准解决方案。

asyncio 核心概念

1. 协程（Coroutine）

协程是 asyncio 的基本执行单元，使用 async def 定义。协程可以被挂起和恢复，不会阻塞事件循环。

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

2. 事件循环（Event Loop）

事件循环是 asyncio 的核心，负责调度和执行协程。它维护一个任务队列，不断检查哪些任务可以执行，并在任务之间切换。

async def main():
    task1 = asyncio.create_task(hello())
    task2 = asyncio.create_task(world())
    await asyncio.gather(task1, task2)

asyncio.run(main())

3. 任务（Task）

任务是对协程的封装，可以并发执行。使用 asyncio.create_task() 创建任务，使用 await 等待任务完成。

4. Future

Future 代表一个异步操作的结果，可以理解为"未来会完成的任务"。Task 是 Future 的子类。

基础用法示例

顺序执行 vs 并发执行

import asyncio
import time

async def fetch_data(name, delay):
    print(f"开始获取 {name} 的数据")
    await asyncio.sleep(delay)
    print(f"获取 {name} 的数据完成")
    return f"{name} 的数据"

# 顺序执行（耗时 3 秒）
async def sequential():
    start = time.time()
    await fetch_data("A", 1)
    await fetch_data("B", 1)
    await fetch_data("C", 1)
    print(f"顺序执行耗时: {time.time() - start:.2f} 秒")

# 并发执行（耗时 1 秒）
async def concurrent():
    start = time.time()
    await asyncio.gather(
        fetch_data("A", 1),
        fetch_data("B", 1),
        fetch_data("C", 1)
    )
    print(f"并发执行耗时: {time.time() - start:.2f} 秒")

asyncio.run(sequential())
asyncio.run(concurrent())

输出结果：

顺序执行耗时: 3.00 秒
并发执行耗时: 1.00 秒

高级用法

1. 超时控制

async def fetch_with_timeout():
    try:
        result = await asyncio.wait_for(
            fetch_data("重要数据", 5),
            timeout=2.0
        )
        print(result)
    except asyncio.TimeoutError:
        print("请求超时")

2. 信号量（Semaphore）限制并发数

async def limited_concurrent():
    semaphore = asyncio.Semaphore(3)  # 最多 3 个并发

    async def limited_task(name):
        async with semaphore:
            await fetch_data(name, 1)

    await asyncio.gather(
        *[limited_task(f"任务{i}") for i in range(10)]
    )

3. 异步上下文管理器

class AsyncDatabase:
    async def __aenter__(self):
        print("连接数据库")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("关闭数据库连接")
        await asyncio.sleep(0.1)

    async def query(self, sql):
        await asyncio.sleep(0.1)
        return f"查询结果: {sql}"

async def use_database():
    async with AsyncDatabase() as db:
        result = await db.query("SELECT * FROM users")
        print(result)

4. 异步生成器

async def async_generator():
    for i in range(5):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for item in async_generator():
        print(item)

实战案例：并发爬虫

import aiohttp
import asyncio

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

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

urls = [
    "https://www.example.com",
    "https://www.python.org",
    "https://www.github.com",
]

results = asyncio.run(crawl(urls))
print(f"成功获取 {len(results)} 个页面的内容")

常见问题与最佳实践

1. 什么时候使用异步？

• I/O 密集型任务：网络请求、文件读写、数据库操作等
• 高并发场景：需要同时处理大量连接
• 实时应用：WebSocket、聊天服务器等

2. 什么时候不要使用异步？

• CPU 密集型任务：使用 multiprocessing 或 threading
• 简单脚本：异步会增加代码复杂度
• 已有同步库：除非有异步版本，否则不要强行异步

3. 常见陷阱

# 错误：在协程中使用阻塞操作
async def bad_example():
    import time
    time.sleep(1)  # 阻塞事件循环！
    return "完成"

# 正确：使用异步版本
async def good_example():
    await asyncio.sleep(1)  # 不阻塞事件循环
    return "完成"

4. 性能优化建议

• 使用 asyncio.gather() 而不是逐个 await
• 使用连接池复用连接（如 aiohttp、aiomysql）
• 使用信号量限制并发数，避免资源耗尽
• 合理使用 asyncio.create_task() 创建后台任务

总结

Python asyncio 是一个强大的异步编程框架，掌握它可以显著提升 I/O 密集型应用的性能。关键概念包括协程、事件循环、任务和 Future。通过合理使用 async/await 语法，可以编写出简洁高效的并发代码。

记住：异步不是银弹，要根据实际场景选择合适的并发模型。对于 I/O 密集型任务，asyncio 是首选；对于 CPU 密集型任务，考虑使用 multiprocessing；对于简单的并发需求，threading 可能更合适。

希望这篇文章能帮助你更好地理解和应用 Python asyncio 异步编程！