企业官网建设流程全解析

简介：基于WebSocket的网页聊天室项目源码，是一份适合前端、后端及全栈初学者研读的轻量级示例，围绕网页即时通信场景，展示如何通过一次HTTP升级请求建立持久连接，实现浏览器与服务器之间的双向实时消息收发。相比传统HTTP轮询，这种全双工通信方式显著降低延迟，提升多用户在线互动体验。压缩包共5个文件，整包仅3KB，类型涵盖Python后端脚本、HTML前端页面、TXT依赖清单、Markdown说明文档及gitignore配置，结构精炼，可直接用于学习与运行调试。目前已有62人学习下载。项目中后端脚本负责连接建立、维护与消息广播，前端页面通过WebSocket API发起连接并处理收发帧，可直观对照学习握手流程、文本帧交互及多客户端消息共享；配套说明文档与依赖清单则帮助快速搭建本地环境，适合作为课堂练习或实时通信技术入门素材。同时可结合跨站脚本注入过滤、加密传输等安全思路进行加固，或扩展在线用户列表、聊天记录持久化、消息格式校验等功能。

1. 为什么我用原生 WebSocket 而不是 socket.io 写聊天室

第一次看到基于 WebSocket 的网页聊天室代码时，我冒出来的念头是「原来实时聊天可以这么朴素」。服务端没有引入 socket.io，前端也只塞了一个 HTML 文件，但几十个用户同时在线、消息秒达、断线自动重连，用起来一点不比大框架差。WebSocket 的核心优势不在「快」，而在它把 HTTP 的请求-响应模式换成了一条全双工通道，服务器能主动推送，客户端也能随时说话，这才是聊天室这类应用该有的形态。

这套资源适合三类人：想真正搞懂 WebSocket 协议怎么落地的人、需要完整 Demo 做毕业设计的在校生、公司内部工具要加一个轻量实时通知面板的开发者。它用 Python 写服务端，页面是原生 HTML 加 JavaScript，几乎没有第三方依赖。读通之后，连接管理、广播、心跳、重连、安全这些 WebSocket 必修点就都过了一遍。

下面我按自己拆这套代码的顺序来写：先看服务端连接生命周期，再看前端心跳与消息协议，然后把我踩过的坑摊开讲，最后落到安全防护和部署验证。

2. 服务端连接管理：从握手到广播的完整链路

2.1 先看握手，别急着写业务代码

WebSocket 连接建立不是魔法，它本质上是一次带着Upgrade头的 HTTP 请求。客户端请求升级协议，服务端返回101 Switching Protocols，之后这个 TCP 连接就从 HTTP 切换到 WebSocket 帧协议，双方开始双向收发。Python 的websockets库把协议层的握手细节都封装好了，不需要手工计算Sec-WebSocket-Accept，但你必须知道：握手失败时，前端不会进onopen，而是直接触发onclose。本地调试时看到浏览器根本没连上，第一步要查服务端有没有正常监听端口，第二步查反向代理有没有放行Upgrade头。

为什么选 Python 的websockets而不是 Tornado 或 Flask-SocketIO？因为这个聊天室的业务就是消息广播加在线列表，单进程能撑几百连接，websockets基于 asyncio，代码量最小，理解起来不绕。框架越重，连接生命周期就越像黑匣子，排障反而困难。

pip install -r requirements.txt python server.py

这两条命令跑起来之后，服务端在0.0.0.0:8765监听。requirements.txt里锁的是websockets库，如果同时装了多个 asyncio 框架，注意版本别冲突。

2.2 连接注册表：为什么用 dict 而不是 list

服务端最核心的数据结构是连接注册表。我之前用 list 存过连接，后来发现清理逻辑很容易出问题：客户端断开后要按值移除，还要额外维护一个昵称映射，两步操作之间一旦抛异常，连接就漏清理了。改用 dict 之后一步到位，key 是 WebSocket 连接对象，value 是客户端昵称。

