WebSocket实时通信应用开发指南：3步构建高性能双向通信系统-尧图网络科技

WebSocket实时通信应用开发指南：3步构建高性能双向通信系统

【免费下载链接】websocketsLibrary for building WebSocket servers and clients in Python项目地址: https://gitcode.com/gh_mirrors/we/websockets

WebSockets是一个专注于正确性、简单性、鲁棒性和性能的Python WebSocket库，为开发者提供了构建实时双向通信应用的完整解决方案。基于Python标准异步I/O框架asyncio，WebSockets提供了优雅的协程API，同时支持线程实现和Sans-I/O实现，满足不同场景下的实时通信需求。无论是构建实时聊天应用、在线游戏、金融交易系统还是物联网设备管理平台，WebSockets都能提供稳定高效的WebSocket协议实现。

WebSocket架构设计与技术选型

传统HTTP轮询的瓶颈与WebSocket解决方案

在实时应用开发中，传统的HTTP轮询方案面临显著的性能瓶颈。每次请求都需要建立完整的HTTP连接，包含大量重复的头部信息，导致带宽浪费和延迟增加。长轮询虽然有所改进，但仍然无法实现真正的实时双向通信。

WebSocket协议通过单一TCP连接提供全双工通信通道，解决了这些根本问题。WebSockets库基于RFC 6455和RFC 7692标准实现，提供以下核心优势：

低延迟通信：建立连接后保持持久连接，无需重复握手
双向数据流：服务器可以主动向客户端推送数据
轻量级帧结构：最小化协议开销，提高传输效率
自动连接管理：内置心跳机制和连接恢复能力

多架构支持：选择最适合的实现方式

WebSockets提供三种不同的实现架构，满足不同应用场景的需求：

asyncio API（默认）：基于Python标准异步I/O框架，适合现代异步应用开发：

import asyncio from websockets.asyncio.server import serve async def echo(websocket): async for message in websocket: await websocket.send(message) async def main(): async with serve(echo, "localhost", 8765) as server: await server.serve_forever() asyncio.run(main())

同步API：基于线程的实现，适合传统同步代码库或需要阻塞式API的场景：

from websockets.sync.client import connect def hello(): with connect("ws://localhost:8765") as websocket: websocket.send("Hello world!") message = websocket.recv() print(f"Received: {message}") hello()

Sans-I/O实现：提供协议层的纯逻辑实现，适合需要自定义传输层的场景。

配置WebSocket连接的3个关键参数

1. 连接超时与重连机制配置

在实际生产环境中，网络不稳定是常见问题。WebSockets提供了完善的连接管理配置：

import asyncio from websockets.asyncio.client import connect async def robust_connection(): # 配置连接参数 async with connect( "ws://localhost:8765", # 连接超时设置 open_timeout=10.0, # 关闭超时设置 close_timeout=5.0, # 最大消息大小（防止内存耗尽） max_size=2**20, # 1MB # 自动ping/pong保持连接 ping_interval=20.0, ping_timeout=20.0, # 压缩扩展支持 compression="deflate" ) as websocket: # 业务逻辑 await websocket.send("ping") response = await websocket.recv() print(f"Response: {response}")

2. 安全认证与TLS配置

对于生产环境，安全配置至关重要。WebSockets支持完整的TLS加密和认证机制：

import ssl import asyncio from websockets.asyncio.client import connect async def secure_connection(): # 创建SSL上下文 ssl_context = ssl.create_default_context() ssl_context.check_hostname = True ssl_context.verify_mode = ssl.CERT_REQUIRED async with connect( "wss://secure-server.com:8765", ssl=ssl_context, # 额外的安全头 extra_headers={ "Authorization": "Bearer your-token-here", "User-Agent": "MyWebSocketClient/1.0" } ) as websocket: # 安全通信 await websocket.send("secure message")

3. 性能优化与资源管理

大型实时应用需要精细的资源管理策略：

