Python 异步编程指南：async/await 完全解析

为什么需要异步编程

在现代应用程序开发中，我们经常需要处理大量 I/O 操作，比如网络请求、文件读写、数据库查询等。传统的同步编程模型会导致程序在等待 I/O 时阻塞，造成资源浪费。

异步编程允许我们在等待 I/O 操作完成的同时执行其他任务，显著提高程序的效率和响应速度。

async/await 基础

Python 3.5 引入了 async 和 await 关键字，为异步编程提供了简洁的语法。

import asyncio

async def fetch_data(url: str) -> dict:
    """异步获取数据"""
    print(f"开始获取: {url}")
    # 模拟网络请求
    await asyncio.sleep(1)
    return {"url": url, "data": "result"}

async def main():
    result = await fetch_data("https://api.example.com/data")
    print(f"获取完成: {result}")

asyncio.run(main())

核心概念

1. 协程 (Coroutine)

协程是异步函数的定义形式，使用 async def 声明。协程不会立即执行，需要被调度到事件循环中运行。

2. 事件循环 (Event Loop)

事件循环是异步编程的核心，它负责调度协程的执行。asyncio.run() 函数会创建一个新的事件循环并运行传入的协程。

3. 任务 (Task)

任务是协程的包装器，可以并发执行多个协程。

import asyncio

async def task1():
    await asyncio.sleep(1)
    return "Task 1 completed"

async def task2():
    await asyncio.sleep(1)
    return "Task 2 completed"

async def main():
    # 创建任务
    t1 = asyncio.create_task(task1())
    t2 = asyncio.create_task(task2())
    
    # 等待所有任务完成
    results = await asyncio.gather(t1, t2)
    print(results)  # ['Task 1 completed', 'Task 2 completed']

asyncio.run(main())

实战案例

并发网络请求

import asyncio
import aiohttp

async def fetch_session(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://api.github.com",
        "https://api.stackexchange.com",
        "https://httpbin.org",
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_session(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for url, result in zip(urls, results):
            print(f"{url}: {len(result)} bytes")

asyncio.run(main())

生产者消费者模式

import asyncio
import time

async def producer(queue: asyncio.Queue, items: list):
    for item in items:
        await queue.put(item)
        print(f"生产: {item}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # 发送结束信号

async def consumer(queue: asyncio.Queue, name: str):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"{name} 消费: {item}")
        await asyncio.sleep(0.2)
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    items = ["item1", "item2", "item3", "item4"]
    
    # 启动生产者
    producer_task = asyncio.create_task(producer(queue, items))
    
    # 启动多个消费者
    consumers = [
        asyncio.create_task(consumer(queue, "Consumer-A")),
        asyncio.create_task(consumer(queue, "Consumer-B")),
    ]
    
    await producer_task
    await asyncio.gather(*consumers)
    print("所有任务完成")

asyncio.run(main())

常见陷阱与解决方案

1. 忘记 await

# ❌ 错误
async def main():
    result = fetch_data("url")  # 返回协程对象，不是结果
    print(result)  # <coroutine object fetch_data at 0x...>

# ✅ 正确
async def main():
    result = await fetch_data("url")
    print(result)

2. 阻塞调用

不要在异步代码中使用阻塞调用，如 time.sleep()：

# ❌ 错误
async def bad_example():
    time.sleep(1)  # 阻塞事件循环

# ✅ 正确
async def good_example():
    await asyncio.sleep(1)

3. 异常处理

async def main():
    try:
        result = await fetch_data("url")
    except asyncio.TimeoutError:
        print("请求超时")
    except Exception as e:
        print(f"发生错误: {e}")

总结

异步编程是现代 Python 开发的重要技能。通过理解和掌握 async/await，你可以编写出更高效、更响应迅速的应用程序。

记住几个关键点：

使用 async def 定义协程
使用 await 等待异步操作
使用 asyncio.gather() 并发执行多个任务
避免在异步代码中使用阻塞调用

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