引言
在现代Python开发中，异步编程已经成为构建高性能应用的重要技术手段。随着并发需求的增长和网络I/O密集型应用的普及，掌握异步编程的核心概念和最佳实践变得尤为重要。本文将深入探讨Python异步编程的各个方面，从基础的asyncio库使用到构建高性能网络服务的完整实践。
什么是异步编程
异步编程的基本概念
异步编程是一种编程范式，它允许程序在等待某些操作完成时执行其他任务，而不是阻塞等待。这种模式特别适用于I/O密集型操作，如网络请求、文件读写、数据库查询等场景。
在传统的同步编程中，当程序执行一个耗时的I/O操作时，整个线程会被阻塞，直到操作完成。而在异步编程中，程序可以发起一个异步操作，然后继续执行其他任务，当异步操作完成后，再通过回调或事件机制处理结果。
异步编程的优势
- 高并发性：异步编程能够在单个线程中处理大量并发连接
- 资源效率：减少了线程创建和切换的开销
- 响应性：应用程序能够保持响应，不会因为长时间的I/O操作而阻塞
- 可扩展性：能够轻松处理更多的并发用户请求
asyncio库详解
asyncio基础概念
asyncio是Python标准库中用于编写异步I/O程序的核心模块。它提供了事件循环、协程、任务和未来对象等核心组件。
import asyncio
import time
# 基本的异步函数定义
async def say_hello(name):
print(f"Hello, {name}!")
await asyncio.sleep(1) # 模拟异步操作
print(f"Goodbye, {name}!")
# 运行异步函数
async def main():
# 并发执行多个异步函数
await asyncio.gather(
say_hello("Alice"),
say_hello("Bob"),
say_hello("Charlie")
)
# 执行入口
if __name__ == "__main__":
asyncio.run(main())
事件循环
事件循环是asyncio的核心，它负责调度和执行异步任务。Python中的asyncio默认使用事件循环来管理协程的执行。
import asyncio
async def task(name, delay):
print(f"Task {name} started")
await asyncio.sleep(delay)
print(f"Task {name} completed after {delay} seconds")
return f"Result from {name}"
async def main():
# 创建事件循环
loop = asyncio.get_event_loop()
# 方法1：使用gather并发执行
results = await asyncio.gather(
task("A", 2),
task("B", 1),
task("C", 3)
)
print("All tasks completed:", results)
# 运行示例
asyncio.run(main())
协程和任务
协程是异步编程的基本单元，而任务则是对协程的包装，提供了更多的控制能力。
import asyncio
import time
async def fetch_data(url, delay):
print(f"Fetching data from {url}")
await asyncio.sleep(delay)
return f"Data from {url}"
async def main():
# 创建任务列表
tasks = [
asyncio.create_task(fetch_data("api1", 2)),
asyncio.create_task(fetch_data("api2", 1)),
asyncio.create_task(fetch_data("api3", 3))
]
# 等待所有任务完成
results = await asyncio.gather(*tasks)
print("Results:", results)
# 运行示例
asyncio.run(main())
异步数据库操作
使用异步数据库驱动
现代Python生态系统提供了多种异步数据库驱动，如asyncpg（PostgreSQL）、aiomysql（MySQL）、motor（MongoDB）等。
import asyncio
import asyncpg
from typing import List, Dict
class AsyncDatabaseManager:
def __init__(self, connection_string: str):
self.connection_string = connection_string
self.pool = None
async def connect(self):
"""建立数据库连接池"""
self.pool = await asyncpg.create_pool(
self.connection_string,
min_size=5,
max_size=20
)
async def close(self):
"""关闭数据库连接池"""
if self.pool:
await self.pool.close()
async def fetch_users(self, limit: int = 10) -> List[Dict]:
"""异步获取用户数据"""
async with self.pool.acquire() as connection:
query = """
SELECT id, name, email, created_at
FROM users
ORDER BY created_at DESC
LIMIT $1
"""
rows = await connection.fetch(query, limit)
return [dict(row) for row in rows]
async def insert_user(self, name: str, email: str) -> int:
"""异步插入用户数据"""
async with self.pool.acquire() as connection:
query = """
INSERT INTO users (name, email, created_at)
VALUES ($1, $2, NOW())
RETURNING id
"""
user_id = await connection.fetchval(query, name, email)
return user_id
async def batch_insert_users(self, users: List[Dict]) -> int:
"""批量插入用户数据"""
async with self.pool.acquire() as connection:
# 使用批量插入提高性能
query = """
INSERT INTO users (name, email, created_at)
VALUES ($1, $2, NOW())
"""
# 准备数据
data = [(user['name'], user['email']) for user in users]
# 批量执行
result = await connection.executemany(query, data)
return len(data)
# 使用示例
async def demo_database_operations():
db_manager = AsyncDatabaseManager("postgresql://user:password@localhost/db")
try:
await db_manager.connect()
# 插入用户
user_id = await db_manager.insert_user("John Doe", "john@example.com")
print(f"Inserted user with ID: {user_id}")
# 批量插入
users = [
{"name": "Alice Smith", "email": "alice@example.com"},
{"name": "Bob Johnson", "email": "bob@example.com"},
{"name": "Carol Brown", "email": "carol@example.com"}
]
inserted_count = await db_manager.batch_insert_users(users)
print(f"Batch inserted {inserted_count} users")
# 查询用户
users = await db_manager.fetch_users(5)
print("Fetched users:", users)
except Exception as e:
print(f"Database error: {e}")
finally:
await db_manager.close()
# 运行示例
# asyncio.run(demo_database_operations())
异步数据库连接池管理
合理的连接池配置对于高性能异步应用至关重要。
import asyncio
import asyncpg
from contextlib import asynccontextmanager
class AsyncConnectionPool:
def __init__(self, connection_string: str, min_size: int = 5, max_size: int = 20):
self.connection_string = connection_string
self.min_size = min_size
self.max_size = max_size
self.pool = None
async def initialize(self):
"""初始化连接池"""
self.pool = await asyncpg.create_pool(
self.connection_string,
min_size=self.min_size,
max_size=self.max_size,
max_inactive_connection_lifetime=300, # 5分钟
command_timeout=60, # 1分钟超时
connect_timeout=10 # 10秒连接超时
)
async def close(self):
"""关闭连接池"""
if self.pool:
await self.pool.close()
@asynccontextmanager
async def get_connection(self):
"""获取数据库连接的上下文管理器"""
if not self.pool:
await self.initialize()
connection = None
try:
connection = await self.pool.acquire()
yield connection
finally:
if connection:
await self.pool.release(connection)
async def execute_with_retry(self, query: str, *args, max_retries: int = 3):
"""带重试机制的查询执行"""
for attempt in range(max_retries):
try:
async with self.get_connection() as conn:
return await conn.execute(query, *args)
except asyncpg.PostgresError as e:
if attempt < max_retries - 1:
print(f"Attempt {attempt + 1} failed: {e}. Retrying...")
await asyncio.sleep(2 ** attempt) # 指数退避
else:
raise
# 使用示例
async def demo_connection_pool():
pool = AsyncConnectionPool(
"postgresql://user:password@localhost/db",
min_size=5,
max_size=10
)
try:
await pool.initialize()
# 使用连接池执行查询
async with pool.get_connection() as conn:
result = await conn.fetch("SELECT version()")
print("Database version:", result[0][0])
except Exception as e:
print(f"Error: {e}")
finally:
await pool.close()
# asyncio.run(demo_connection_pool())
异步HTTP客户端
使用aiohttp构建异步HTTP客户端
aiohttp是Python中流行的异步HTTP客户端和服务器库。
import aiohttp
import asyncio
import time
from typing import Dict, List, Optional
class AsyncHttpClient:
def __init__(self, timeout: int = 30, max_connections: int = 100):
self.timeout = aiohttp.ClientTimeout(total=timeout)
self.connector = aiohttp.TCPConnector(
limit=max_connections,
limit_per_host=30,
ttl_dns_cache=300,
use_dns_cache=True
)
self.session = None
async def __aenter__(self):
"""异步上下文管理器入口"""
self.session = aiohttp.ClientSession(
timeout=self.timeout,
connector=self.connector
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器出口"""
if self.session:
await self.session.close()
async def get(self, url: str, headers: Optional[Dict] = None) -> Dict:
"""异步GET请求"""
try:
async with self.session.get(url, headers=headers) as response:
return {
'status': response.status,
'headers': dict(response.headers),
'data': await response.text()
}
except Exception as e:
return {
'error': str(e),
'status': 0
}
async def post(self, url: str, data: Dict, headers: Optional[Dict] = None) -> Dict:
"""异步POST请求"""
try:
async with self.session.post(url, json=data, headers=headers) as response:
return {
'status': response.status,
'headers': dict(response.headers),
'data': await response.text()
}
except Exception as e:
return {
'error': str(e),
'status': 0
}
async def fetch_multiple(self, urls: List[str]) -> List[Dict]:
"""并发获取多个URL"""
tasks = [self.get(url) for url in urls]
return await asyncio.gather(*tasks, return_exceptions=True)
# 使用示例
async def demo_http_client():
urls = [
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/2',
'https://httpbin.org/json',
'https://httpbin.org/user-agent'
]
async with AsyncHttpClient(timeout=10) as client:
start_time = time.time()
# 并发请求
results = await client.fetch_multiple(urls)
end_time = time.time()
print(f"Completed {len(results)} requests in {end_time - start_time:.2f} seconds")
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Request {i+1} failed: {result}")
else:
print(f"Request {i+1} status: {result['status']}")
# asyncio.run(demo_http_client())
高性能HTTP客户端优化
import aiohttp
import asyncio
from typing import List, Dict, Optional
import json
class OptimizedAsyncHttpClient:
def __init__(self):
# 使用连接池配置
self.connector = aiohttp.TCPConnector(
limit=100, # 总连接数限制
limit_per_host=30, # 每个主机的连接数限制
ttl_dns_cache=300, # DNS缓存时间（秒）
use_dns_cache=True,
enable_cleanup_closed=True, # 清理关闭的连接
force_close=True # 强制关闭连接
)
# 请求超时设置
self.timeout = aiohttp.ClientTimeout(
total=30, # 总超时时间
connect=10, # 连接超时
sock_read=15, # 套接字读取超时
sock_connect=10 # 套接字连接超时
)
self.session = aiohttp.ClientSession(
timeout=self.timeout,
connector=self.connector,
headers={
'User-Agent': 'AsyncHttpClient/1.0',
'Accept': 'application/json'
}
)
async def fetch_json(self, url: str) -> Optional[Dict]:
"""获取JSON数据"""
try:
async with self.session.get(url) as response:
if response.status == 200:
return await response.json()
else:
print(f"HTTP {response.status} for {url}")
return None
except Exception as e:
print(f"Error fetching {url}: {e}")
return None
async def fetch_multiple_json(self, urls: List[str]) -> List[Optional[Dict]]:
"""并发获取多个JSON数据"""
# 使用任务列表确保并发执行
tasks = [self.fetch_json(url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [result if not isinstance(result, Exception) else None for result in results]
async def fetch_with_retries(self, url: str, max_retries: int = 3) -> Optional[Dict]:
"""带重试机制的请求"""
for attempt in range(max_retries):
try:
async with self.session.get(url) as response:
if response.status == 200:
return await response.json()
elif response.status >= 500: # 服务器错误，重试
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt) # 指数退避
continue
else:
print(f"HTTP {response.status} for {url}")
return None
except Exception as e:
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt)
continue
else:
print(f"Failed after {max_retries} attempts: {e}")
return None
return None
async def close(self):
"""关闭会话"""
await self.session.close()
# 使用示例
async def demo_optimized_client():
client = OptimizedAsyncHttpClient()
try:
urls = [
'https://jsonplaceholder.typicode.com/posts/1',
'https://jsonplaceholder.typicode.com/posts/2',
'https://jsonplaceholder.typicode.com/posts/3'
]
# 并发获取JSON数据
results = await client.fetch_multiple_json(urls)
print("Results:", len([r for r in results if r is not None]))
# 带重试的请求
result = await client.fetch_with_retries('https://httpbin.org/delay/1')
print("Retry result:", result is not None)
finally:
await client.close()
# asyncio.run(demo_optimized_client())
高性能网络服务构建
使用FastAPI构建异步API服务
FastAPI是一个现代、快速（高性能）的Web框架，支持异步编程。
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import List, Optional
import asyncio
import time
from datetime import datetime
# 数据模型定义
class User(BaseModel):
id: int
name: str
email: str
created_at: datetime
class UserCreate(BaseModel):
name: str
email: str
# 模拟数据库存储
fake_database = {
1: User(id=1, name="Alice", email="alice@example.com", created_at=datetime.now()),
2: User(id=2, name="Bob", email="bob@example.com", created_at=datetime.now())
}
app = FastAPI(title="Async API Service", version="1.0.0")
# 异步数据处理函数
async def simulate_database_operation():
"""模拟数据库操作延迟"""
await asyncio.sleep(0.1) # 模拟I/O等待
async def simulate_network_call(url: str):
"""模拟网络调用"""
await asyncio.sleep(0.2)
return f"Response from {url}"
# 异步路由处理
@app.get("/")
async def root():
"""根路由"""
return {"message": "Hello Async World!"}
@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""获取单个用户"""
await simulate_database_operation()
user = fake_database.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.get("/users")
async def get_users(skip: int = 0, limit: int = 100):
"""获取用户列表"""
await simulate_database_operation()
users = list(fake_database.values())[skip:skip+limit]
return {"users": users, "count": len(users)}
@app.post("/users")
async def create_user(user: UserCreate, background_tasks: BackgroundTasks):
"""创建用户"""
# 模拟异步处理
await simulate_database_operation()
new_id = max(fake_database.keys()) + 1 if fake_database else 1
new_user = User(
id=new_id,
name=user.name,
email=user.email,
created_at=datetime.now()
)
fake_database[new_id] = new_user
# 后台任务处理
background_tasks.add_task(send_welcome_email, new_user.email)
return new_user
async def send_welcome_email(email: str):
"""发送欢迎邮件（后台任务）"""
await asyncio.sleep(1) # 模拟邮件发送延迟
print(f"Welcome email sent to {email}")
@app.get("/health")
async def health_check():
"""健康检查"""
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
@app.get("/slow-endpoint")
async def slow_endpoint():
"""慢速端点测试并发性能"""
# 模拟长时间运行的异步操作
await asyncio.sleep(2)
return {"message": "Slow operation completed"}
# 并发处理示例
@app.get("/concurrent-test")
async def concurrent_test():
"""并发测试端点"""
start_time = time.time()
# 并发执行多个异步操作
tasks = [
simulate_network_call(f"https://api.example.com/data/{i}")
for i in range(5)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
end_time = time.time()
return {
"results": results,
"execution_time": end_time - start_time
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
异步WebSocket服务
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio
import json
from datetime import datetime
from typing import Dict, Set
app = FastAPI()
# 存储连接的客户端
connected_clients: Set[WebSocket] = set()
class MessageHandler:
def __init__(self):
self.clients: Dict[str, WebSocket] = {}
async def broadcast_message(self, message: str):
"""广播消息给所有连接的客户端"""
tasks = []
for client in list(self.clients.values()):
try:
tasks.append(client.send_text(message))
except Exception as e:
print(f"Error sending to client: {e}")
# 移除断开连接的客户端
if client in self.clients.values():
await self.remove_client(client)
await asyncio.gather(*tasks, return_exceptions=True)
async def add_client(self, websocket: WebSocket, client_id: str):
"""添加客户端"""
self.clients[client_id] = websocket
connected_clients.add(websocket)
print(f"Client {client_id} connected")
async def remove_client(self, websocket: WebSocket):
"""移除客户端"""
if websocket in connected_clients:
connected_clients.remove(websocket)
# 从clients字典中移除
client_ids = [id for id, ws in self.clients.items() if ws == websocket]
for client_id in client_ids:
del self.clients[client_id]
print(f"Client disconnected")
message_handler = MessageHandler()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
"""WebSocket端点"""
await websocket.accept()
# 添加客户端
await message_handler.add_client(websocket, client_id)
try:
while True:
# 接收消息
data = await websocket.receive_text()
# 解析消息
try:
message_data = json.loads(data)
except json.JSONDecodeError:
message_data = {"text": data}
# 构造响应消息
response = {
"client_id": client_id,
"received_at": datetime.now().isoformat(),
"message": message_data.get("text", "No text provided")
}
# 广播给所有客户端
await message_handler.broadcast_message(json.dumps(response))
except WebSocketDisconnect:
print(f"Client {client_id} disconnected")
await message_handler.remove_client(websocket)
except Exception as e:
print(f"WebSocket error: {e}")
await message_handler.remove_client(websocket)
@app.get("/ws/stats")
async def websocket_stats():
"""获取WebSocket连接统计"""
return {
"connected_clients": len(connected_clients),
"registered_clients": len(message_handler.clients)
}
# 异步任务管理
async def periodic_task():
"""定期任务示例"""
while True:
await asyncio.sleep(30) # 每30秒执行一次
# 广播系统消息
message = {
"type": "system",
"timestamp": datetime.now().isoformat(),
"content": "Periodic system message"
}
await message_handler.broadcast_message(json.dumps(message))
# 启动后台任务
@app.on_event("startup")
async def startup_event():
"""应用启动时执行"""
# 启动定期任务
asyncio.create_task(periodic_task())
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
性能优化最佳实践
异步编程中的性能监控
import asyncio
import time
from functools import wraps
from typing import Any, Callable
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def async_timer(func: Callable) -> Callable:
"""异步函数执行时间装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs) -> Any:
start_time = time.time()
try:
result = await func(*args, **kwargs)
execution_time = time.time() - start_time
logger.info(f"{func.__name__} executed in {execution_time:.4f}s")
return result
except Exception as e:
execution_time = time.time() - start_time
logger.error(f"{func.__name__} failed after {execution_time:.4f}s: {e}")
raise
return wrapper
class AsyncPerformanceMonitor:
"""异步性能监控器"""
def __init__(self):
self.metrics = {}
async def monitor_operation(self, operation_name: str, func: Callable, *args, **kwargs):
"""监控操作执行时间"""
start_time = time.perf_counter()
try:
result = await func(*args, **kwargs)
end_time = time.perf_counter()
execution_time = end_time - start_time
# 记录指标
if operation_name not in self.metrics:
self.metrics[operation_name] = []
self.metrics[operation_name].append(execution_time)
logger.info(f"Operation '{operation_name}' completed in {execution_time:.4f}s")
return result
except Exception as e:
end_time = time.perf_counter()
execution_time = end_time - start_time
logger.error(f"Operation '{operation_name}' failed after {execution_time:.4f}s: {e}")
raise
def get_statistics(self, operation_name: str) -> Dict[str, float]:
"""获取操作统计信息"""
if operation_name not in self.metrics:
return {}
times = self.metrics[operation_name]
return {
"count": len(times),
"min": min(times),
"max": max(times),
"avg": sum(times) / len(times),
"total": sum(times)
}
# 使用示例
async def sample_async_operation(name: str, delay: float = 0.1):
"""示例异步操作"""
await asyncio.sleep(delay)
return f"Operation {name} completed"
async def demo_performance_monitor():
monitor = AsyncPerformanceMonitor()
# 执行多次操作
operations = [
("operation_1", 0.1),
("operation_2", 0.2),
("operation_3", 0.15)
]
for name, delay in operations:
await monitor.monitor_operation(name, sample_async_operation, name, delay)
# 获取统计信息
for operation_name in ["operation_1", "operation_2", "operation_3"]:
stats = monitor.get_statistics(operation_name)
if stats:
print(f"{operation_name}: {stats}")
# asyncio.run(demo_performance_monitor())
异步并发控制
import asyncio
from typing import List, Any
import time
class AsyncConcurrencyController:
"""异步并发控制器"""
def __init__(self, max_concurrent: int = 10):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.max_concurrent = max_concurrent
评论 (0)