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

Python异步编程详解

news 来源：原创 2025/5/21 5:09:10

Python异步编程详解

在这里插入图片描述

引言

异步编程是Python中处理并发操作的重要方式，它允许程序在等待I/O操作时执行其他任务，从而提高程序的整体效率。本文将详细介绍Python异步编程的概念、实现方式以及实际应用场景。

1. 异步编程基础

1.1 什么是异步编程？

异步编程是一种非阻塞的编程模式，它允许程序在等待某个操作完成时继续执行其他任务。与传统的同步编程相比，异步编程更适合处理I/O密集型任务。

# 同步方式
def sync_function():result1 = long_running_operation1()  # 阻塞result2 = long_running_operation2()  # 阻塞return result1 + result2# 异步方式
async def async_function():result1 = await long_running_operation1()  # 非阻塞result2 = await long_running_operation2()  # 非阻塞return result1 + result2

1.2 异步编程的优势

提高I/O密集型应用的性能
更好的资源利用率
支持高并发操作
避免线程切换开销

2. 异步编程的实现

2.1 使用asyncio

import asyncioasync def main():# 创建任务task1 = asyncio.create_task(say_after(1, 'hello'))task2 = asyncio.create_task(say_after(2, 'world'))# 等待任务完成await task1await task2async def say_after(delay, what):await asyncio.sleep(delay)print(what)# 运行主函数
asyncio.run(main())

2.2 异步函数定义

import asyncioasync def fetch_data(url):# 模拟网络请求await asyncio.sleep(1)return f"Data from {url}"async def process_data(data):# 模拟数据处理await asyncio.sleep(0.5)return f"Processed: {data}"async def main():# 并发执行多个任务urls = ['url1', 'url2', 'url3']tasks = [fetch_data(url) for url in urls]results = await asyncio.gather(*tasks)# 处理结果processed_results = await asyncio.gather(*[process_data(result) for result in results])print(processed_results)asyncio.run(main())

3. 实际应用场景

3.1 网络请求

import aiohttp
import asyncioasync def fetch_url(session, url):async with session.get(url) as response:return await response.text()async def main():async with aiohttp.ClientSession() as session:urls = ['http://example.com/1','http://example.com/2','http://example.com/3']tasks = [fetch_url(session, url) for url in urls]results = await asyncio.gather(*tasks)return results# 运行示例
results = asyncio.run(main())

3.2 文件操作

import asyncio
import aiofilesasync def read_file(filename):async with aiofiles.open(filename, mode='r') as f:return await f.read()async def write_file(filename, content):async with aiofiles.open(filename, mode='w') as f:await f.write(content)async def main():# 并发读写文件content = await read_file('input.txt')await write_file('output.txt', content.upper())asyncio.run(main())

4. 高级特性

4.1 超时控制

import asyncioasync def long_running_task():await asyncio.sleep(10)return "Task completed"async def main():try:# 设置超时时间result = await asyncio.wait_for(long_running_task(), timeout=5.0)print(result)except asyncio.TimeoutError:print("Task timed out")asyncio.run(main())

4.2 并发限制

import asyncio
import aiohttpasync def fetch_with_semaphore(sem, session, url):async with sem:async with session.get(url) as response:return await response.text()async def main():# 限制并发数为3sem = asyncio.Semaphore(3)async with aiohttp.ClientSession() as session:tasks = [fetch_with_semaphore(sem, session, url)for url in ['http://example.com'] * 10]results = await asyncio.gather(*tasks)return resultsasyncio.run(main())

5. 最佳实践

5.1 错误处理

import asyncioasync def risky_operation():if random.random() < 0.5:raise ValueError("Random error")return "Success"async def main():try:result = await risky_operation()print(result)except ValueError as e:print(f"Error occurred: {e}")except Exception as e:print(f"Unexpected error: {e}")asyncio.run(main())

5.2 资源管理

import asyncio
from contextlib import asynccontextmanager@asynccontextmanager
async def managed_resource():print("Acquiring resource")try:yield "resource"finally:print("Releasing resource")async def main():async with managed_resource() as resource:print(f"Using {resource}")await asyncio.sleep(1)asyncio.run(main())

6. 实际应用示例

6.1 异步Web服务器

