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

Python实现WebSocket通信（从零开始构建实时Web应用）

主机测评网
Python
2025-12-08
948

在现代Web开发中，Python WebSocket通信已成为实现实时数据交互的关键技术。与传统的HTTP请求-响应模式不同，WebSocket允许服务器和客户端之间建立持久连接，实现双向、低延迟的数据传输。本文将手把手教你使用Python搭建一个简单的WebSocket服务器，并通过浏览器进行实时通信。

Python实现WebSocket通信（从零开始构建实时Web应用） Python WebSocket通信 WebSocket实时通信 Web开发 WebSocket服务器第1张

什么是WebSocket？

WebSocket是一种网络通信协议，于2011年成为IETF标准（RFC 6455）。它通过单个TCP连接提供全双工通信通道，特别适合需要WebSocket实时通信的场景，如聊天室、在线游戏、股票行情推送等。

准备工作

我们将使用Python的websockets库来实现WebSocket服务器。首先，请确保已安装Python（建议3.7+），然后通过pip安装所需依赖：

pip install websockets

创建WebSocket服务器

下面是一个最简单的WebSocket服务器代码。它会接收客户端消息，并原样返回（回显服务）：

import asyncioimport websockets# 定义处理客户端连接的协程函数async def echo(websocket, path):    print("新客户端已连接")    try:        async for message in websocket:            print(f"收到消息: {message}")            # 将消息原样返回给客户端            await websocket.send(message)    except websockets.exceptions.ConnectionClosedOK:        print("客户端正常断开连接")    except Exception as e:        print(f"连接异常: {e}")# 启动WebSocket服务器start_server = websockets.serve(echo, "localhost", 8765)print("WebSocket服务器启动在 ws://localhost:8765")asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()

将上述代码保存为server.py，然后在终端运行：

python server.py

创建HTML客户端

接下来，我们编写一个简单的HTML页面作为WebSocket客户端：

<!DOCTYPE html><html lang="zh-CN"><head>    <meta charset="UTF-8">    <title>WebSocket测试客户端</title></head><body>    <h2>Python WebSocket通信测试</h2>    <input type="text" id="messageInput" placeholder="输入消息..." />    <button onclick="sendMessage()">发送</button>    <div id="messages" style="margin-top: 20px; padding: 10px; border: 1px solid #ccc; height: 200px; overflow-y: auto;"></div>    <script>        // 创建WebSocket连接        const socket = new WebSocket('ws://localhost:8765');        socket.onopen = function(event) {            addMessage('✅ 已连接到服务器');        };        socket.onmessage = function(event) {            addMessage('📥 收到: ' + event.data);        };        socket.onclose = function(event) {            addMessage('❌ 连接已关闭');        };        function sendMessage() {            const input = document.getElementById('messageInput');            const message = input.value;            if (message.trim()) {                socket.send(message);                addMessage('📤 发送: ' + message);                input.value = '';            }        }        function addMessage(text) {            const messagesDiv = document.getElementById('messages');            const p = document.createElement('p');            p.textContent = text;            messagesDiv.appendChild(p);            messagesDiv.scrollTop = messagesDiv.scrollHeight;        }        // 回车发送        document.getElementById('messageInput').addEventListener('keypress', function(e) {            if (e.key === 'Enter') {                sendMessage();            }        });    </script></body></html>

将此代码保存为client.html，用浏览器打开即可与服务器通信。

进阶：构建多人聊天室

要实现多人聊天功能，我们需要维护所有连接的客户端列表，并向每个客户端广播消息。以下是改进后的服务器代码：

import asyncioimport websockets# 存储所有连接的客户端connected_clients = set()async def chat_handler(websocket, path):    # 新客户端加入    connected_clients.add(websocket)    try:        async for message in websocket:            # 向所有其他客户端广播消息            if connected_clients:                await asyncio.wait(                    [client.send(message) for client in connected_clients]                )    finally:        # 客户端断开时移除        connected_clients.remove(websocket)start_server = websockets.serve(chat_handler, "localhost", 8765)asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()