行业知识
如何使用Python实现WebScoket
2026-10-21 09:07:52
来源：互联网转载
大家好，后���开发领域迅速发展，需要满足今天应用程序多样化需求的协议。默认的HTTP协议设计用于无状态和短生命周期请求，但在需要实时交互的场景中（如实时信令、聊天应用和协同编辑），它显然不够。
（）为了解决这些局限，开发人员正在探索更好满足现代应用程序动态需求的替代协议。
所有提到的限制都可以通过使用构建在TCP协议之上的WebSocket来克服。让我们深入了解一下它们的一些关键特性。
（）全双工通信
WebSocket支持同时进行的双向通信，客户端和服务器都可以独立发送消息，无需等待请求-响应周期。
低延迟
WebSocket建立的持久连接显著降低了与传统HTTP连接相比的延迟。这使得WebSocket非常适用于实时应用，如聊天应用、在线游戏、实时通知和协同编辑。
资源有效利用
WebSocket消除了为每次通信重复打开和关闭连接的需要。长寿命的连接减少了建立新连接的开销，提高了资源利用效率。
广泛浏览器支持
现代Web浏览器支持WebSocket，使其成为实时Web应用程序广泛采用的技术。大多数编程语言和Web框架还提供处理WebSocket连接的库或模块。
安全连接
WebSocket可以在安全连接上使用（WebSocket Secure，或WSS），使用与保护HTTPS连接相同的TLS/SSL协议。这确保了通信是加密且安全的。
废话不多说，让我们进入代码。
构建WebSocket通信的最简代码
我使用Tornado，它使我能够轻松实现这一点，只需覆盖现有方法。
服务器端
pip install tornado
让我们首先编写最小的代码。
from typing import Awaitable import tornado.ioloop import tornado.web import tornado.websocket class WebSocketHandler(tornado.websocket.WebSocketHandler): """ 此处理程序类用于处理WebSocket连接。 """ def open(self, *args: str, **kwargs: str) -> Awaitable[None] | None: """ 打开WebSocket连接。 覆盖此方法以处理WebSocket的打开。 当打开新的WebSocket时调用此方法。 """ print("WebSocket已打开") return super().open(*args, **kwargs) def on_message(self, message: str | bytes) -> Awaitable[None] | None: """ 处理WebSocket上的传入消息。 覆盖此方法以处理传入的消息。 当收到新的WebSocket消息时调用此方法。 """ print(f"收到消息：{message}") self.write_message(f"你说：{message}") def on_close(self) -> None: """ 处理WebSocket的关闭。 覆盖此方法以处理WebSocket的关闭。 当WebSocket关闭时调用此方法。 """ print("WebSocket已关闭") return super().on_close() def app() -> tornado.web.Application: return tornado.web.Application( [ (r"/websocket", WebSocketHandler), ] ) if __name__ == "__main__": application = app() application.listen(8080) print("监听端口8080") tornado.ioloop.IOLoop.current().start() 代码解析:
- import tornado.ioloop：tornado.ioloop模块提供了用于管理异步操作的I/O循环。I/O循环是Tornado非阻塞架构的核心。它异步处理事件，如传入请求和响应，而不阻塞其他任务的执行。
- import tornado.web：tornado.web模块包含用于构建Web应用程序的类和工具。它提供了关键组件，如RequestHandler类，允许您定义如何响应HTTP请求。此模块对于创建Web应用程序的结构、处理路由和处理传入的HTTP请求至关重要。
- import tornado.websocket：用于处理WebSocket连接的模块。它提供了WebSocketHandler类，允许您定义应用程序如何响应基于WebSocket的通信。
- tornado.web.Application：入口点，用于将各种处理程序、设置和配置绑定在一起，创建完整的Web应用程序。它接受一个列表，包含**(绑定路径，请求处理程序)**。
- application.listen(8080)：开始在端口8080上监听传入请求。
- tornado.ioloop.IOLoop.current().start()：在当前上下文中启动事件循环，以异步处理请求。
运行WebSocket服务器:
python3 websockets.py
客户端
我使用的是Node而不是浏览器控制台，但您也可以使用浏览器控制台。
对于Node，请首先安装支持WebSocket的库。
npm install ws
const WebSocket = require("ws"); var socket = new WebSocket("ws://localhost:8080/websocket"); socket.on("open", function() { console.log("连接已打开") socket.send("你好，服务器！"); }) socket.on("message", function(message) { console.log('新消息 ' + message) if (message.data == 'ping') { socket.send('pong'); } }) socket.on("close", function() { console.log("连接已关闭...") })如果您尝试在浏览器控制台中运行，则需要将跨源设置为True。class WebSocketHandler(tornado.websocket.WebSocketHandler): ... def check_origin(self, origin: str) -> bool: return True
在浏览器中运行
var socket = new WebSocket("ws://localhost:8080/websocket"); socket.onopen = function(event) { console.log("WebSocket连接已打开"); socket.send("你好，服务器！"); }; socket.onmessage = function(event) { console.log("从服务器接收到消息:", event.data); }; socket.onclose = function(event) { console.log("WebSocket连接已关闭"); };WebSocket连接已打开 VM53:9 从服务器接收到消息：你好，服务器！
第一次翻译：
应用: 基于房间的聊天系统
让我们添加一些代码，实现用户房间和用户名功能。
服务器端代码
from typing import Awaitable import tornado.ioloop import tornado.web import tornado.websocket class ChatRoom: def __init__(self): self.clients = set() def add_client(self, client): self.clients.add(client) def remove_client(self, client): self.clients.remove(client) def broadcast(self, message, sender=None): for client in self.clients: if client.user_name != sender.user_name: client.write_message(f"{sender.user_name}: {message}") else: client.write_message("") class WebSocketHandler(tornado.websocket.WebSocketHandler): room_id = None user_name = None def open(self, room_id, user_name) -> Awaitable[None] | None: self.room_id = room_id self.user_name = user_name room = self.application.get_or_create_room(self.room_id) room.add_client(self) print(f"WebSocket为 {self.user_name} 在房间 {self.room_id} 中打开") def on_message(self, message: str | bytes) -> Awaitable[None] | None: print(f"从 {self.user_name} 收到消息：{message}") room = self.application.get_or_create_room(self.room_id) room.broadcast(message, sender=self) def on_close(self) -> None: print(f"WebSocket为 {self.user_name} 关闭") self.application.get_or_create_room(self.room_id).remove_client(self) return super().on_close() def check_origin(self, origin: str) -> bool: return True class ApplicationWithRooms(tornado.web.Application): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.rooms = {} def get_or_create_room(self, room_id): if room_id not in self.rooms: self.rooms[room_id] = ChatRoom() return self.rooms[room_id] def app() -> ApplicationWithRooms: return ApplicationWithRooms( [ (r"/websocket/(\w+)/(\w+)", WebSocketHandler), ] ) if __name__ == "__main__": application = app() application.listen(8080) print("监听端口8080") tornado.ioloop.IOLoop.current().start()代码解析:
- ChatRoom: 这个类管理客户端。即添加、删除客户端和向同一房间中的所有客户端广播消息。
- WebSocketHandler: 我们修改了现有的处理程序以将新客户端添加到房间中，在收到新消息时在创建房间之前广播消息。
- ApplicationWithRooms: 我们用这个类替换了默认的Application，只是为了处理聊天室逻辑。
- /websocket/(\w+)/(\w+): 处理程序URL也已更改，以在URL中接收聊天室和用户ID。
客户端代码
客户端代码也已更新，以在连接后提示发送消息到网络。
const WebSocket = require("ws"); const readline = require("readline"); const rl = readline.createInterface({ input: process.stdin, output: process.stdout }) function generate_random_name() { let name = ""; for (let i = 0; i > `); rl.prompt(); rl.on("line", function (line) { if (line === "exit") { socket.close(); return; } socket.send(line); }) }) socket.on("message", function (message) { if (message.length > 0) { console.log('[' + '' + message + ']') } rl.prompt(); }) socket.on("close", function () { console.log("连接已关闭...") rl.close(); })好了，这就是最小代码实现，你可以上手运行一下，看看实际效果。
总结
虽然提供的示例展示了使用WebSocket的基本功能，但值得注意的是，生产级别的WebSocket应用程序需要更全面的方法。
必须仔细实现强大的授权、安全的身份验证机制和其他重要的考虑因素，以确保您的实时通信解决方案的安全性、可扩展性和可靠性。
始终根据最佳实践和安全标准调整WebSocket实现，以满足生产环境的多样化要求。