from aiohttp import web
import asyncioasync def handle(request):name = request.match_info.get('name', "Anonymous")text = f"Hello, {name}"return web.Response(text=text)async def main():app = web.Application()app.router.add_get('/', handle)app.router.add_get('/{name}', handle)runner = web.AppRunner(app)await runner.setup()site = web.TCPSite(runner, 'localhost', 8080)await site.start()print("Server started at http://localhost:8080")while True:await asyncio.sleep(3600)  # 运行1小时asyncio.run(main())

6.2 异步数据库操作

import asyncio
import asyncpgasync def main():# 创建连接池pool = await asyncpg.create_pool(user='postgres',password='password',database='mydb',host='localhost')async with pool.acquire() as conn:# 执行查询rows = await conn.fetch('SELECT * FROM users')for row in rows:print(row)await pool.close()asyncio.run(main())

练习

1. 实现一个异步的聊天服务器

import asyncio
import json
from datetime import datetimeclass ChatServer:def __init__(self, host='localhost', port=8888):self.host = hostself.port = portself.clients = {}  # 存储客户端连接self.messages = []  # 存储聊天记录async def handle_client(self, reader, writer):# 获取客户端地址addr = writer.get_extra_info('peername')print(f"New connection from {addr}")# 生成客户端IDclient_id = f"client_{len(self.clients)}"self.clients[client_id] = writertry:# 发送欢迎消息welcome_msg = {'type': 'system','content': f'Welcome! Your ID is {client_id}','timestamp': datetime.now().strftime('%H:%M:%S')}writer.write(json.dumps(welcome_msg).encode() + b'\n')await writer.drain()# 广播新用户加入await self.broadcast({'type': 'system','content': f'User {client_id} joined the chat','timestamp': datetime.now().strftime('%H:%M:%S')}, exclude=client_id)# 处理客户端消息while True:data = await reader.readline()if not data:breakmessage = json.loads(data.decode())message['timestamp'] = datetime.now().strftime('%H:%M:%S')message['sender'] = client_id# 存储消息self.messages.append(message)# 广播消息await self.broadcast(message)except Exception as e:print(f"Error handling client {client_id}: {e}")finally:# 清理客户端连接del self.clients[client_id]writer.close()await writer.wait_closed()# 广播用户离开await self.broadcast({'type': 'system','content': f'User {client_id} left the chat','timestamp': datetime.now().strftime('%H:%M:%S')})print(f"Connection from {addr} closed")async def broadcast(self, message, exclude=None):"""广播消息给所有客户端"""message_str = json.dumps(message) + '\n'for client_id, writer in self.clients.items():if client_id != exclude:try:writer.write(message_str.encode())await writer.drain()except Exception as e:print(f"Error broadcasting to {client_id}: {e}")async def start(self):"""启动服务器"""server = await asyncio.start_server(self.handle_client, self.host, self.port)print(f"Chat server running on {self.host}:{self.port}")async with server:await server.serve_forever()# 客户端代码
class ChatClient:def __init__(self, host='localhost', port=8888):self.host = hostself.port = portself.reader = Noneself.writer = Noneasync def connect(self):"""连接到服务器"""self.reader, self.writer = await asyncio.open_connection(self.host, self.port)print("Connected to chat server")async def receive_messages(self):"""接收消息"""while True:try:data = await self.reader.readline()if not data:breakmessage = json.loads(data.decode())self.display_message(message)except Exception as e:print(f"Error receiving message: {e}")breakdef display_message(self, message):"""显示消息"""timestamp = message['timestamp']if message['type'] == 'system':print(f"[{timestamp}] System: {message['content']}")else:sender = message['sender']content = message['content']print(f"[{timestamp}] {sender}: {content}")async def send_message(self, content):"""发送消息"""if self.writer:message = {'type': 'message','content': content}self.writer.write(json.dumps(message).encode() + b'\n')await self.writer.drain()async def run(self):"""运行客户端"""await self.connect()# 启动接收消息的任务receive_task = asyncio.create_task(self.receive_messages())try:while True:message = input("> ")if message.lower() == 'quit':breakawait self.send_message(message)except KeyboardInterrupt:passfinally:if self.writer:self.writer.close()await self.writer.wait_closed()receive_task.cancel()# 使用示例
async def main():# 启动服务器server = ChatServer()server_task = asyncio.create_task(server.start())# 等待服务器启动await asyncio.sleep(1)# 启动客户端client = ChatClient()await client.run()# 清理server_task.cancel()if __name__ == "__main__":asyncio.run(main())