import asyncio from websockets.asyncio.server import serve from websockets.asyncio.client import connect # 服务器端配置 async def optimized_server(): async with serve( handler, "localhost", 8765, # 连接限制 max_size=16 * 1024 * 1024, # 16MB # 内存优化 read_limit=2**16, # 64KB write_limit=2**16, # 64KB # 压缩配置 compression="deflate", # 连接空闲超时 timeout=30.0 ) as server: await server.serve_forever() # 客户端连接池管理 class ConnectionPool: def __init__(self, max_connections=10): self.max_connections = max_connections self.connections = [] async def get_connection(self, uri): if len(self.connections) < self.max_connections: connection = await connect(uri) self.connections.append(connection) return connection # 连接池管理逻辑...

实时应用开发实战：构建在线协作计数器

问题场景：多用户实时状态同步

在协作编辑、实时仪表盘或多人游戏中，需要将状态变化实时同步给所有连接的客户端。传统方案使用轮询会导致延迟和不一致性。

解决方案：WebSocket广播机制

WebSockets提供了高效的广播机制，确保所有客户端实时接收状态更新：

import asyncio import json import logging from websockets.asyncio.server import broadcast, serve logging.basicConfig(level=logging.INFO) # 全局状态管理 CONNECTED_USERS = set() COUNTER_VALUE = 0 def create_users_event(): """创建用户数量变化事件""" return json.dumps({ "type": "users", "count": len(CONNECTED_USERS), "timestamp": asyncio.get_event_loop().time() }) def create_value_event(): """创建计数器值变化事件""" return json.dumps({ "type": "value", "value": COUNTER_VALUE, "timestamp": asyncio.get_event_loop().time() }) async def handle_counter(websocket): """处理计数器连接的协程""" global CONNECTED_USERS, COUNTER_VALUE try: # 注册新用户 CONNECTED_USERS.add(websocket) logging.info(f"用户连接，当前用户数: {len(CONNECTED_USERS)}") # 广播用户数量变化 await broadcast(CONNECTED_USERS, create_users_event()) # 发送当前状态给新用户 await websocket.send(create_value_event()) # 处理用户消息 async for message in websocket: try: event = json.loads(message) if event["action"] == "increment": COUNTER_VALUE += 1 logging.info(f"计数器增加: {COUNTER_VALUE}") await broadcast(CONNECTED_USERS, create_value_event()) elif event["action"] == "decrement": COUNTER_VALUE -= 1 logging.info(f"计数器减少: {COUNTER_VALUE}") await broadcast(CONNECTED_USERS, create_value_event()) elif event["action"] == "reset": COUNTER_VALUE = 0 logging.info("计数器重置") await broadcast(CONNECTED_USERS, create_value_event()) else: logging.warning(f"未知操作: {event}") await websocket.send(json.dumps({ "type": "error", "message": "未知操作" })) except json.JSONDecodeError: logging.error("消息格式错误") await websocket.send(json.dumps({ "type": "error", "message": "无效的JSON格式" })) except Exception as e: logging.error(f"连接处理错误: {e}") finally: # 用户断开连接 if websocket in CONNECTED_USERS: CONNECTED_USERS.remove(websocket) logging.info(f"用户断开，剩余用户数: {len(CONNECTED_USERS)}") await broadcast(CONNECTED_USERS, create_users_event()) async def main(): """主服务器函数""" server = await serve( handle_counter, "0.0.0.0", # 监听所有接口 6789, # 生产环境配置 ping_interval=20, ping_timeout=20, max_size=10 * 1024 * 1024, # 10MB最大消息 compression="deflate" ) logging.info("计数器服务器启动在 ws://0.0.0.0:6789") try: await server.serve_forever() except KeyboardInterrupt: logging.info("服务器关闭") finally: await server.close() if __name__ == "__main__": asyncio.run(main())

客户端实现：实时状态同步

