Python 异步编程实战：async/await 避坑指南

背景

异步编程能显著提升 I/O 密集型应用的性能，但很多开发者在使用 async/await 时容易踩坑。

常见陷阱

1. 在同步函数中调用异步函数

# ❌ 错误写法
def sync_func():
    result = await async_func()  # SyntaxError!

# ✅ 正确写法
async def async_func_wrapper():
    result = await async_func()

2. 阻塞操作污染事件循环

# ❌ 错误 - time.sleep 会阻塞整个事件循环
async def bad_example():
    await asyncio.sleep(1)  # 正确
    time.sleep(1)  # 错误！阻塞!

# ✅ 正确 - 使用 run_in_executor
async def good_example():
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, blocking_func)

3. 忘记 await

# ❌ 错误
async def fetch_data():
    task = asyncio.create_task(get_data())
    return task  # 返回的是 Task 对象，不是结果!

# ✅ 正确
async def fetch_data():
    task = asyncio.create_task(get_data())
    result = await task
    return result

最佳实践

使用 asyncio.gather 并发执行

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

添加超时控制

async def fetch_with_timeout(url, timeout=5):
    try:
        return await asyncio.wait_for(fetch(url), timeout=timeout)
    except asyncio.TimeoutError:
        print(f"请求超时：{url}")
        return None

使用 Semaphore 限制并发数

semaphore = asyncio.Semaphore(10)

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

调试技巧

1. 使用 asyncio.run() 作为入口点
2. 启用 debug 模式检测阻塞调用
3. 使用 aiomonitor 监控事件循环

asyncio.run(main(), debug=True)

总结

异步编程需要思维转变。记住：

• 所有 I/O 操作都应该是异步的
• 避免在 async 函数中执行阻塞操作
• 合理使用并发控制工具

#Python #异步编程 #async #await #编程技巧