Python 异步编程完全指南（三）：四大经典实战场景
·
Python 异步编程完全指南（三）：四大经典实战场景
前言
理论学习只是开始，真正的掌握来自于实践。本篇将通过 4 个完整的实战项目，帮你把异步编程知识转化为实战能力。
本篇实战项目：
- 批量下载图片（带并发控制）
- 异步 API 客户端（带重试机制）
- 实时数据处理管道（生产者 - 消费者模式）
- WebSocket 实时通信
案例一：批量下载图片
1.1 需求分析
- 从网络批量下载大量图片
- 控制并发数，避免过载
- 统计下载成功/失败数量
- 支持异步文件写入
1.2 完整代码
import asyncio
import aiohttp
import aiofiles
from pathlib import Path
import time
from dataclasses import dataclass
from typing import List
@dataclass
class DownloadResult:
"""下载结果"""
url: str
status: str # success, failed, error
size: int = 0
error: str = ""
async def download_image(
session: aiohttp.ClientSession,
url: str,
save_path: Path
) -> DownloadResult:
"""下载单张图片"""
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as response:
if response.status == 200:
content = await response.read()
# 异步写入文件
async with aiofiles.open(save_path, 'wb') as f:
await f.write(content)
return DownloadResult(
url=url,
status="success",
size=len(content)
)
else:
return DownloadResult(
url=url,
status="failed",
error=f"HTTP {response.status}"
)
except asyncio.TimeoutError:
return DownloadResult(url=url, status="error", error="Timeout")
except Exception as e:
return DownloadResult(url=url, status="error", error=str(e))
async def batch_download(
urls: List[str],
save_dir: str,
max_concurrent: int = 10
) -> dict:
"""批量下载图片，限制并发数"""
# 创建保存目录
save_path = Path(save_dir)
save_path.mkdir(parents=True, exist_ok=True)
# 使用信号量控制并发
semaphore = asyncio.Semaphore(max_concurrent)
async def limited_download(session, url, path):
async with semaphore:
print(f"⬇️ 正在下载：{url[:50]}...")
result = await download_image(session, url, path)
if result.status == "success":
print(f"✅ 完成：{path.name} ({result.size} bytes)")
else:
print(f"❌ 失败：{path.name} - {result.error}")
return result
# 创建 HTTP 会话
connector = aiohttp.TCPConnector(limit=max_concurrent)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = []
for i, url in enumerate(urls):
# 从 URL 提取文件名或使用序号
filename = f"image_{i:04d}.jpg"
file_path = save_path / filename
task = asyncio.create_task(
limited_download(session, url, file_path)
)
tasks.append(task)
# 等待所有任务完成
results = await asyncio.gather(*tasks, return_exceptions=True)
# 统计结果
success = sum(1 for r in results if isinstance(r, DownloadResult) and r.status == "success")
failed = sum(1 for r in results if isinstance(r, DownloadResult) and r.status == "failed")
error = sum(1 for r in results if isinstance(r, DownloadResult) and r.status == "error")
total_size = sum(r.size for r in results if isinstance(r, DownloadResult))
return {
"total": len(urls),
"success": success,
"failed": failed,
"error": error,
"total_size": total_size
}
async def main():
# 示例：使用 Lorem Picsum 生成随机图片 URL
urls = [
f"https://picsum.photos/800/600?random={i}"
for i in range(20)
]
print(f"🚀 开始下载 {len(urls)} 张图片...")
print("=" * 50)
start = time.time()
result = await batch_download(
urls,
save_dir="./downloads",
max_concurrent=5 # 最多 5 个并发
)
elapsed = time.time() - start
print("=" * 50)
print(f"📊 下载统计:")
print(f" 总数：{result['total']}")
print(f" 成功：{result['success']}")
print(f" 失败：{result['failed']}")
print(f" 错误：{result['error']}")
print(f" 总大小：{result['total_size'] / 1024:.2f} KB")
print(f" 总耗时：{elapsed:.2f} 秒")
print(f" 平均速度：{result['total'] / elapsed:.1f} 张/秒")
if __name__ == "__main__":
asyncio.run(main())
1.3 关键技术点
┌─────────────────────────────────────────────────────────────┐
│ 批量下载架构图 │
├─────────────────────────────────────────────────────────────┤
│ │
│ URL 列表 │
│ [url1, url2, url3, ..., url100] │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────┐ │
│ │ 信号量 (Semaphore=10) │ │
│ │ 控制最多 10 个任务同时执行 │ │
│ └─────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────┬─────────┬─────────┬─────────┐ │
│ │ Task 1 │ Task 2 │ ... │ Task 10 │ │
│ │ 下载 │ 下载 │ │ 下载 │ │
│ │ url1 │ url2 │ │ url10 │ │
│ └────┬────┴────┬────┴────┬────┴────┬────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────┐ │
│ │ aiofiles 异步写入 │ │
│ └─────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
案例二：异步 API 客户端
2.1 需求分析
- 封装 RESTful API 调用
- 支持自动重试（指数退避）
- 支持超时控制
- 支持批量请求
2.2 完整代码
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import Optional, Any, Dict, List
import time
@dataclass
class APIResponse:
"""API 响应数据类"""
status_code: int
data: Any
error: Optional[str] = None
elapsed: float = 0.0
@property
def is_success(self) -> bool:
return 200 <= self.status_code < 300
class AsyncAPIClient:
"""异步 API 客户端"""
def __init__(
self,
base_url: str,
timeout: int = 30,
max_retries: int = 3,
headers: Dict[str, str] = None
):
self.base_url = base_url.rstrip('/')
self.timeout = aiohttp.ClientTimeout(total=timeout)
self.max_retries = max_retries
self.default_headers = headers or {}
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
"""异步上下文管理器入口"""
self._session = aiohttp.ClientSession(
timeout=self.timeout,
headers=self.default_headers
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器出口"""
if self._session:
await self._session.close()
async def _request(
self,
method: str,
endpoint: str,
**kwargs
) -> APIResponse:
"""发送请求（带重试机制）"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
for attempt in range(self.max_retries):
try:
start_time = time.time()
async with self._session.request(method, url, **kwargs) as response:
elapsed = time.time() - start_time
# 尝试解析 JSON
try:
data = await response.json()
except:
data = await response.text()
return APIResponse(
status_code=response.status,
data=data,
elapsed=elapsed
)
except asyncio.TimeoutError:
if attempt == self.max_retries - 1:
return APIResponse(
status_code=0,
data=None,
error="Timeout"
)
# 指数退避
wait_time = 2 ** attempt
print(f"⏳ 超时，{wait_time}秒后重试 ({attempt + 1}/{self.max_retries})")
await asyncio.sleep(wait_time)
except aiohttp.ClientError as e:
if attempt == self.max_retries - 1:
return APIResponse(
status_code=0,
data=None,
error=str(e)
)
wait_time = 2 ** attempt
print(f"⚠️ 连接错误，{wait_time}秒后重试 ({attempt + 1}/{self.max_retries})")
await asyncio.sleep(wait_time)
return APIResponse(status_code=0, data=None, error="Max retries exceeded")
async def get(
self,
endpoint: str,
params: Dict = None
) -> APIResponse:
"""GET 请求"""
return await self._request("GET", endpoint, params=params)
async def post(
self,
endpoint: str,
data: Dict = None,
json: Dict = None
) -> APIResponse:
"""POST 请求"""
return await self._request("POST", endpoint, data=data, json=json)
async def put(
self,
endpoint: str,
json: Dict = None
) -> APIResponse:
"""PUT 请求"""
return await self._request("PUT", endpoint, json=json)
async def delete(self, endpoint: str) -> APIResponse:
"""DELETE 请求"""
return await self._request("DELETE", endpoint)
async def batch_get(
self,
endpoints: List[str],
max_concurrent: int = 10
) -> List[APIResponse]:
"""批量 GET 请求"""
semaphore = asyncio.Semaphore(max_concurrent)
async def limited_get(endpoint):
async with semaphore:
return await self.get(endpoint)
tasks = [limited_get(ep) for ep in endpoints]
return await asyncio.gather(*tasks)
# 使用示例
async def main():
# 使用 JSONPlaceholder 作为测试 API
base_url = "https://jsonplaceholder.typicode.com"
async with AsyncAPIClient(base_url, timeout=10, max_retries=3) as client:
print("=" * 50)
print("📡 异步 API 客户端示例")
print("=" * 50)
# 1. 单个 GET 请求
print("\n1️⃣ 单个 GET 请求:")
response = await client.get("/posts/1")
print(f" 状态码：{response.status_code}")
print(f" 耗时：{response.elapsed:.3f}s")
print(f" 标题：{response.data.get('title', 'N/A')[:40]}...")
# 2. POST 请求
print("\n2️⃣ POST 请求:")
response = await client.post("/posts", json={
"title": "Test Post",
"body": "This is a test post",
"userId": 1
})
print(f" 状态码：{response.status_code}")
print(f" 创建 ID: {response.data.get('id', 'N/A')}")
# 3. 批量请求
print("\n3️⃣ 批量 GET 请求 (10 个):")
start = time.time()
endpoints = [f"/posts/{i}" for i in range(1, 11)]
results = await client.batch_get(endpoints, max_concurrent=5)
elapsed = time.time() - start
success = sum(1 for r in results if r.is_success)
print(f" 总耗时：{elapsed:.2f}s")
print(f" 成功：{success}/{len(results)}")
print(f" 平均耗时：{elapsed/len(results):.3f}s/请求")
# 显示部分结果
print("\n 前 3 条结果:")
for i, r in enumerate(results[:3], 1):
title = r.data.get('title', 'N/A')[:30] if r.is_success else r.error
print(f" {i}. {title}...")
if __name__ == "__main__":
asyncio.run(main())
2.3 关键技术点
- 指数退避重试：失败后等待时间翻倍（1s → 2s → 4s）
- 上下文管理器：自动管理会话生命周期
- 信号量控制：防止并发请求过多
案例三：实时数据处理管道
3.1 需求分析
- 生产者不断产生数据
- 多个消费者并行处理
- 使用队列解耦生产和消费
- 优雅的启动和停止
3.2 完整代码
import asyncio
from asyncio import Queue
import random
import time
from dataclasses import dataclass
from typing import List, Any
@dataclass
class DataItem:
"""数据项"""
id: int
content: str
created_at: float
class AsyncDataPipeline:
"""异步数据处理管道"""
def __init__(self, num_workers: int = 3, queue_size: int = 100):
self.num_workers = num_workers
self.input_queue: Queue = Queue(maxsize=queue_size)
self.output_queue: Queue = Queue()
self._running = False
self._stats = {
"produced": 0,
"processed": 0,
"failed": 0
}
async def producer(self, data_source: List[Any], rate_limit: float = 0.1):
"""生产者：将数据放入队列"""
print(f"📤 生产者启动，共 {len(data_source)} 条数据")
for i, item in enumerate(data_source):
if not self._running:
break
data_item = DataItem(
id=i,
content=str(item),
created_at=time.time()
)
await self.input_queue.put(data_item)
self._stats["produced"] += 1
# 模拟数据流入速率
await asyncio.sleep(rate_limit)
# 发送结束信号
for _ in range(self.num_workers):
await self.input_queue.put(None)
print(f"📤 生产者完成，共生产 {self._stats['produced']} 条")
async def worker(self, worker_id: int):
"""消费者：处理队列中的数据"""
processed = 0
print(f"⚙️ Worker-{worker_id} 启动")
while True:
# 从队列获取数据
item = await self.input_queue.get()
# 检查结束信号
if item is None:
break
try:
# 模拟数据处理（耗时操作）
await asyncio.sleep(random.uniform(0.1, 0.3))
# 处理数据
result = f"[Worker-{worker_id}] 处理完成：{item.content}"
# 将结果放入输出队列
await self.output_queue.put({
"worker": worker_id,
"item_id": item.id,
"result": result,
"latency": time.time() - item.created_at
})
processed += 1
self._stats["processed"] += 1
except Exception as e:
self._stats["failed"] += 1
print(f"❌ Worker-{worker_id} 处理失败：{e}")
finally:
self.input_queue.task_done()
print(f"⚙️ Worker-{worker_id} 完成，共处理 {processed} 条")
async def collector(self):
"""收集器：收集处理结果"""
results = []
while self._running or not self.output_queue.empty():
try:
result = await asyncio.wait_for(
self.output_queue.get(),
timeout=1.0
)
results.append(result)
print(f"✅ {result['result'][:50]}... (延迟：{result['latency']:.3f}s)")
except asyncio.TimeoutError:
continue
return results
async def run(self, data_source: List[Any]) -> dict:
"""运行管道"""
self._running = True
start_time = time.time()
print("=" * 60)
print("🚀 数据处理管道启动")
print(f" 数据量：{len(data_source)}")
print(f" Worker 数：{self.num_workers}")
print("=" * 60)
# 创建所有任务
producer_task = asyncio.create_task(
self.producer(data_source)
)
worker_tasks = [
asyncio.create_task(self.worker(i))
for i in range(self.num_workers)
]
collector_task = asyncio.create_task(
self.collector()
)
# 等待生产者和所有 Worker 完成
await producer_task
await asyncio.gather(*worker_tasks)
# 标记运行结束，让收集器退出
self._running = False
results = await collector_task
elapsed = time.time() - start_time
print("=" * 60)
print("📊 管道统计:")
print(f" 生产：{self._stats['produced']}")
print(f" 处理：{self._stats['processed']}")
print(f" 失败：{self._stats['failed']}")
print(f" 总耗时：{elapsed:.2f}s")
print(f" 吞吐量：{self._stats['processed'] / elapsed:.1f} 条/秒")
print("=" * 60)
return {
"results": results,
"stats": self._stats,
"elapsed": elapsed
}
async def main():
# 创建数据源
data_source = [f"data_{i}" for i in range(30)]
# 创建并运行管道
pipeline = AsyncDataPipeline(num_workers=3)
result = await pipeline.run(data_source)
if __name__ == "__main__":
asyncio.run(main())
3.3 架构图
┌─────────────────────────────────────────────────────────────────┐
│ 数据处理管道架构 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 数据源 │
│ [data_0, data_1, data_2, ..., data_n] │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ 生产者 │ rate_limit=0.1s │
│ │ Producer │ │
│ └──────┬──────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────┐ │
│ │ 输入队列 (Input Queue) │ │
│ │ maxsize=100 │ │
│ └────┬────────────┬────────────┬──────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │Worker-0│ │Worker-1│ │Worker-2│ │
│ │ 处理 │ │ 处理 │ │ 处理 │ │
│ └───┬────┘ └───┬────┘ └───┬────┘ │
│ │ │ │ │
│ └────────────┼────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────┐ │
│ │ 输出队列 (Output Queue) │ │
│ └────────────────┬────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ 收集器 │ │
│ │ Collector │ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
案例四：WebSocket 实时通信
4.1 需求分析
- 建立 WebSocket 连接
- 实现消息收发
- 支持心跳保活
- 优雅断开连接
4.2 完整代码
import asyncio
import json
from datetime import datetime
from typing import Callable, Optional
from dataclasses import dataclass
# 注意：需要安装 websockets 库
# pip install websockets
try:
import websockets
from websockets.exceptions import ConnectionClosed
except ImportError:
print("请先安装 websockets: pip install websockets")
exit(1)
@dataclass
class Message:
"""消息结构"""
type: str
content: str
timestamp: str
class WebSocketClient:
"""WebSocket 客户端"""
def __init__(self, uri: str):
self.uri = uri
self.websocket = None
self._running = False
self._reconnect_interval = 5
self._heartbeat_interval = 30
async def connect(self) -> bool:
"""建立连接"""
try:
self.websocket = await websockets.connect(
self.uri,
ping_interval=20,
ping_timeout=10
)
self._running = True
print(f"✅ 已连接到 {self.uri}")
return True
except Exception as e:
print(f"❌ 连接失败：{e}")
return False
async def disconnect(self):
"""断开连接"""
self._running = False
if self.websocket:
await self.websocket.close()
print("🔌 连接已关闭")
async def send(self, message: dict):
"""发送消息"""
if self.websocket and self._running:
msg = Message(
type=message.get("type", "message"),
content=message.get("content", ""),
timestamp=datetime.now().isoformat()
)
await self.websocket.send(json.dumps(msg.__dict__))
print(f"📤 发送：{msg.content[:50]}")
async def receive(self) -> Optional[dict]:
"""接收消息"""
if self.websocket and self._running:
try:
data = await self.websocket.recv()
return json.loads(data)
except json.JSONDecodeError:
return {"type": "raw", "content": data}
return None
async def listen(self, callback: Callable):
"""持续监听消息"""
print("👂 开始监听消息...")
while self._running:
try:
message = await self.receive()
if message:
await callback(message)
except ConnectionClosed:
print("⚠️ 连接断开")
if self._running:
print(f"🔄 {self._reconnect_interval}秒后重连...")
await asyncio.sleep(self._reconnect_interval)
await self.connect()
except Exception as e:
print(f"❌ 监听错误：{e}")
await asyncio.sleep(1)
async def heartbeat(self):
"""心跳保活"""
while self._running:
try:
await self.send({"type": "ping", "content": "heartbeat"})
await asyncio.sleep(self._heartbeat_interval)
except Exception as e:
print(f"💔 心跳失败：{e}")
break
async def message_handler(message: dict):
"""消息处理回调"""
timestamp = datetime.now().strftime("%H:%M:%S")
msg_type = message.get("type", "unknown")
content = message.get("content", str(message))
if msg_type == "ping":
print(f"[{timestamp}] 💓 心跳响应")
else:
print(f"[{timestamp}] 📩 收到消息：{content[:100]}")
async def demo_echo_server():
"""演示：使用公共 Echo 服务器"""
# 使用公共的 WebSocket Echo 服务器
# 它会原样返回你发送的消息
client = WebSocketClient("wss://echo.websocket.org")
print("=" * 50)
print("🌐 WebSocket 客户端演示")
print(" 服务器：wss://echo.websocket.org")
print(" (Echo 服务器会返回你发送的消息)")
print("=" * 50)
if not await client.connect():
return
# 并发执行：监听 + 定时发送
async def send_messages():
messages = [
"Hello WebSocket!",
"这是一条测试消息",
"Async Python is awesome!",
"第四条消息",
"最后一条消息"
]
for i, msg in enumerate(messages):
await asyncio.sleep(2) # 每 2 秒发送一条
await client.send({"type": "message", "content": msg})
# 发送完毕后等待一会儿再断开
await asyncio.sleep(2)
await client.disconnect()
# 同时运行监听和发送
await asyncio.gather(
client.listen(message_handler),
send_messages()
)
async def main():
await demo_echo_server()
if __name__ == "__main__":
asyncio.run(main())
4.3 关键技术点
- 双向通信：同时收发消息
- 心跳机制：定期发送 ping 保持连接
- 自动重连：连接断开后自动尝试重连
- 回调模式：通过回调函数处理收到的消息
五、本篇小结
通过 4 个实战案例，我们学习了：
| 案例 | 核心技术 | 适用场景 |
|---|---|---|
| 批量下载 | Semaphore + aiofiles | 爬虫、资源下载 |
| API 客户端 | 重试机制 + 上下文管理 | API 调用、微服务 |
| 数据管道 | Queue + 生产者消费者 | 数据处理、ETL |
| WebSocket | 双向通信 + 心跳保活 | 实时应用、聊天 |
代码仓库
本篇所有代码都可以直接运行（安装相应依赖后）：
# 安装依赖
pip install aiohttp aiofiles websockets
下篇预告
在下一篇 高级技巧篇 中，我们将学习：
- 信号量精确控制并发
- 超时控制的多种方式
- 异步上下文管理器
- 异步迭代器与生成器
- 性能优化技巧
如果这篇文章对你有帮助，欢迎点赞、收藏、关注！有问题欢迎评论区讨论。
- - - - - - - - - - - - - - - - - - - 5. 4. 3. 2. >
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
8
0- 0
已为社区贡献18条内容
所有评论(0)