import asyncio import json import logging from websockets.asyncio.client import connect logging.basicConfig(level=logging.INFO) class CounterClient: def __init__(self, uri="ws://localhost:6789"): self.uri = uri self.websocket = None async def connect(self): """连接到服务器""" try: self.websocket = await connect( self.uri, ping_interval=20, ping_timeout=20, max_size=10 * 1024 * 1024 ) logging.info("成功连接到服务器") return True except Exception as e: logging.error(f"连接失败: {e}") return False async def send_action(self, action): """发送操作到服务器""" if not self.websocket: logging.error("未连接到服务器") return False try: message = json.dumps({"action": action}) await self.websocket.send(message) logging.info(f"发送操作: {action}") return True except Exception as e: logging.error(f"发送失败: {e}") return False async def listen(self): """监听服务器消息""" try: async for message in self.websocket: data = json.loads(message) if data["type"] == "value": print(f"计数器值更新: {data['value']}") elif data["type"] == "users": print(f"在线用户数: {data['count']}") elif data["type"] == "error": print(f"错误: {data['message']}") except Exception as e: logging.error(f"监听错误: {e}") async def run(self): """运行客户端""" if not await self.connect(): return # 启动监听任务 listener = asyncio.create_task(self.listen()) # 交互循环 try: while True: print("\n可用操作: increment, decrement, reset, quit") action = input("请输入操作: ").strip().lower() if action == "quit": break elif action in ["increment", "decrement", "reset"]: await self.send_action(action) else: print("未知操作") except KeyboardInterrupt: print("\n客户端关闭") finally: listener.cancel() if self.websocket: await self.websocket.close() async def main(): client = CounterClient() await client.run() if __name__ == "__main__": asyncio.run(main())

性能优化与生产环境部署

连接管理与资源优化

在生产环境中，WebSocket服务器需要处理大量并发连接。以下是关键优化策略：

import asyncio import logging from websockets.asyncio.server import serve from websockets.asyncio.router import Router logging.basicConfig(level=logging.INFO) class OptimizedServer: def __init__(self, host="0.0.0.0", port=8765): self.host = host self.port = port self.router = Router() self.connection_stats = { "total_connections": 0, "active_connections": 0, "messages_processed": 0 } async def handle_connection(self, websocket, path): """优化后的连接处理器""" self.connection_stats["total_connections"] += 1 self.connection_stats["active_connections"] += 1 client_id = id(websocket) logging.info(f"客户端 {client_id} 连接，路径: {path}") try: # 设置接收消息超时 websocket.recv_timeout = 30.0 async for message in websocket: self.connection_stats["messages_processed"] += 1 # 消息处理逻辑 await self.process_message(websocket, message, path) # 定期清理无效连接 if self.connection_stats["messages_processed"] % 1000 == 0: await self.cleanup_inactive_connections() except asyncio.TimeoutError: logging.warning(f"客户端 {client_id} 超时断开") except Exception as e: logging.error(f"客户端 {client_id} 错误: {e}") finally: self.connection_stats["active_connections"] -= 1 logging.info(f"客户端 {client_id} 断开") async def process_message(self, websocket, message, path): """消息处理逻辑""" # 根据路径路由到不同的处理器 if path == "/chat": await self.handle_chat(websocket, message) elif path == "/notifications": await self.handle_notification(websocket, message) else: await websocket.send(f"未知路径: {path}") async def handle_chat(self, websocket, message): """聊天处理器""" # 实现聊天逻辑 response = f"收到消息: {message}" await websocket.send(response) async def handle_notification(self, websocket, message): """通知处理器""" # 实现通知逻辑 await websocket.send(f"通知已处理: {message}") async def cleanup_inactive_connections(self): """清理无效连接""" # 实现连接清理逻辑 logging.info("执行连接清理") async def start(self): """启动优化服务器""" server = await serve( self.handle_connection, self.host, self.port, # 性能优化配置 max_size=16 * 1024 * 1024, # 16MB max_queue=32, # 消息队列大小 read_limit=2**20, # 1MB读取限制 write_limit=2**20, # 1MB写入限制 compression="deflate", # 连接限制 max_connections=10000, # 超时设置 ping_interval=20, ping_timeout=20, close_timeout=10, # 日志配置 logger=logging.getLogger("websockets") ) logging.info(f"优化服务器启动在 ws://{self.host}:{self.port}") try: await server.serve_forever() except KeyboardInterrupt: logging.info("服务器关闭") finally: await server.close() logging.info(f"服务器统计: {self.connection_stats}") async def main(): server = OptimizedServer() await server.start() if __name__ == "__main__": asyncio.run(main())

