Day 085 — WebSocket 通信练习清单

✅ 今日学习目标

[ ] 理解 WebSocket 协议的工作原理
[ ] 掌握 websockets 库的基本用法
[ ] 学会并发连接管理与消息广播
[ ] 了解心跳检测与断线重连机制
[ ] 能够构建一个基础聊天室

📝 基础练习

练习 1：回显服务器

目标：编写一个最简单的 WebSocket 回显服务器

要求： 1. 启动一个 WebSocket 服务器监听 localhost:8770 2. 客户端发送任何消息，服务器原样返回 3. 在服务端打印客户端地址和收到的消息

验证：用浏览器开发者工具或 Python 客户端连接测试

提示：

python

async def echo_handler(websocket):
    async for message in websocket:
        await websocket.send(message)

练习 2：计数器服务器

目标：服务器维护一个计数器，每次收到消息时递增

要求： 1. 服务器维护一个全局计数器 2. 每次收到消息，计数器 +1 3. 返回格式：{"count": N, "message": "你发送的是: xxx"} 4. 支持 /reset 命令重置计数器

提示：用 JSON 格式通信

练习 3：广播服务器

目标：实现消息广播功能

要求： 1. 支持多个客户端同时连接 2. 任何一个客户端发送的消息，广播给其他所有客户端 3. 客户端收到的消息带上发送者标识 4. 用 set 存储连接，广播时用 .copy()

🔧 进阶练习

练习 4：私聊系统

目标：实现一对一私聊

要求： 1. 客户端连接时注册用户名 2. 支持 /msg <用户名> <消息> 命令发送私聊 3. 私聊消息只发给指定用户 4. 发送方收到确认：[私聊] 消息已发送给 xxx

提示：用字典 username -> websocket 映射

练习 5：心跳检测

目标：实现自定义心跳机制

要求： 1. 服务端每 15 秒发送 {"type": "ping"} 消息 2. 客户端收到后回复 {"type": "pong"} 3. 如果 30 秒内没有收到 pong，服务端主动断开连接 4. 客户端断开后打印日志

不使用 websockets 内置的 ping/pong，手动实现

练习 6：消息历史

目标：记录并查询聊天历史

要求： 1. 服务器保存最近 50 条消息 2. 新用户加入时发送最近 10 条历史 3. 支持 /history 命令查看最近 20 条 4. 支持 /search <关键词> 搜索历史消息

🚀 实战练习

练习 7：在线状态广播

目标：实时显示用户在线/离线状态

要求： 1. 用户加入时广播："xxx 上线了" 2. 用户离开时广播："xxx 下线了" 3. 支持 /who 命令查看当前在线用户列表 4. 在线人数实时更新

练习 8：简单聊天机器人

目标：在聊天室中加入一个自动回复机器人

要求： 1. 创建一个名为 Bot 的特殊用户 2. 当有人 @Bot 或消息包含特定关键词时自动回复 3. 支持以下触发词： - "你好"/"hi"/"hello" → 回复问候语 - "时间" → 回复当前时间 - "天气" → 回复预设的天气信息 - "笑话" → 随机回复一个笑话 4. 机器人消息带上 [Bot] 标识

练习 9：文件传输

目标：通过 WebSocket 发送文件

要求： 1. 客户端选择本地文件 2. 将文件内容 base64 编码后发送 3. 服务端接收并保存到指定目录 4. 支持发送进度通知："正在发送... 50%" 5. 发送完成后通知所有用户

提示：使用 base64 模块处理二进制数据

💡 思考题

为什么广播时要用 .copy()？ 如果不用会出什么问题？写代码验证。
WebSocket 连接数有上限吗？ 如何测试一台服务器最多能支持多少并发连接？
如何实现"正在输入..."功能？ 当用户正在打字时，其他人能看到提示。
消息顺序保证：在 asyncio 单线程模型中，消息发送顺序是否一定等于接收顺序？
安全考虑：WebSocket 连接需要哪些安全措施？（认证、加密、限流等）

📚 扩展阅读

用 websockets 实现简易 RPC 框架
WebSocket 与 Redis Pub/Sub 结合实现跨进程消息
用 aiohttp 同时提供 HTTP 和 WebSocket 服务
WebSocket 连接池管理与负载均衡

Day 085 — WebSocket 通信 练习清单