Python WebSocket 服务端通信实战：从原理到完整代码示例

周予安

发布时间：2026-10-13 | 2人浏览过

原创

本文深入解析 WebSocket 协议原理，对比传统 HTTP 轮询的局限性，展示其在双向实时通信中的优势。通过 Python 的 websockets 库与 JavaScript 客户端的完整示例，演示如何建立连接、身份验证及消息收发。代码涵盖服务端监听、客户端交互及异常处理，适用于需要低延迟通信的 Web 应用开发。

WebSocket 技术背景与原理

1、唠唠叨叨

近期重新梳理了 WebSocket 技术，发现相关细节已生疏。为此，我使用 JavaScript 编写了客户端，并采用 Python 实现了服务端，以此重温并巩固这一领域的核心知识。

WebSocket 是一种定义明确的网络协议，旨在实现客户端与服务端之间的全双工数据交换。尽管常与 HTTP 一同提及，但 WebSocket 并非基于 HTTP 协议，而是构建在 TCP 连接之上的一种独立实现机制。

在 WebSocket 普及之前，若客户端需获取服务端处理进度，通常需采用 Ajax 进行频繁轮询，即浏览器每隔数秒向服务器发起一次请求，这会给服务器带来较大负载。另一种方案是 Long Poll（长轮询），其机制类似保持通话：客户端建立连接后，若无消息返回，服务器将保持响应阻塞状态，直到有数据可发。

WebSocket 有效解决了上述 HTTP 模式的痛点。一旦服务器完成协议升级（从 HTTP 切换至 WebSocket），即可主动向客户端推送信息，消除了轮询带来的同步延迟。由于 WebSocket 仅需一次 HTTP 握手即可建立持久连接，服务端可在此期间持续与客户端通信直至连接关闭，从而避免了反复解析 HTTP 协议的开销，显著降低了资源消耗。

通信效果演示

2、先看一下效果吧

2.1、效果1（一个客户端连上服务的并发送消息）

2.2、效果2（另一个客户端连上服务的并发送消息）

2.3、效果3（服务的收到客户端的全部消息并返回消息）

2.4、效果4（一个客户端掉线并不影响其它socket连接）

2.5、效果5（列取全部连接客户端对象和当前发消息的客户端对象）

Python 服务端与 JS 客户端实现

3、核心代码

3.1、Python


#! -*- coding: utf-8 -*-
"""
Author: ZhenYuSha
Create Time: 2019-1-14
Info: Websocket 的使用示例
"""
import asyncio
import websockets

websocket_users = set()


# 检测客户端权限，用户名密码通过才能退出循环
async def check_user_permit(websocket):
 print("new websocket_users:", websocket)
 websocket_users.add(websocket)
 print("websocket_users list:", websocket_users)
 while True:
  recv_str = await websocket.recv()
  cred_dict = recv_str.split(":")
  if cred_dict[0] == "admin" and cred_dict[1] == "123456":
   response_str = "Congratulation, you have connect with server..."
   await websocket.send(response_str)
   print("Password is ok...")
   return True
  else:
   response_str = "Sorry, please input the username or password..."
   print("Password is wrong...")
   await websocket.send(response_str)


# 接收客户端消息并处理，这里只是简单把客户端发来的返回回去
async def recv_user_msg(websocket):
 while True:
  recv_text = await websocket.recv()
  print("recv_text:", websocket.pong, recv_text)
  response_text = f"Server return: {recv_text}"
  print("response_text:", response_text)
  await websocket.send(response_text)


# 服务器端主逻辑
async def run(websocket, path):
 while True:
  try:
   await check_user_permit(websocket)
   await recv_user_msg(websocket)
  except websockets.ConnectionClosed:
   print("ConnectionClosed...", path) # 链接断开
   print("websocket_users old:", websocket_users)
   websocket_users.remove(websocket)
   print("websocket_users new:", websocket_users)
   break
  except websockets.InvalidState:
   print("InvalidState...") # 无效状态
   break
  except Exception as e:
   print("Exception:", e)


if __name__ == '__main__':
 print("127.0.0.1:8181 websocket...")
 asyncio.get_event_loop().run_until_complete(websockets.serve(run, "127.0.0.1", 8181))
 asyncio.get_event_loop().run_forever()

3.2、Html（JS）


 <script>
  var socket;
  if ("WebSocket" in window) {
   var ws = new WebSocket("ws://127.0.0.1:8181/test");
   socket = ws;
   ws.onopen = function() {
    console.log('连接成功');
    alert("连接成功, 请输入账号和密码");
   };
   ws.onmessage = function(evt) {
    var received_msg = evt.data;
    document.getElementById("showMes").value+=received_msg+"\n";
   };
   ws.onclose = function() {
    alert("断开了连接");
   };
  } else {
   alert("浏览器不支持WebSocket");
  }
  function sendMeg(){
   var message=document.getElementById("name").value+":"+document.getElementById("mes").value;
   document.getElementById("showMes").value+=message+"\n\n";
   socket.send(message);
  }
 </script>