部署架构与负载均衡

对于高并发场景，需要考虑分布式部署：

多进程部署：使用进程管理器如supervisor或systemd
负载均衡：配合Nginx或HAProxy进行连接分发
会话持久化：使用Redis或数据库存储会话状态
监控告警：集成Prometheus和Grafana进行性能监控

常见问题与解决方案

连接稳定性问题

问题：客户端频繁断开连接，网络抖动导致连接中断。

解决方案：

import asyncio from websockets.asyncio.client import connect from websockets.exceptions import ConnectionClosed class ResilientClient: def __init__(self, uri, max_retries=5): self.uri = uri self.max_retries = max_retries self.retry_delay = 1.0 async def connect_with_retry(self): """带重试机制的连接""" for attempt in range(self.max_retries): try: websocket = await connect( self.uri, open_timeout=10.0, close_timeout=5.0, ping_interval=20.0, ping_timeout=20.0 ) return websocket except Exception as e: if attempt == self.max_retries - 1: raise await asyncio.sleep(self.retry_delay * (2 ** attempt)) async def send_with_retry(self, message): """带重试的消息发送""" websocket = await self.connect_with_retry() try: await websocket.send(message) response = await websocket.recv() return response except ConnectionClosed: # 连接关闭，重新连接 return await self.send_with_retry(message) finally: await websocket.close()

内存泄漏排查

问题：长时间运行后内存持续增长。

诊断工具：

import tracemalloc import logging class MemoryMonitor: def __init__(self): tracemalloc.start() self.snapshots = [] def take_snapshot(self, label=""): """记录内存快照""" snapshot = tracemalloc.take_snapshot() self.snapshots.append((label, snapshot)) # 分析内存使用 top_stats = snapshot.statistics('lineno') logging.info(f"内存快照 {label}:") for stat in top_stats[:10]: logging.info(stat) def compare_snapshots(self, index1, index2): """比较两个快照""" if index1 < len(self.snapshots) and index2 < len(self.snapshots): label1, snapshot1 = self.snapshots[index1] label2, snapshot2 = self.snapshots[index2] stats = snapshot2.compare_to(snapshot1, 'lineno') logging.info(f"内存变化 {label1} -> {label2}:") for stat in stats[:10]: logging.info(stat) # 在WebSocket处理器中使用 async def handle_with_memory_monitor(websocket, path): monitor = MemoryMonitor() monitor.take_snapshot("连接开始") try: async for message in websocket: # 处理消息 await websocket.send(f"Echo: {message}") # 定期记录内存使用 if random.random() < 0.01: # 1%概率记录 monitor.take_snapshot(f"处理消息后") finally: monitor.take_snapshot("连接结束") monitor.compare_snapshots(0, -1)

性能瓶颈分析

使用内置的性能分析工具识别瓶颈：

import cProfile import pstats import io from websockets.asyncio.server import serve def profile_websocket_server(): """性能分析装饰器""" def decorator(func): async def wrapper(*args, **kwargs): pr = cProfile.Profile() pr.enable() try: result = await func(*args, **kwargs) finally: pr.disable() # 输出性能报告 s = io.StringIO() ps = pstats.Stats(pr, stream=s).sort_stats('cumulative') ps.print_stats(20) # 显示前20个最耗时的函数 print(s.getvalue()) return result return wrapper return decorator @profile_websocket_server() async def optimized_handler(websocket, path): """经过性能分析的处理器""" # 业务逻辑 async for message in websocket: # 优化关键路径 processed = await process_message_efficiently(message) await websocket.send(processed)

