返回首页
## 为什么需要异步编程
在现代应用程序开发中，我们经常需要处理大量 I/O 操作，比如网络请求、文件读写、数据库查询等。传统的同步编程模型会导致程序在等待 I/O 时阻塞，造成资源浪费。
异步编程允许我们在等待 I/O 操作完成的同时执行其他任务，显著提高程序的效率和响应速度。
## async/await 基础
Python 3.5 引入了 `async` 和 `await` 关键字，为异步编程提供了简洁的语法。
```python
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)
任务是协程的包装器，可以并发执行多个协程。
```python
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())
```
## 实战案例
### 并发网络请求
```python
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())
```
### 生产者消费者模式
```python
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
```python
# ❌ 错误
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()`：
```python
# ❌ 错误
async def bad_example():
time.sleep(1) # 阻塞事件循环
# ✅ 正确
async def good_example():
await asyncio.sleep(1)
```
### 3. 异常处理
```python
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 异步编程！*