Python WebSocket 实时推送服务器与 Web UI 实现指南

简介这是一套可自托管的消息推送服务器源码面向需要私有化实时通知能力的开发者与运维人员。它通过REST-API发送消息、WebSocket实时接收并支持管理用户、客户端与应用配套一个时尚的Web UI和Android客户端适合对数据隐私敏感、希望摆脱第三方推送服务的团队自建使用。资源包共226个文件约1.11MB以113个Go文件构成服务端核心逻辑29个ts与27个tsx文件实现前端界面另有png图标、md文档、json与yml配置、Dockerfile及arm64/armv7构建文件等覆盖从后端到部署的完整链路。目前已有221人学习下载。读者可从中获得一套结构清晰、可直接二次开发的消息服务实现理解REST与WebSocket双通道通信设计、多用户与客户端管理模型以及容器化多架构部署思路适合作为自托管推送系统的学习与改造基础。1. 一个 WebSocket 服务器该长什么样从「能连上」到「看得见」很多人第一次写 WebSocket 服务端卡住的地方不是协议本身而是「我到底该做成什么形态」。标题里说的「一个简单的服务器用于每个 WebSocket 实时发送和接收消息包括一个时尚的 Web UI」翻译成工程语言就是一个进程同时承担三件事——维护 WebSocket 连接、在连接之间转发消息、再挂一个浏览器页面让人肉眼看到消息在流动。它解决的是「websocket 实时推送数据」这条链路上最原始的验证需求后端到底有没有把消息推出去前端到底有没有收到中间有没有断。适合谁适合正在做「python django websocket 实现后台有数据前端推送」这类需求、但还没跑通最小闭环的人也适合想给内部工具加一个实时面板、又不想引入 Kafka 或消息队列的团队。它不适合高并发生产网关但对调试、演示、内网小规模广播来说够用且透明。下面这套方案我用 Python 的websockets库加一个单页 HTML 来落地全程不依赖框架方便你看清每一层在干什么。2. 选型与协议底座为什么用原生 WebSocket 而不是轮询2.1 握手阶段到底发生了什么WebSocket 的本质是「借 HTTP 完成一次握手然后把这条 TCP 连接升级成双向通道」。客户端发一个带Upgrade: websocket和Sec-WebSocket-Key的 GET 请求服务端把 Key 拼上一个固定 GUID 做 SHA-1再 Base64回写Sec-WebSocket-Accept。这一步过了后面的帧就不再走 HTTP 语义而是走 WebSocket 自己的帧格式一个字节的操作码加掩码位后面跟长度和载荷。理解这一点很关键因为它决定了你排错的方向。如果握手失败浏览器控制台会直接报WebSocket connection to ... failed这时候去看 HTTP 状态码如果握手成功但收不到消息那问题在帧的收发逻辑或业务代码跟 HTTP 已经没关系了。很多人把这两类问题混在一起查白白浪费时间。2.2 为什么不用轮询和 SSE轮询是客户端每隔几秒发一次 HTTP 请求问「有没有新数据」。它的缺点是延迟等于轮询间隔且大量请求头开销。SSE 是服务端单向推浏览器支持不错但它只能服务端到客户端客户端要发消息还得另开 HTTP 接口。WebSocket 是唯一在浏览器里原生支持、又真正双向的方案。热搜里常出现的「websocket test client」工具本质也是在验证这条双向通道。选websockets库而不是aiohttp或FastAPI的 WebSocket是因为它足够薄。websockets只做协议层不掺路由和中间件你写出来的代码就是连接管理的骨架。等你摸清了骨架再迁到框架里就是换个壳的事。2.3 最小可运行服务端先装依赖然后写服务端。下面这段是核心我把它拆成「连接注册」和「广播」两块。pip install websocketsimport asyncio import json import websockets # 用一个集合保存所有活跃连接集合天然去重 CLIENTS set() async def handler(websocket): # 新连接进来先登记 CLIENTS.add(websocket) try: async for raw in websocket: # 收到的原始消息是字符串约定用 JSON 传 msg json.loads(raw) # 补一个服务端时间戳方便前端排序 msg[server_ts] asyncio.get_event_loop().time() payload json.dumps(msg, ensure_asciiFalse) # 广播给所有连接包括发送者自己 await broadcast(payload) except websockets.ConnectionClosed: pass finally: # 无论怎么退出都要摘掉否则集合会泄漏 CLIENTS.discard(websocket) async def broadcast(payload): if not CLIENTS: return # gather 并发发送return_exceptions 防止一个坏连接拖垮整轮 await asyncio.gather( *(c.send(payload) for c in CLIENTS), return_exceptionsTrue, ) async def main(): async with websockets.serve(handler, 0.0.0.0, 8765): await asyncio.Future() # 永久挂起保持服务运行 if __name__ __main__: asyncio.run(main())逻辑说明handler是每个连接独立的协程async for会在连接关闭时自然退出。CLIENTS用set而不是list是因为断开时discard是 O(1)且重复添加不会产生脏数据。broadcast里用gather加return_exceptionsTrue这是血泪经验——只要有一个客户端网络卡住不加这个参数整个广播会抛异常中断其他正常客户端也收不到消息。参数说明端口8765是社区常用的 WebSocket 测试端口换成别的也行但要和前端一致。0.0.0.0表示监听所有网卡本地调试用127.0.0.1更安全。ensure_asciiFalse是为了中文不被转义成\uXXXX前端直接可读。3. 把 Web UI 接上一个页面看清消息流向3.1 前端连接与重连逻辑「时尚的 Web UI」不需要框架一个 HTML 文件加几十行 JS 就够。关键是重连——WebSocket 断开是常态不写重连的页面活不过一次网络抖动。!DOCTYPE html html langzh head meta charsetutf-8 titleWebSocket 实时面板/title style body { font-family: system-ui, sans-serif; margin: 0; background: #0f1115; color: #e6e6e6; } #log { height: 60vh; overflow-y: auto; padding: 16px; font-family: ui-monospace, monospace; } .row { padding: 6px 10px; border-radius: 6px; margin-bottom: 6px; background: #1a1d24; } .ts { color: #6b7280; margin-right: 8px; } #bar { display: flex; gap: 8px; padding: 12px 16px; background: #15181f; } input { flex: 1; padding: 10px; border-radius: 6px; border: 1px solid #2a2f3a; background: #0f1115; color: #e6e6e6; } button { padding: 10px 18px; border: 0; border-radius: 6px; background: #3b82f6; color: #fff; cursor: pointer; } #status { padding: 4px 16px; font-size: 13px; color: #9ca3af; } /style /head body div idstatus连接中…/div div idlog/div div idbar input idtext placeholder输入消息回车发送 / button onclicksend()发送/button /div script const logEl document.getElementById(log); const statusEl document.getElementById(status); let ws null; let retry 0; function connect() { ws new WebSocket(ws://localhost:8765); ws.onopen () { retry 0; statusEl.textContent 已连接; }; ws.onmessage (e) { const msg JSON.parse(e.data); const row document.createElement(div); row.className row; // 用服务端时间戳做展示避免各端时钟不一致 row.innerHTML span classts${msg.server_ts.toFixed(2)}/span${msg.text}; logEl.appendChild(row); logEl.scrollTop logEl.scrollHeight; }; ws.onclose () { statusEl.textContent 已断开重连中…; // 指数退避最多等 10 秒避免疯狂重连打爆服务端 const delay Math.min(1000 * 2 ** retry, 10000); retry; setTimeout(connect, delay); }; } function send() { const input document.getElementById(text); if (!input.value || ws.readyState ! WebSocket.OPEN) return; ws.send(JSON.stringify({ text: input.value })); input.value ; } document.getElementById(text).addEventListener(keydown, (e) { if (e.key Enter) send(); }); connect(); /script /body /html逻辑说明onclose里做指数退避重连retry每次翻倍封顶 10 秒。这是「websocket 实时推送数据」在生产里必须有的兜底否则服务端重启一次所有页面就永久失联。onmessage里用server_ts而不是本地时间是因为多客户端场景下本地时钟可能差几秒展示会乱。参数说明ws://localhost:8765要和后端端口一致如果页面和后端不同源注意浏览器不会拦 WebSocket 的跨域但反向代理层可能拦部署时要放行Upgrade头。3.2 用 curl 和命令行验证通道UI 之前先用命令行确认服务端活着能省掉一半「到底是前端还是后端」的纠结。websocat或 Python 自带的客户端都行。# 用 Python 快速起一个测试客户端连上后发一条消息 python -c import asyncio, websockets async def t(): async with websockets.connect(ws://localhost:8765) as ws: await ws.send({\text\:\hello from cli\}) print(await ws.recv()) asyncio.run(t()) 逻辑说明这段脚本连上后发一条 JSON然后等一条回显。因为服务端是广播自己也会收到自己发的消息所以recv能立刻返回。如果这里卡住不返回说明广播逻辑没走到回去检查broadcast是否被调用。参数说明websockets.connect默认会做握手校验如果服务端返回的Sec-WebSocket-Accept不对会直接抛异常这正好帮你验证握手实现是否正确。4. 避坑与排查那些让连接「玄学断开」的细节4.1 现象本地全通一上服务器就连不上原因云服务器安全组或防火墙没放行 WebSocket 端口或者反向代理Nginx没配Upgrade头透传。WebSocket 握手是 HTTP 请求但升级后是长连接很多代理默认按短连接处理几十秒就掐断。解决安全组放行端口Nginx 里加proxy_set_header Upgrade $http_upgrade;和proxy_set_header Connection upgrade;并把proxy_read_timeout调大否则默认 60 秒无数据就断。4.2 现象消息偶尔丢客户端收不到广播原因broadcast里用了await c.send()串行发送某个客户端 TCP 缓冲区满时会阻塞后面的客户端全被拖住。或者发送时没捕获ConnectionClosed一个已断开的连接让整轮广播抛异常。解决用asyncio.gather并发发送并加return_exceptionsTrue发送前判断连接状态发送后如果异常就把该连接从CLIENTS里摘掉。4.3 现象中文变乱码或显示成\u4f60\u597d原因json.dumps默认ensure_asciiTrue把非 ASCII 字符全转义了。前端如果直接当纯文本渲染看到的就是转义串。解决json.dumps(msg, ensure_asciiFalse)同时确保 HTML 的meta charsetutf-8存在服务端响应头也带 UTF-8。4.4 现象页面刷新后旧消息全没了原因服务端只做实时转发没有历史缓存。这是设计取舍不是 bug。但演示时用户会以为丢了数据。解决在服务端加一个固定长度的collections.deque(maxlen100)存最近消息新连接建立时先推一遍历史。注意maxlen要设否则内存会随运行时间无限涨。4.5 现象连接数一多CPU 飙升原因每个连接一个协程本身没问题但如果在handler里做了同步阻塞操作比如读文件、调同步库会卡住整个事件循环。解决所有阻塞操作丢进asyncio.to_thread或线程池用pprof之类的工具看热点时先确认瓶颈在协议层还是业务层。单机几千连接对websockets是正常量级飙升通常是业务代码写错了。5. 进阶把消息通道做成可观测、可扩展的形态跑通最小闭环之后真正决定这套东西能不能长期用的是两件事可观测和可扩展。先说可观测。我在handler里加了一个连接计数和消息计数每 10 秒打一行日志这样一眼能看出是「没人连」还是「连了但不发消息」。import time STATS {conns: 0, msgs: 0} async def stats_reporter(): while True: await asyncio.sleep(10) print(f[stats] conns{STATS[conns]} msgs{STATS[msgs]}) # 在 main 里并发启动 async def main(): asyncio.create_task(stats_reporter()) async with websockets.serve(handler, 0.0.0.0, 8765): await asyncio.Future()逻辑说明stats_reporter是一个独立协程和serve并发跑。STATS用普通 dict 就行因为单线程事件循环里不会有竞态。连接数在handler的add和discard处增减消息数在广播前累加。参数说明10 秒是经验值太密刷屏太疏看不出抖动。生产里应该换成结构化日志或指标接口但调试阶段print最快。再说可扩展。单进程广播的上限是单机连接数和带宽。要横向扩常见做法是引入一个 Redis 的 pub/sub每个服务实例订阅同一个频道收到消息后广播给本地连接。这样多实例之间就同步了。但要注意Redis 只做转发不做持久化历史消息还得另存。另一个方向是给消息加类型字段前端按类型渲染不同样式这样同一个通道能承载聊天、通知、进度条多种用途不用为每种场景单开服务。最后说一个我自己的习惯任何 WebSocket 服务上线前我都会先用websocket test client类的工具手动连一次发一条、收一条、断开、重连四个动作走完才算过。这套动作能覆盖握手、广播、断线清理、重连四个最容易翻车的点。别嫌土它比任何自动化测试都先发现问题。希望帮到你。本文还有配套的精品资源点击获取