进阶应用场景与最佳实践

实时数据推送系统

构建股票行情、体育比分等实时数据推送系统：

import asyncio import json import time from datetime import datetime from websockets.asyncio.server import serve, broadcast class RealTimeDataFeed: def __init__(self): self.subscribers = {} self.data_sources = { "stocks": self.generate_stock_data, "sports": self.generate_sports_data, "weather": self.generate_weather_data } async def generate_stock_data(self): """生成模拟股票数据""" return { "symbol": "AAPL", "price": 150.0 + (time.time() % 10), "volume": 1000000, "timestamp": datetime.now().isoformat() } async def generate_sports_data(self): """生成模拟体育数据""" return { "match": "TeamA vs TeamB", "score": {"TeamA": 2, "TeamB": 1}, "time": "45:00", "timestamp": datetime.now().isoformat() } async def generate_weather_data(self): """生成模拟天气数据""" return { "city": "Beijing", "temperature": 20 + (time.time() % 5), "humidity": 60, "timestamp": datetime.now().isoformat() } async def handle_subscription(self, websocket, path): """处理数据订阅""" # 解析订阅主题 topic = path.lstrip('/') if topic not in self.data_sources: await websocket.send(json.dumps({ "error": f"未知主题: {topic}", "available_topics": list(self.data_sources.keys()) })) return # 注册订阅者 if topic not in self.subscribers: self.subscribers[topic] = set() self.subscribers[topic].add(websocket) try: # 发送初始数据 data = await self.data_sources[topic]() await websocket.send(json.dumps(data)) # 保持连接，定期推送更新 while True: await asyncio.sleep(1.0) # 1秒更新频率 data = await self.data_sources[topic]() await websocket.send(json.dumps(data)) except Exception as e: print(f"订阅错误: {e}") finally: # 取消订阅 if topic in self.subscribers and websocket in self.subscribers[topic]: self.subscribers[topic].remove(websocket) async def main(): feed = RealTimeDataFeed() server = await serve( feed.handle_subscription, "localhost", 8765, max_size=10 * 1024 * 1024, compression="deflate" ) print("实时数据推送服务启动在 ws://localhost:8765") await server.serve_forever() if __name__ == "__main__": asyncio.run(main())

微服务间通信

在微服务架构中使用WebSocket进行服务间通信：

import asyncio import json from typing import Dict, Any from websockets.asyncio.client import connect class MicroserviceClient: def __init__(self, service_name, registry_url="ws://registry:8000"): self.service_name = service_name self.registry_url = registry_url self.connections: Dict[str, Any] = {} async def register_service(self): """向注册中心注册服务""" async with connect(self.registry_url) as websocket: await websocket.send(json.dumps({ "action": "register", "service": self.service_name, "endpoint": f"ws://{self.service_name}:8765" })) response = await websocket.recv() print(f"注册响应: {response}") async def discover_service(self, target_service): """发现其他服务""" async with connect(self.registry_url) as websocket: await websocket.send(json.dumps({ "action": "discover", "service": target_service })) response = json.loads(await websocket.recv()) return response.get("endpoint") async def call_service(self, target_service, method, params): """调用远程服务""" if target_service not in self.connections: endpoint = await self.discover_service(target_service) if not endpoint: raise Exception(f"服务未找到: {target_service}") self.connections[target_service] = await connect(endpoint) websocket = self.connections[target_service] await websocket.send(json.dumps({ "method": method, "params": params, "caller": self.service_name })) response = await websocket.recv() return json.loads(response) async def handle_incoming(self, websocket): """处理其他服务的调用""" async for message in websocket: request = json.loads(message) # 根据method路由到不同的处理函数 method = request.get("method") params = request.get("params", {}) caller = request.get("caller") result = await self.route_request(method, params, caller) await websocket.send(json.dumps({ "result": result, "error": None })) async def route_request(self, method, params, caller): """请求路由""" # 实现具体的业务逻辑路由 if method == "get_data": return await self.get_data(params) elif method == "process_data": return await self.process_data(params) else: raise Exception(f"未知方法: {method}")

