我来为您详细介绍 Python 中的主要异步库及其比较： ## 主要异步库 ### 1. **asyncio** (官方标准库) ```python import asyncio async d
我来为您详细介绍 Python 中的主要异步库及其比较：
主要异步库
1. asyncio (官方标准库)
1
2
3
4
5
6
7
8
|
import asyncio
async def main():
print('Hello')
await asyncio.sleep(1)
print('World')
asyncio.run(main())
|
特点：
- Python 3.4+ 内置，最基础的异步框架
- 提供事件循环、协程、任务等核心功能
- 生态丰富，很多第三方库基于 asyncio
2. aiohttp (HTTP 客户端/服务器)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
import aiohttp
import asyncio
async def fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
# Web 服务器
from aiohttp import web
async def handle(request):
return web.Response(text="Hello")
app = web.Application()
app.router.add_get('/', handle)
|
3. Trio (现代化替代品)
1
2
3
4
5
6
7
8
9
10
11
12
13
|
import trio
async def child():
print("Child: started!")
await trio.sleep(1)
print("Child: finished!")
async def parent():
async with trio.open_nursery() as nursery:
nursery.start_soon(child)
nursery.start_soon(child)
trio.run(parent)
|
4. Curio (轻量级替代品)
1
2
3
4
5
6
7
8
9
10
11
12
13
|
import curio
async def countdown(n):
while n > 0:
print('T-minus', n)
await curio.sleep(1)
n -= 1
async def main():
task = await curio.spawn(countdown, 10)
await task.join()
curio.run(main())
|
5. AnyIO (统一抽象层)
1
2
3
4
5
6
7
8
9
10
11
12
13
|
import anyio
async def task(name, seconds):
print(f"{name} starting")
await anyio.sleep(seconds)
print(f"{name} finished")
async def main():
async with anyio.create_task_group() as tg:
tg.start_soon(task, "Task 1", 1)
tg.start_soon(task, "Task 2", 2)
anyio.run(main)
|
详细比较
| 特性 |
asyncio |
Trio |
Curio |
aiohttp |
| 学习曲线 |
中等 |
平缓 |
简单 |
简单 |
| 错误处理 |
一般 |
优秀 |
良好 |
良好 |
| 并发模型 |
Task/Future |
Nursery |
Task |
基于asyncio |
| 取消机制 |
复杂 |
优雅 |
简单 |
一般 |
| 超时处理 |
需要手动 |
内置支持 |
需要手动 |
内置支持 |
| 社区生态 |
丰富 |
增长中 |
较小 |
HTTP领域丰富 |
性能对比
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
# asyncio 性能示例
import asyncio
import time
async def worker(name, delay):
start = time.time()
await asyncio.sleep(delay)
end = time.time()
return f"{name}: {end-start:.2f}s"
async def benchmark():
tasks = [worker(f"Task-{i}", 1) for i in range(5)]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
# asyncio.run(benchmark())
|
选择建议
推荐场景：
- 新手入门: 从
asyncio 开始
- 生产环境:
asyncio + aiohttp 组合
- 高可靠性要求:
Trio
- 教育目的:
Curio (代码清晰)
- 跨框架兼容:
AnyIO
实际应用示例：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
# 使用 asyncio + aiohttp 的完整示例
import asyncio
import aiohttp
from datetime import datetime
class AsyncFetcher:
def __init__(self):
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
await self.session.close()
async def fetch_urls(self, urls):
tasks = []
for url in urls:
task = asyncio.create_task(self._fetch_single(url))
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
async def _fetch_single(self, url):
async with self.session.get(url) as response:
data = await response.text()
return {
'url': url,
'status': response.status,
'size': len(data),
'timestamp': datetime.now()
}
async def main():
urls = [
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/2',
'https://httpbin.org/delay/1'
]
async with AsyncFetcher() as fetcher:
results = await fetcher.fetch_urls(urls)
for result in results:
print(result)
# asyncio.run(main())
|
总结
- asyncio: 生态系统最完善，适合大多数项目
- Trio: 设计更现代化，错误处理更好
- Curio: 概念简单，适合学习和理解原理
- aiohttp: HTTP 相关任务的绝佳选择
建议根据项目需求和个人偏好选择合适的库。对于新项目，建议优先考虑 asyncio 生态系统。