Python 异步编程实战：async/await 最佳实践指南

引言

在 Python 3.5+ 中，async/await 语法彻底改变了我们编写并发代码的方式。但很多开发者在使用时仍会踩坑。本文将分享异步编程的核心实践。

1. 理解 async/await 的本质

async 定义协程，await 挂起执行等待异步操作完成。关键点：await 只能用于可等待对象。

import asyncio
import aiohttp

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

2. 避免常见陷阱

陷阱 1：在同步函数中调用异步代码

❌ 错误做法：

def sync_function():
    result = asyncio.run(async_func())  # 可能引发事件循环冲突

✅ 正确做法：将整个调用链改为异步

陷阱 2：阻塞事件循环

❌ 错误：

async def process():
    time.sleep(1)  # 阻塞整个事件循环！

✅ 正确：

async def process():
    await asyncio.sleep(1)  # 非阻塞等待

3. 并发执行模式

使用 asyncio.gather()

async def fetch_all(urls):
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    return results

使用 asyncio.create_task()

async def process_with_tasks():
    task1 = asyncio.create_task(fetch_data(url1))
    task2 = asyncio.create_task(fetch_data(url2))
    # 可以做其他事情
    result1 = await task1
    result2 = await task2

4. 超时与取消

from asyncio import timeout, CancelledError

async def fetch_with_timeout(url, timeout_sec=5):
    try:
        async with timeout(timeout_sec):
            return await fetch_data(url)
    except TimeoutError:
        print(f"请求超时：{url}")
        return None
    except CancelledError:
        print("任务被取消")
        raise

5. 异步上下文管理器

自定义异步上下文管理器处理资源：

class AsyncDatabase:
    async def __aenter__(self):
        self.conn = await create_connection()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

# 使用
async with AsyncDatabase() as db:
    data = await db.query("SELECT * FROM users")

6. 性能优化建议

1. 批量操作：减少 await 次数，使用 gather 批量处理
2. 连接池：复用 HTTP 连接、数据库连接
3. 限制并发数：使用 Semaphore 控制并发量
4. 监控慢查询：记录耗时超过阈值的操作

from asyncio import Semaphore

semaphore = Semaphore(10)  # 最多 10 个并发

async def limited_fetch(url):
    async with semaphore:
        return await fetch_data(url)

总结

异步编程的核心是非阻塞和并发。掌握这些实践能让你写出更高效、更可靠的 Python 代码。记住：

• 保持调用链异步
• 避免阻塞操作
• 合理使用并发原语
• 做好错误处理和超时控制

实践出真知，多写多练才能熟练掌握！