测试与质量保证

单元测试策略

WebSockets提供了完善的测试工具和模式：

import asyncio import unittest from unittest.mock import AsyncMock, patch from websockets.asyncio.server import serve from websockets.asyncio.client import connect class TestWebSocketServer(unittest.IsolatedAsyncioTestCase): async def test_echo_server(self): """测试回声服务器""" async def echo_handler(websocket, path): async for message in websocket: await websocket.send(f"Echo: {message}") # 启动测试服务器 server = await serve(echo_handler, "localhost", 8765) try: # 测试客户端连接 async with connect("ws://localhost:8765") as websocket: # 发送测试消息 await websocket.send("Hello, World!") response = await websocket.recv() # 验证响应 self.assertEqual(response, "Echo: Hello, World!") # 测试多个消息 await websocket.send("Test Message") response = await websocket.recv() self.assertEqual(response, "Echo: Test Message") finally: await server.close() async def test_broadcast_functionality(self): """测试广播功能""" connected_clients = set() async def broadcast_handler(websocket, path): connected_clients.add(websocket) try: async for message in websocket: # 广播给所有客户端 for client in connected_clients: if client != websocket: await client.send(message) finally: connected_clients.remove(websocket) server = await serve(broadcast_handler, "localhost", 8766) try: # 创建多个客户端 async with connect("ws://localhost:8766") as client1: async with connect("ws://localhost:8766") as client2: async with connect("ws://localhost:8766") as client3: # 客户端1发送消息 await client1.send("Broadcast Test") # 其他客户端应该收到消息 msg2 = await client2.recv() msg3 = await client3.recv() self.assertEqual(msg2, "Broadcast Test") self.assertEqual(msg3, "Broadcast Test") # 客户端1不应该收到自己的广播 with self.assertRaises(asyncio.TimeoutError): await asyncio.wait_for(client1.recv(), timeout=0.5) finally: await server.close() async def test_error_handling(self): """测试错误处理""" async def error_handler(websocket, path): try: async for message in websocket: if message == "raise_error": raise ValueError("Test error") await websocket.send(f"OK: {message}") except Exception as e: await websocket.send(f"Error: {str(e)}") server = await serve(error_handler, "localhost", 8767) try: async with connect("ws://localhost:8767") as websocket: # 测试正常消息 await websocket.send("Normal message") response = await websocket.recv() self.assertEqual(response, "OK: Normal message") # 测试错误消息 await websocket.send("raise_error") response = await websocket.recv() self.assertIn("Error: Test error", response) finally: await server.close() class TestWebSocketClient(unittest.IsolatedAsyncioTestCase): async def test_connection_retry(self): """测试连接重试机制""" with patch('websockets.asyncio.client.connect') as mock_connect: # 模拟第一次连接失败，第二次成功 mock_connect.side_effect = [ ConnectionError("First attempt failed"), AsyncMock() ] client = ResilientClient("ws://test:8765", max_retries=2) # 应该成功连接（经过一次重试） websocket = await client.connect_with_retry() self.assertIsNotNone(websocket) self.assertEqual(mock_connect.call_count, 2) async def test_message_serialization(self): """测试消息序列化""" async def json_handler(websocket, path): async for message in websocket: data = json.loads(message) response = {"echo": data, "timestamp": time.time()} await websocket.send(json.dumps(response)) server = await serve(json_handler, "localhost", 8768) try: async with connect("ws://localhost:8768") as websocket: # 测试复杂JSON数据 test_data = { "id": 123, "name": "Test", "nested": {"key": "value"}, "list": [1, 2, 3] } await websocket.send(json.dumps(test_data)) response = await websocket.recv() result = json.loads(response) self.assertEqual(result["echo"], test_data) self.assertIn("timestamp", result) finally: await server.close() if __name__ == "__main__": unittest.main()

集成测试与性能测试

