当前位置：首页 > news >正文

Python 异步编程最佳实践

news 2026/3/26 21:19:12

Python 异步编程最佳实践

什么是异步编程？

异步编程是一种编程范式，允许程序同时处理多个任务，而不需要按顺序等待每个任务完成。

同步 vs 异步

同步代码:

importtimedeffetch_data():time.sleep(1)# 模拟耗时操作return"data"result1=fetch_data()# 等待 1 秒result2=fetch_data()# 再等待 1 秒result3=fetch_data()# 再等待 1 秒# 总共耗时 3 秒

异步代码:

importasyncioasyncdeffetch_data():awaitasyncio.sleep(1)# 模拟耗时操作return"data"asyncdefmain():results=awaitasyncio.gather(fetch_data(),fetch_data(),fetch_data())# 总共只耗时 1 秒！asyncio.run(main())

asyncio 核心概念

1. async/await 语法

asyncdefsay_hello():print("Hello")awaitasyncio.sleep(1)print("World")```### 2. 事件循环 (Event Loop)事件循环是异步编程的核心机制： ```python loop=asyncio.get_event_loop()loop.run_until_complete(main())loop.close()

3. Task 对象

Task 用于并发执行多个协程：

task1=asyncio.create_task(fetch_url('url1'))task2=asyncio.create_task(fetch_url('url2'))results=awaitasyncio.gather(task1,task2)

实战案例

案例 1: 批量网页抓取

importaiohttpasyncdeffetch_page(session,url):asyncwithsession.get(url)asresponse:returnawaitresponse.text()asyncdefmain():urls=['url1','url2','url3']asyncwithaiohttp.ClientSession()assession:tasks=[fetch_page(session,url)forurlinurls]results=awaitasyncio.gather(*tasks)```### 案例 2: 定时任务```pythonasyncdefperiodic_task():whileTrue:print("执行任务...")awaitasyncio.sleep(60)# 每分钟执行一次asyncio.run(periodic_task())

常见陷阱

1. 阻塞事件循环

❌错误示范:

asyncdefbad_example():time.sleep(1)# 这会阻塞整个事件循环！``` ✅**正确做法**:```pythonasyncdefgood_example():awaitasyncio.sleep(1)# 使用异步版本```### 2. 忘记 await```pythonasyncdefmain():task=some_async_function()# ❌ 忘记 await# 应该写成：await some_async_function()```### 3. 异常处理```pythontry:awaitrisky_operation()exceptExceptionase:print(f"出错了：{e}")```## 性能对比|方式|10个请求耗时|100个请求耗时||------|------------|-------------||同步|~10秒|~100秒||异步|~1秒|~1秒|**效率提升**:**10倍+**🚀## 最佳实践### 1. 使用异步 HTTP 库-`aiohttp`-异步 HTTP 客户端/服务器--`httpx`-支持同步和异步的 HTTP 客户端### 2. 合理控制并发量```python semaphore=asyncio.Semaphore(10)# 限制最多 10 个并发asyncdeflimited_task(url):asyncwithsemaphore:awaitfetch_url(url)```### 3. 优雅地取消任务```python task=asyncio.create_task(long_running_task())# 稍后取消task.cancel()try:awaittaskexceptasyncio.CancelledError:print("任务已取消")```## 总结掌握 Python 异步编程，让你的程序效率起飞！ ✅**适用场景**:-IO 密集型任务（网络请求、文件读写）--需要高并发的应用--实时数据处理 ❌**不适用场景**:-CPU 密集型计算（使用多进程更好）--简单的脚本程序---**学习资源**:-[Python asyncio 官方文档](https://docs.python.org/3/library/asyncio.html)--[Real Python AsyncIO 教程](https://realpython.com/async-io-python/)--[AIOHTTP 文档](https://docs.aiohttp.org/)---*希望本文对你有帮助！欢迎点赞收藏。*