连接建立时，我们其实还不知道客户端叫什么。所以协议里要有一条register消息：客户端连上后先发昵称，服务端校验通过再把它放进注册表。这样广播系统消息时能同时拿到完整用户列表和在线人数。

import asyncio import json import time clients = {} # 连接注册表：websocket -> nickname async def chat_handler(ws): nickname = None try: # 客户端连上后必须先发一条 register 消息完成注册 while nickname is None: raw = await ws.recv() msg = json.loads(raw) if msg.get("type") != "register": continue nickname = sanitize(msg.get("name", "匿名用户")) clients[ws] = nickname # 广播系统消息，通知其他客户端有人上线 await broadcast({ "type": "system", "text": f"{nickname} 进入了聊天室", "online": len(clients), }) # 主循环：持续接收该客户端的聊天消息 async for raw in ws: msg = json.loads(raw) if msg.get("type") == "chat": await broadcast({ "type": "chat", "nickname": nickname, "text": msg.get("text", ""), "ts": int(time.time()), }) except websockets.exceptions.ConnectionClosed: pass finally: if nickname: clients.pop(ws, None) await broadcast({ "type": "system", "text": f"{nickname} 离开了聊天室", "online": len(clients), })

逻辑说明：注册循环会卡住，直到收到一条有效的register消息，这个设计防止了匿名连接直接进入广播池；加入注册表之后再广播上线通知，保证online字段是最新值；async for raw in ws是websockets库的惯用法，底层等价于while True + await ws.recv()。finally块是必须的，无论客户端正常关闭还是异常掉线，注册表都要清理，否则连接对象会一直留在内存里。

参数说明：昵称在入库前先做sanitize，这属于安全底线；每次广播都带online字段，前端拿这个值刷新在线人数，不需要再单独拉一个接口。这是 WebSocket 聊天室很常用的一种设计，消息即状态。

2.3 广播：用 gather 并发发送，但必须容忍坏连接

广播函数最简单的写法是for循环逐个await send，但这样有一个隐患：如果某个客户端已经断开，send会抛ConnectionClosed异常，后面的客户端全部被卡住。线上聊天室最怕这种情况——一个人关浏览器，整个房间的消息都发不出去。

我一般用asyncio.gather同时发送所有消息，同时设置return_exceptions=True，把单个连接的异常吞掉，不影响整轮广播。

async def broadcast(message): if not clients: return payload = json.dumps(message, ensure_ascii=False) # return_exceptions=True 保证个别客户端断开不拖垮整轮广播 await asyncio.gather( *(client.send(payload) for client in list(clients.keys())), return_exceptions=True )

逻辑说明：先统一序列化成 JSON，再并发发送给注册表里所有客户端。return_exceptions=True的意思很明确：某个连接发送失败就失败，后续由主循环的ConnectionClosed异常兜底清理。list(clients.keys())这一步是为了防止广播过程中注册表被其他协程并发修改。

参数说明：ensure_ascii=False让中文消息原样输出，前端排查 payload 时一眼能看懂。如果广播频率高到一秒钟几十条，建议把gather换成asyncio.Queue加消费者任务，避免大量协程堆积；这个规模的项目用不到，但到了两三百连接以上就要考虑这个优化了。

3. 前端实时通信：WebSocket API、消息协议与心跳重连

3.1 浏览器里的 WebSocket 对象到底做了什么

前端index.html里最核心的代码就是一个WebSocket对象。它跟fetch不一样的地方是：不受同源策略限制，任意域名都能发起连接，所以后端必须自己做来源校验。四个回调函数各有分工：onopen表示握手成功，onmessage收到消息帧，onclose连接关闭，onerror出错。注意onerror之后通常会紧跟着一个onclose，所以重连逻辑要统一放在onclose里，不要放在onerror里，否则一次断线会触发两次重连。

常见做法是包一层connect函数，断线时递归调用自己，用指数退避控制重连频率：