2. 创建一个支持并发下载的下载管理器

import asyncio
import aiohttp
import os
from urllib.parse import urlparse
from typing import List, Dictclass DownloadManager:def __init__(self, max_concurrent: int = 3):self.max_concurrent = max_concurrentself.semaphore = asyncio.Semaphore(max_concurrent)self.downloads: Dict[str, float] = {}  # 存储下载进度async def download_file(self, url: str, save_path: str = None):"""下载单个文件"""if save_path is None:save_path = os.path.basename(urlparse(url).path)async with self.semaphore:try:async with aiohttp.ClientSession() as session:async with session.get(url) as response:if response.status == 200:total_size = int(response.headers.get('content-length', 0))downloaded = 0with open(save_path, 'wb') as f:async for chunk in response.content.iter_chunked(8192):f.write(chunk)downloaded += len(chunk)self.downloads[url] = downloaded / total_size if total_size > 0 else 1print(f"Downloaded {save_path}")return Trueelse:print(f"Failed to download {url}: HTTP {response.status}")return Falseexcept Exception as e:print(f"Error downloading {url}: {e}")return Falseasync def download_files(self, urls: List[str], save_dir: str = None):"""并发下载多个文件"""if save_dir:os.makedirs(save_dir, exist_ok=True)tasks = []for url in urls:save_path = os.path.join(save_dir, os.path.basename(urlparse(url).path)) if save_dir else Nonetasks.append(self.download_file(url, save_path))results = await asyncio.gather(*tasks)return all(results)def get_progress(self, url: str) -> float:"""获取下载进度"""return self.downloads.get(url, 0)# 使用示例
async def main():downloader = DownloadManager(max_concurrent=3)urls = ['http://example.com/file1.zip','http://example.com/file2.zip','http://example.com/file3.zip']# 开始下载success = await downloader.download_files(urls, save_dir='downloads')if success:print("All downloads completed successfully")else:print("Some downloads failed")if __name__ == "__main__":asyncio.run(main())

3. 实现一个异步的任务队列

import asyncio
from typing import Any, Callable, Dict, List
from datetime import datetime
import uuidclass AsyncTaskQueue:def __init__(self, max_workers: int = 3):self.max_workers = max_workersself.semaphore = asyncio.Semaphore(max_workers)self.tasks: Dict[str, asyncio.Task] = {}self.results: Dict[str, Any] = {}self.errors: Dict[str, Exception] = {}async def add_task(self, func: Callable, *args, **kwargs) -> str:"""添加任务到队列"""task_id = str(uuid.uuid4())async def wrapped_task():async with self.semaphore:try:result = await func(*args, **kwargs)self.results[task_id] = resultexcept Exception as e:self.errors[task_id] = efinally:del self.tasks[task_id]self.tasks[task_id] = asyncio.create_task(wrapped_task())return task_idasync def get_result(self, task_id: str, timeout: float = None) -> Any:"""获取任务结果"""if task_id in self.results:return self.results[task_id]if task_id in self.errors:raise self.errors[task_id]if task_id in self.tasks:try:await asyncio.wait_for(self.tasks[task_id], timeout)return self.results.get(task_id)except asyncio.TimeoutError:raise TimeoutError(f"Task {task_id} timed out")raise KeyError(f"Task {task_id} not found")async def cancel_task(self, task_id: str):"""取消任务"""if task_id in self.tasks:self.tasks[task_id].cancel()try:await self.tasks[task_id]except asyncio.CancelledError:passdef get_task_status(self, task_id: str) -> str:"""获取任务状态"""if task_id in self.results:return "completed"if task_id in self.errors:return "failed"if task_id in self.tasks:return "running"return "unknown"# 使用示例
async def example_task(task_id: int, delay: float) -> str:"""示例任务"""await asyncio.sleep(delay)return f"Task {task_id} completed at {datetime.now()}"async def main():queue = AsyncTaskQueue(max_workers=2)# 添加任务task_ids = []for i in range(5):task_id = await queue.add_task(example_task, i, 1.0)task_ids.append(task_id)print(f"Added task {i} with ID {task_id}")# 等待所有任务完成for task_id in task_ids:try:result = await queue.get_result(task_id)print(f"Task {task_id} result: {result}")except Exception as e:print(f"Task {task_id} failed: {e}")if __name__ == "__main__":asyncio.run(main())