本文主要是介绍Python Websockets库的使用指南,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
《PythonWebsockets库的使用指南》pythonwebsockets库是一个用于创建WebSocket服务器和客户端的Python库,它提供了一种简单的方式来实现实时通信,支持异步和同步...
一、WebSocket 简介
WebSocket 是一种基于 TCP 的协议,支持全双工通信(服务器和客户端可以同时发送消息),适用于实时性要求高的场景(如聊天、实时数据监控、在线游戏等)。与 HTTP 不同,WebSocket 连接一旦建立,会保持持久化,避免了 HTTP 轮询的开销。
二、TMsEcRNWPython 的 websockets 库
websockets
是 Python 中用于构建 WebSocket 服务器和客户端的异步库,基于 asyncio
实现,支持高并发和低延迟通信。
安装
pip install websockets
三、完整代码示例
1. WebSocket 服务器
import asyncio import websockets from datetime import datetime # 保存所有活跃的客户端连接 connected_clients = set() async def handle_client(websocket, path): """ 处理单个客户端连接 """ # 将新客户端添加到集合 connected_clients.add(websocket) try: async for message in websocket: # 广播消息给所有客户端 timestamp = datetime.now().strftime("%H:%M:%S") broadcast_message = f"[{timestamp}] {message}" js await broadcast(broadcast_message) except websockets.exceptions.ConnectionClosed: print("客户端断开连接") finally: # 客户端断开后移除 connected_clients.remove(websocket) async def broadcast(message): """ 向所有连接的客户端广播消息 """ if connected_clients: await asyncio.gather( *[client.send(message) for client in connected_clients] ) async def start_server(): """ 启动 WebSocket 服务器 """ async with websockets.serve(handle_client, "localhost", 8765): print("服务器已启动,监听端口 8765...") await asyncio.Future() # 永久运行 if __name__ == "__main__": asyncio.run(start_server())
2. WebSocket 客户端
import asyncio import websockets async def client(): """ WebSocket 客户端实现 """ async with websockets.connect("ws://localhost:8765") as websocket: # 创建两个任务:一个接收消息,一个发送消息 receive_task = asyncio.create_task(receive_messages(websocket)) senTMsEcRNWd_task = asyncio.create_task(send_messages(websocket)) await asyncio.gather(receive_task, send_task) async def receive_messages(websocket): """ 接收服务器消息 """ async for message in websocket: print(f"\n收到消息: {message}") async def send_messages(websocket): """ 发送用户输入的消息 """ while True: message = await asyncio.get_event_ljsoop().run_in_executor(None, input, "输入消息 (输入 'exit' 退出): ") if message.strip().lower() == 'exit': break await websocket.send(message) if __name__ == "__main__": asyncio.run(client())
四、代码详解
服务器端
connected_clients
:使用集合保存所有活跃连接,便于广播。handle_client
:- 接受新连接并添加到集合。
- 循环接收客户端消息,收到后调用
broadcast
广播。 - 处理连接断开异常,确保客户端被移除。
broadcast
:使用asyncio.gather
并发向所有客户端发送消息。start_server
:启动服务器并永久运行。
客户端
client
:连接服务器并启动收发任务。receive_messages
:循环接收服务器消息并打印。send_messages
:读取用户输入,发送到服务器(run_in_executor
用于异步处理阻塞的input
)。
五、运行与测试
- 启动服务器:
python server.py
- 启动多个客户端:
python client.py
- 测试:在任一客户端输入消息,所有客户端都会收到广播。
六、高级用法
- 消息协议:可定义 jsON 格式的消息,添加类型字段(如
{"type": "chajst", "content": "Hello"}
)。 - 认证:在连接时验证 Token(通过
path
参数或首次握手消息)。 - 心跳机制:定期发送 Ping/Pong 保持连接活跃。
- 集成 FastAPI:结合 ASGI 框架提供 HTTP + WebSocket 混合服务。
七、应用场景
- 实时聊天应用
- 多人在线游戏
- 股票行情推送
- 物联网设备监控
八、总结
通过 websockets
库,可以轻松构建高性能的实时应用。本文提供的代码示例覆盖了服务器和客户端的基本实现,注释详细,可直接扩展用于实际项目。
到此这篇关于Python Websockets库的使用指南的文章就介绍到这了,更多相关Python Websockets库使用内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程China编程(www.chinasem.cn)!
这篇关于Python Websockets库的使用指南的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!