function connect() { const ws = new WebSocket(wsUrl); ws.onopen = () => { retry = 1; setStatus("在线"); }; ws.onclose = () => { setStatus("离线"); setTimeout(connect, 1000 * retry); retry = Math.min(retry * 2, 30); }; }

逻辑说明：onclose里用setTimeout重连，而不是在onerror里直接重连，是为了防止一次连接故障触发两条重连链路。retry从 1 秒开始翻倍，封顶 30 秒，服务器如果正在重启，前端高频重连只会加重负担，指数退避是标准解法。

3.2 自定义消息协议：用 JSON 语义化，别发裸字符串

聊天的消息协议我在服务端章节已经定了基调：所有消息都是 JSON。最忌讳的写法是前端直接ws.send("你好")，服务端拿到字符串还要猜这是昵称还是聊天内容。定义好消息类型，前后端才能收敛：

前端发送统一走JSON.stringify，收到消息统一JSON.parse，然后按type分发给不同的 UI 逻辑。WebSocket 的帧天然带边界，一个 JSON 对象一帧就能发完，不需要做类似 TCP 粘包的分包处理——浏览器底层会重组分片帧，这个不用操心。

ws.onmessage = (event) => { const msg = JSON.parse(event.data); if (msg.type === "chat") { appendChat(msg.nickname, msg.text); } else if (msg.type === "system") { appendSystem(msg.text); } else if (msg.type === "online") { updateOnline(msg.online); } };

这里online字段虽然挂在system消息里，前端拿到后单独更新在线人数，不需要请求额外接口。把状态变化塞进消息协议里，是 WebSocket 应用减少请求量的典型做法。

3.3 心跳与自动重连：让聊天室撑过凌晨的空闲期

聊到热词「websocket心跳机制实现」了。浏览器的 WebSocket API 不提供发送协议级 Ping 帧的能力，所以必须用应用层心跳：前端每隔一段时间发给服务端一条ping，服务端回一条pong。这对维持长连接非常关键——Nginx、云负载均衡器、防火墙都会回收空闲连接，判定标准就是一段时间内没有数据流动。没有心跳的长连接，往往是半夜静悄悄地被设备掐断，第二天用户一来发现页面掉线了。

const heartbeat = setInterval(() => { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: "ping", ts: Date.now() })); } }, 30000);

逻辑说明：30 秒发一次ping，服务端收到后回pong。前端如果 60 秒内既没发消息也没收到任何帧，说明连接已经半死，主动close()触发重连，比挂在一个僵尸连接上等用户发现要强得多。

参数说明：心跳间隔 30 秒、超时 60 秒是常用起步值。如果中间有内网代理设备，建议缩到 15 秒 / 30 秒，因为内网设备回收空闲连接更积极。注意服务端本身还有一套ping_timeout配置，两者要协调，不要让服务端先于前端判定超时。

4. 避坑指南：WebSocket 调试中的五个高频翻车点

4.1 连接一开就断，浏览器报 1006 异常关闭

现象：前端onopen都没触发，onclose立即执行，浏览器 console 里关闭码是 1006，含义是连接异常关闭，服务端没有发送标准关闭帧。

原因：本地直连服务端没问题，但一旦挂到 Nginx 后面就翻车，九成是反向代理没有放行 WebSocket 的Upgrade头。Nginx 默认按 HTTP 处理转发，握手请求到不了 Python 服务端。

解决：Nginx 的location配置里显式声明proxy_set_header Upgrade $http_upgrade和proxy_set_header Connection "upgrade"。具体配置在第 6 章有完整写法，这里先记住一句话：WebSocket 走反代，Upgrade和Connection两个头必须手动透传，Nginx 默认不带。

4.2 心跳发出去了，连接还是被断开

现象：前端日志里ping每 30 秒正常发送，onclose依然偶尔触发，间隔大概在一分钟到几分钟不等。

原因：服务端只处理了register和chat，没处理ping，或者处理了但没回pong。前端发出的心跳像石沉大海，链路仍然被判定为半死。更像黑匣子的是：被防火墙静默掐断后，前端还傻等，直到下一次send失败才触发onclose。