import asyncio import time import statistics from concurrent.futures import ThreadPoolExecutor from websockets.sync.client import connect as sync_connect class PerformanceTest: def __init__(self, server_url, num_clients=100, messages_per_client=100): self.server_url = server_url self.num_clients = num_clients self.messages_per_client = messages_per_client self.results = [] def sync_client_test(self, client_id): """同步客户端性能测试""" latencies = [] try: with sync_connect(self.server_url) as websocket: for i in range(self.messages_per_client): message = f"Client{client_id}-Message{i}" start_time = time.perf_counter() websocket.send(message) response = websocket.recv() end_time = time.perf_counter() latency = (end_time - start_time) * 1000 # 转换为毫秒 latencies.append(latency) # 验证响应 assert response == f"Echo: {message}", f"响应不匹配: {response}" except Exception as e: print(f"客户端 {client_id} 错误: {e}") return latencies def run_sync_test(self): """运行同步性能测试""" print(f"启动 {self.num_clients} 个同步客户端...") with ThreadPoolExecutor(max_workers=50) as executor: futures = [ executor.submit(self.sync_client_test, i) for i in range(self.num_clients) ] all_latencies = [] for i, future in enumerate(futures): latencies = future.result() all_latencies.extend(latencies) if (i + 1) % 10 == 0: print(f"已完成 {i + 1}/{self.num_clients} 个客户端") # 统计结果 if all_latencies: avg_latency = statistics.mean(all_latencies) p95_latency = statistics.quantiles(all_latencies, n=20)[18] # 95百分位 p99_latency = statistics.quantiles(all_latencies, n=100)[98] # 99百分位 print(f"\n性能测试结果:") print(f" 消息总数: {len(all_latencies)}") print(f" 平均延迟: {avg_latency:.2f}ms") print(f" P95延迟: {p95_latency:.2f}ms") print(f" P99延迟: {p99_latency:.2f}ms") print(f" 最大延迟: {max(all_latencies):.2f}ms") print(f" 最小延迟: {min(all_latencies):.2f}ms") return all_latencies async def async_client_test(self, client_id): """异步客户端性能测试""" latencies = [] try: async with connect(self.server_url) as websocket: for i in range(self.messages_per_client): message = f"AsyncClient{client_id}-Message{i}" start_time = time.perf_counter() await websocket.send(message) response = await websocket.recv() end_time = time.perf_counter() latency = (end_time - start_time) * 1000 latencies.append(latency) assert response == f"Echo: {message}", f"响应不匹配: {response}" except Exception as e: print(f"异步客户端 {client_id} 错误: {e}") return latencies async def run_async_test(self): """运行异步性能测试""" print(f"启动 {self.num_clients} 个异步客户端...") tasks = [ self.async_client_test(i) for i in range(self.num_clients) ] results = await asyncio.gather(*tasks) all_latencies = [latency for sublist in results for latency in sublist] # 统计结果 if all_latencies: avg_latency = statistics.mean(all_latencies) p95_latency = statistics.quantiles(all_latencies, n=20)[18] p99_latency = statistics.quantiles(all_latencies, n=100)[98] print(f"\n异步性能测试结果:") print(f" 消息总数: {len(all_latencies)}") print(f" 平均延迟: {avg_latency:.2f}ms") print(f" P95延迟: {p95_latency:.2f}ms") print(f" P99延迟: {p99_latency:.2f}ms") return all_latencies # 运行性能测试 async def main(): # 先启动测试服务器 async def echo_server(websocket, path): async for message in websocket: await websocket.send(f"Echo: {message}") server = await serve(echo_server, "localhost", 8769) print("测试服务器已启动") try: # 运行性能测试 tester = PerformanceTest("ws://localhost:8769", num_clients=50, messages_per_client=20) print("\n=== 同步客户端测试 ===") sync_results = tester.run_sync_test() print("\n=== 异步客户端测试 ===") async_results = await tester.run_async_test() finally: await server.close() print("测试服务器已关闭") if __name__ == "__main__": asyncio.run(main())