解决：服务端收到ping必须回pong，这个逻辑要写进主循环，不能漏。正确的处理是：

if msg.get("type") == "ping": await ws.send(json.dumps({"type": "pong", "ts": msg.get("ts")}))

把前端的时间戳原样带回来，前端还能顺便算一下 RTT。

4.3 Task was destroyed but it is pending：连接清理没兜底

现象：服务端控制台持续输出Task was destroyed but it is pending，过一会儿内存占用明显上涨，广播开始变慢。

原因：客户端断开后，注册表里的连接对象没被移除，async for循环还在等待一个永远不会到达的消息，任务的协程永远不会结束。

解决：这就是为什么第 2 章的finally块必须存在。except websockets.exceptions.ConnectionClosed负责吞掉正常异常，finally负责清除注册表并广播下线。两件事缺一不可，这是连接生命周期里最容易被漏掉的一段。

4.4 电脑上好的，手机连不上

现象：PC 浏览器访问正常，同一局域网里的手机打开页面，连接一直pending，最后超时失败。

原因：服务端websockets.serve绑定了127.0.0.1，只监听回环地址，局域网内的设备自然无法访问。

解决：监听地址改成0.0.0.0：

async with websockets.serve(chat_handler, "0.0.0.0", 8765): await asyncio.Future()

顺带检查一下防火墙有没有放行 8765 端口。很多发行版默认防火墙只放行了 22 和 80，自测时最容易忽略。

4.5 幽灵在线：断开后在线列表还挂着这个人

现象：用户直接关掉浏览器标签页，服务端finally执行了，注册表也清理了，但前端在线列表里这个人一直挂着，在线人数也不减。

原因：finally里只做了clients.pop()，没有广播下线通知，或者广播的时间点不对——先广播再 pop，会导致broadcast函数里len(clients)仍然包含这个人，数值多算一个。

解决：先pop再广播，顺序不能反。我在第 2 章给的代码就是这个顺序：clients.pop(ws, None)执行完，再调用broadcast生成带正确online人数的系统消息。这个细节看起来不起眼，但每个没做对的项目都会出现幽灵在线。

5. 聊天室安全：XSS 过滤、来源校验与 wss 部署

5.1 XSS 的两道防线：服务端过滤 + 前端 textContent

聊天室是 XSS 攻击的重灾区。攻击者发一条<img src=x onerror=alert(1)>，如果服务端不处理、前端用innerHTML渲染，每个在线用户都会执行这段脚本。第一道防线是服务端清洗输入：

def sanitize(text: str, max_len=200) -> str: text = text.strip() if not text: return "" if len(text) > max_len: text = text[:max_len] # 只做清洗，不做 HTML 转义，转义交给前端 textContent return text.replace("\r", " ").replace("\n", " ")

逻辑说明：截断长度、去掉控制字符，保证广播出去的消息不会携带恶意 HTML。这里特意不做html.escape，是因为前端统一用textContent渲染时本来就会转义特殊字符，服务端再转义一次，页面上反而会显示<script>这种双重编码乱码。

前端这一道防线同样关键：

const div = document.createElement("div"); div.textContent = `${msg.nickname}: ${msg.text}`; container.appendChild(div);

用textContent而不是innerHTML，浏览器会自动把消息内容里的 HTML 标签当纯文本显示，不需要额外过滤。如果以后要做表情图片这类富文本渲染，再考虑在服务端做白名单过滤，前端用innerHTML时绝不能直接拼用户输入。

5.2 Origin 校验和连接认证：别让任何人都能连进来

WebSocket 不受同源策略限制，任何网站里的 JavaScript 都可以向你的服务端发起连接。如果不校验来源，攻击者可以在自己页面上写脚本，往你服务器灌消息，或者偷听广播内容。服务端必须在握手阶段检查Origin头：

origin = ws.request_headers.get("Origin", "") if origin and origin != f"http://{allowed_host}": await ws.close(code=1008, reason="origin denied")

逻辑说明：1008在 WebSocket 协议里对应策略违规，浏览器会显示1008 Policy Violated。注意Origin头可能为空字符串，一些原生客户端不带这个头，所以条件里要多一个if origin，避免误杀。

更进一步的做法是阶段认证：客户端连接成功后，在register消息里带一个 token，服务端校验通过才把连接加入注册表。这个聊天室如果在公开网络部署，token 校验必须加；如果只在局域网内部使用，Origin校验基本够用。

5.3 从 ws 到 wss：线上部署的强制要求

浏览器有混合内容限制：https页面里连接ws://会被直接拦截。线上部署时，只要前端是通过https访问的，WebSocket 地址就必须是wss://。实现方式一般是 Nginx 终结 SSL，然后在内网把明文ws://转发给 Python 服务端，这个配置下一章会展开。

另外提醒一点：wss://依赖有效的 SSL 证书，别用自签名证书临时顶上，浏览器对 WebSocket 的证书校验和普通 HTTPS 一样严格，证书不合法直接拒绝连接。

6. 进阶实践：Nginx 反代、连接数监控与压测验证

6.1 Nginx 反代 WebSocket 的标准写法

线上部署建议用 Nginx 做 TLS 终结和反向代理，Python 服务端只在内网监听。关键的location配置如下：

location /ws/ { proxy_pass http://127.0.0.1:8765; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_read_timeout 3600s; }

proxy_http_version 1.1必须显式声明，Nginx 默认用 HTTP/1.0，而 WebSocket 握手依赖 HTTP/1.1 的Upgrade机制。proxy_read_timeout要和服务端心跳周期匹配，设成 3600 秒是为了避免 Nginx 侧先掐断连接。

6.2 复用消息协议的 /stats 监控入口

给聊天室加监控，最省事的做法不是单独开 HTTP 端口，而是复用 WebSocket 协议，像这样：

elif msg.get("type") == "stats": await ws.send(json.dumps({ "type": "stats", "online": len(clients), "names": list(clients.values()), }, ensure_ascii=False))

前端或运维脚本定期发一条{"type": "stats"}，就能拿到在线人数和用户列表，不需要额外暴露端口。这个小入口在排查幽灵连接时非常有用，一眼就能看出注册表里到底存了哪些连接。

6.3 用 100 个并发客户端做一次端到端压测

聊天室改完上线前，我习惯跑一遍并发压测。脚本不用复杂的工具，Python 的websockets库就够了：

import asyncio, json, time, websockets async def one_client(idx): async with websockets.connect("ws://127.0.0.1:8765") as ws: await ws.send(json.dumps({"type": "register", "name": f"test{idx}"})) await asyncio.sleep(0.2) # 等广播真正建立 start = time.time() await ws.send(json.dumps({"type": "chat", "text": "hi", "idx": idx})) while True: raw = await ws.recv() msg = json.loads(raw) if msg.get("idx") == idx: # 等到自己那条消息的回包 return time.time() - start async def main(): rtts = await asyncio.gather(*[one_client(i) for i in range(100)]) print(f"平均 RTT: {sum(rtts)/len(rtts)*1000:.1f} ms") asyncio.run(main())

压测时我给chat消息临时加了一个idx字段，用于识别回包。100 个客户端同时发消息，服务端广播给所有人，每个客户端持续recv，直到收到自己的编号为止，这个定向等待比固定sleep准得多。跑完如果平均 RTT 在 50ms 以内，说明单机聊天室扛这个并发量没问题。

从那以后我每次接 WebSocket 项目，无论多急，都会先在本地裸端口跑一遍 100 连接压测，再挂到 Nginx 后面验证Upgrade头，最后才把域名暴露到公网。这套固定动作帮我避开了大部分「线上才出现」的诡异问题。希望帮到你。

本文还有配套的精品资源，点击获取