fastapi sse websocket 奶茶店实时订单看板

fastapi sse websocket 奶茶店实时订单看板

用 FastAPI 从零搭一个「奶茶店实时订单看板」，把 SSE（单向推送） 和 WebSocket（双向通道） 两个实时通信技术放在同一个项目里对比学习。

这份教只要跟着章节顺序读、按步骤敲代码，就能独立写完所有文件并让项目跑起来。每一章都包含：

📖 本合订本目录

🎯 学完你能做到什么

• 理解 HTTP 长连接 的两种实现方式（SSE / WebSocket）及其差异；
• 用 FastAPI 写出一个后台广播任务（心跳源），每 0.8 秒推进一次订单状态；
• 用 StreamingResponse 实现 SSE，让浏览器「只读」地接收实时订单；
• 用 WebSocket 实现双向通道，让前端发命令（ping / 加单 / 取餐 / 快照）并拿到回执；
• 写一个纯前端页面（EventSource + WebSocket），把实时数据渲染成卡片看板；
• 独立完成项目骨架 → 数据层 → 传输层 → 入口 → 前端 → 运行全流程。

🛠 技术栈

📁 最终目录结构

fastapi-sse-ws-奶茶看板/
├── pyproject.toml # 依赖与项目元信息
├── .python-version # 固定 Python 版本 = 3.14
├── .gitignore
├── docs/ # 教程目录（分章原文件 + 本合订本）
│ ├── README.md
│ ├── 00-项目概述.md ~ 09-SSE-vs-WebSocket.md
│ ├── 教程合订本.md # ← 本文件
│ ├── dashboard.png # 运行截图（SSE 看板）
│ └── dashboard-ws.png # 运行截图（WS 操作台 + 通信日志）
├── src/ # 后端 Python 包
│ ├── __init__.py
│ ├── app.py # FastAPI 应用入口
│ ├── broadcaster.py # 后台广播任务（心跳源）
│ ├── models.py # 数据层：菜单 / 订单 / 推进逻辑
│ ├── sse.py # SSE 路由
│ └── ws.py # WebSocket 路由
└── static/
└── index.html # 前端看板页面

🚀 30 秒速览

# 1. 安装依赖（用 uv，更快；没有 uv 用 pip 也行，见第 1 章）
uv sync # 或: pip install fastapi "uvicorn[standard]" websockets

# 2. 启动服务
uv run python -m uvicorn src.app:app –host 127.0.0.1 –port 8000 –reload

# 3. 打开浏览器
# http://127.0.0.1:8000/

启动成功后你会看到这样的看板（左 SSE 单向推送、右 WebSocket 双向通道）：

点一下右下角的操作台按钮，还能看到 WebSocket 的双向通信日志：

第 0 章 · 项目概述

本章目标

在看任何代码之前，先用一张图、一张截图搞清楚三件事：

建立全局观之后，后面每一章的代码你就知道它「安在哪一层」。

0.1 项目做了什么

模拟一家奶茶店的实时订单看板：

• 后台每 0.8 秒 自动推进一次所有订单的制作进度（待制作 → 制作中 → 已完成 → 已取餐），偶尔随机生成新订单；
• 左侧看板用 SSE（单向推送）：服务器只管往下推最新订单列表，浏览器只管收和渲染；
• 右侧操作台用 WebSocket（双向通道）：店员可以主动发命令——ping（探活）、add（加单）、pickup（取餐）、snapshot（拉快照）——服务器对每条命令给明确回执。

设计意图：把两种实时技术放在同一个数据源上对比，直观体会「单向」和「双向」的差异。

0.2 整体架构图

0.3 三个角色

0.4 关键概念：SSE vs WebSocket（先记住一句话）

• SSE：服务器→浏览器的单向水管。基于 HTTP，浏览器开 EventSource 就连上，服务器用 text/event-stream 一帧一帧地推文本。断线浏览器自动重连。
• WebSocket：全双工。握手后升级成 ws://，双方都能随时发消息。适合有交互的场景（发命令、拿回执）。

本项目里：

• 「看订单」用 SSE——只需服务器推、浏览器收，最简单；
• 「店员操作」用 WS——店员要发命令，必须双向。

0.5 数据流动（一次 tick）

broadcaster 每 0.8s 触发一次:
1. advance() ← models.py 推进所有订单，返回快照 dict
2. json.dumps(snapshot) ← 序列化成字符串 payload
3. 对每个 sse_clients 里的 Queue:
queue.put({"event":"orders", "data": payload})
→ sse.py 的 event_stream() 从 Queue 取出，拼成 SSE 帧 yield 出去
4. 对每个 ws_clients 里的 WebSocket:
ws.send_text(payload) ← 直接发，失败则从集合移除

• SSE 端：每个连接有自己的 Queue，广播器只往 Queue 里塞，不直接发——保证发送不阻塞广播器。
• WS 端：直接 send_text，因为 WS 原生支持异步发送且能感知连接断开。

0.6 运行效果

启动服务、打开 http://127.0.0.1:8000/ 后，你会看到：

• 顶部：标题 + SSE/WS 连接状态灯 + 实时时钟；
• 统计条：总订单 / 制作中 / 已完成 / 已取餐；
• 左侧 📋 实时订单（标签「SSE 单向推送」）：每 0.8s 刷新的订单卡片，带进度条和状态色块；
• 右侧 🎛️ 店员操作台（标签「WebSocket 双向通道」）：同样的订单卡片 + 命令输入框 + ping/add/snapshot 按钮 + 黑底通信日志。

在右侧操作台点几下按钮，会看到 WS 通信日志里出现 → 发出的命令 和 ← 收到的回执，订单也会实时变化：

0.7 阅读建议

• 如果你是初学者：严格按章节顺序，每章先抄代码、再看讲解、再验证。
• 如果你已经会 FastAPI：可以直接跳到第 3、4 章看 SSE/WS 两个传输模块的实现差异，再看第 5 章广播器怎么把它们串起来。
• 第 9 章是对比总结 + 扩展练习，全部写完后回头看，会有更深理解。

下一章 → 01-环境与项目骨架.md：装环境、建目录、写配置文件。

第 1 章 · 环境与项目骨架

本章目标

把开发环境装好、把项目目录和配置文件建好。这一章不写业务逻辑，但它是后面所有代码的「地基」。

学完本章你会有这些文件（全是配置/空壳）：

fastapi-sse-ws-奶茶看板/
├── pyproject.toml
├── .python-version
├── .gitignore
├── src/
│ └── __init__.py
└── static/ # 空目录，第 7 章放 index.html

1.1 准备 Python 环境

本项目用 Python 3.14。先确认你装了 Python 3.14：

python –version
# 应输出类似: Python 3.14.x

如果没装，去 python.org 下载安装。

1.2 安装 uv（推荐）

uv 是一个超快的 Python 包管理器，比 pip 快很多。装一次即可：

# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell)
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

# 或用 pip 装也行
pip install uv

验证：

uv –version
# 应输出: uv 0.11.x …

没 uv 也能学：后面所有 uv xxx 命令都给出对应的 pip 版本。用 uv 只是更快、更省心。

1.3 创建项目目录

mkdir fastapi-sse-ws-奶茶看板
cd fastapi-sse-ws-奶茶看板

1.4 .python-version —— 固定 Python 版本

原始完整代码

新建文件 .python-version，内容只有一行：

3.14

讲解

• 这个文件被 uv、pyenv 等工具识别，表示「这个项目用 Python 3.14」；
• 团队协作时，大家 uv sync 就会自动装对应版本，不用口头约定；
• 内容就是 3.14，不带小版本号，表示「3.14.x 都行」。

1.5 pyproject.toml —— 项目元信息 + 依赖

原始完整代码

新建文件 pyproject.toml：

[project]
name = "fastapi-sse-ws"
version = "0.1.0"
description = "Add your description here"
requires-python = ">=3.14"
dependencies = [
"fastapi>=0.141.1",
"uvicorn[standard]>=0.52.1",
"websockets>=17.0.1",
]

讲解

• [project]：现代 Python 项目的标准元信息段（PEP 621）；
• name / version：项目名和版本，uv / pip 都会用它生成包信息；
• requires-python = ">=3.14"：本项目用到了 3.14 的特性（如 datetime.UTC 别名），所以要求 3.14+；
• dependencies 三个依赖：
  • fastapi：Web 框架，提供路由、StreamingResponse、WebSocket 等；
  • uvicorn[standard]：ASGI 服务器。[standard] 表示装「带额外性能优化」的版本（含 uvloop、httptools 等）；
  • websockets：WebSocket 协议库，Uvicorn 处理 WS 时需要它。

为什么 FastAPI 不自动带 websockets？因为很多项目用不到 WS。uvicorn[standard] 已经把 websockets 作为可选依赖装上了，这里显式声明是教学清晰起见。

1.6 安装依赖

# 用 uv（推荐）：会根据 pyproject.toml + .python-version 自动建虚拟环境并装依赖
uv sync

# 也可以直接用 pip（需要自己先 python -m venv .venv 并激活）
pip install fastapi "uvicorn[standard]" websockets

uv sync 执行完后，目录里会多出 .venv/（虚拟环境）和 uv.lock（锁文件，记录精确版本）。

1.7 .gitignore —— 不提交的东西

原始完整代码

新建文件 .gitignore：

# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

讲解

• __pycache__/、*.py[oc]：Python 运行时生成的字节码缓存，没必要进版本库；
• build/ dist/ wheels/ *.egg-info：打包产物；
• .venv：虚拟环境目录，每个人本地一份，不进版本库。

1.8 src/__init__.py —— 把 src 变成包

原始完整代码

新建目录 src/，里面建空文件 __init__.py：

"""src 包: 教学模块集合."""

讲解

• 有 __init__.py 的目录，Python 才把它当成一个包，里面的模块才能用 from src.app import app 这种方式导入；
• 这一行 docstring 是包的说明，运行 help(src) 时会显示；
• 后面我们要写的 app.py models.py sse.py ws.py broadcaster.py 都放在 src/ 下，它们之间用 from .models import … 这种相对导入互相引用。

1.9 建 static/ 目录

mkdir static

暂时留空，第 7 章往里放 index.html。

static/ 目录在 app.py 里会被挂载成静态资源，浏览器访问 /static/xxx 就能拿到里面的文件。

1.10 现在的目录结构

fastapi-sse-ws-奶茶看板/
├── .gitignore
├── .python-version
├── pyproject.toml
├── uv.lock # uv sync 自动生成（用 pip 时没有）
├── .venv/ # uv sync 自动生成（用 pip 时需手动建）
├── src/
│ └── __init__.py
└── static/ # 空目录

1.11 动手验证

# 在项目根目录执行，确认依赖装好了
uv run python -c "import fastapi, uvicorn, websockets; print('OK')"
# 用 pip 的话: python -c "import fastapi, uvicorn, websockets; print('OK')"

# 期望输出:
# OK

如果输出 OK，说明地基打好了。下一章我们往 src/ 里写第一个业务文件：数据层 models.py。

下一章 → 02-数据层-models.md：菜单、订单类型、状态推进逻辑。

第 2 章 · 数据层 models.py

本章目标

写项目的「大脑」——数据层。这一层负责：

• 定义菜单、口味选项；
• 定义订单（Order）和快照（Snapshot）的类型；
• 维护所有订单的内存状态；
• 提供 advance()：每被调用一次，就把所有订单往前推进一步，并随机生成新订单。

关键设计：数据层完全不知道 SSE / WebSocket 的存在。它只管数据，谁要推数据谁自己去拿。这样后面讲两种传输协议时，数据层一行都不用改。

2.1 原始完整代码

新建文件 src/models.py，完整照抄：

"""
数据层: 菜单、订单状态、推进逻辑.

与传输层(SSE / WebSocket)解耦, 方便单独讲解状态管理.
"""

import random
from datetime import UTC, datetime
from itertools import count
from typing import TypedDict

# ———- 菜单与口味 ———-

class MenuItem(TypedDict):
name: str
base: str
price: int

MENU: list[MenuItem] = [
{"name": "珍珠奶茶", "base": "红茶", "price": 12},
{"name": "杨枝甘露", "base": "芒果", "price": 18},
{"name": "芋泥啵啵", "base": "乌龙", "price": 15},
{"name": "柠檬养乐多", "base": "绿茶", "price": 13},
{"name": "草莓奶盖", "base": "茉莉", "price": 16},
{"name": "薄荷冰可可", "base": "可可", "price": 14},
]
SUGAR = ["全糖", "七分", "半糖", "三分", "无糖"]
ICE = ["正常冰", "少冰", "去冰", "热饮"]
TOPPING = ["珍珠", "椰果", "布丁", "芋圆", "奶盖", "不加料"]

STATUSES = ["待制作", "制作中", "已完成", "已取餐"]

# ———- 类型定义 ———-

class Order(TypedDict, total=False):
"""订单结构. total=False 因为 ready_since 仅在制作完成后才存在."""
id: int
drink: str
base: str
price: int
sugar: str
ice: str
topping: str
status: str
progress: int
created_at: str
ready_since: int

class Snapshot(TypedDict):
time: str
orders: list[Order]

# ———- 共享状态 ———-

order_seq = count(1) # 订单号自增器
orders: dict[int, Order] = {} # order_id -> order 详情

def now_iso() –> str:
"""ISO8601 时间戳 (Python 3.14 推荐用 UTC 别名, 避免废弃的 utcnow())."""
return datetime.now(UTC).isoformat()

def new_order() –> Order:
"""随机生成一份新订单."""
m = random.choice(MENU)
return Order(
id=next(order_seq),
drink=m["name"],
base=m["base"],
price=m["price"],
sugar=random.choice(SUGAR),
ice=random.choice(ICE),
topping=random.choice(TOPPING),
status="待制作",
progress=0,
created_at=now_iso(),
)

def advance() –> Snapshot:
"""推进所有未完成订单, 返回最新快照 (按 id 倒序)."""
# 1) 偶尔(20% 概率)生成新订单
if random.random() < 0.2 or not orders:
o = new_order()
orders[o["id"]] = o

# 2) 推进进度
completed: list[int] = []
for oid, o in orders.items():
if o["status"] == "已取餐":
continue
if o["status"] == "待制作":
o["status"] = "制作中"
if o["status"] == "制作中":
o["progress"] = min(100, o["progress"] + random.randint(8, 18))
if o["progress"] >= 100:
o["status"] = "已完成"
# 已完成 3 个 tick 后自动出队
if o["status"] == "已完成" and o["progress"] >= 100:
o["ready_since"] = o.get("ready_since", 0) + 1
if o["ready_since"] >= 3:
completed.append(oid)

for oid in completed:
orders[oid]["status"] = "已取餐"

# 3) 队列最多保留 8 条已取餐
finished = [oid for oid, o in orders.items() if o["status"] == "已取餐"]
for oid in finished[:–8] if len(finished) > 8 else []:
orders.pop(oid, None)

return Snapshot(time=now_iso(), orders=sorted(orders.values(), key=lambda o: –o["id"]))

def snapshot() –> Snapshot:
"""返回当前快照, 不改变状态."""
return Snapshot(time=now_iso(), orders=sorted(orders.values(), key=lambda o: –o["id"]))

⚠️ 抄完先不要运行——advance() 里用到 orders，但目前没人调用它。下一章装好路由、第 5 章装好广播器才会跑起来。

2.2 逐段讲解

2.2.1 导入与背景

import random
from datetime import UTC, datetime
from itertools import count
from typing import TypedDict

• random：生成新订单、随机进度增量；
• datetime.UTC、datetime：生成时间戳。注意 UTC：Python 3.12+ 才有 datetime.UTC 常量（3.14 推荐用），老的 datetime.utcnow() 已废弃（时区不明确）；
• itertools.count：一个无限自增计数器，用来生成订单号 1, 2, 3, …；
• typing.TypedDict：声明「带固定字段的 dict 类型」。

2.2.2 菜单与口味常量

class MenuItem(TypedDict):
name: str
base: str
price: int

MENU: list[MenuItem] = [ ... 6 款饮品 ... ]
SUGAR = ["全糖", "七分", "半糖", "三分", "无糖"]
ICE = ["正常冰", "少冰", "去冰", "热饮"]
TOPPING = ["珍珠", "椰果", "布丁", "芋圆", "奶盖", "不加料"]

STATUSES = ["待制作", "制作中", "已完成", "已取餐"]

• MenuItem 用 TypedDict 定义了菜单项的三个字段（名字、茶底、价格）。TypedDict 的好处是：写 m["name"] 时 IDE 有补全、有类型检查；
• MENU 是 6 款奶茶，每款有固定的茶底和价格；
• SUGAR / ICE / TOPPING：随机选口味用；
• STATUSES：订单的四种状态，前端用来渲染状态色块。

2.2.3 Order 与 Snapshot 类型

class Order(TypedDict, total=False):
"""订单结构. total=False 因为 ready_since 仅在制作完成后才存在."""
id: int
drink: str
base: str
price: int
sugar: str
ice: str
topping: str
status: str
progress: int
created_at: str
ready_since: int

• total=False 是关键：默认 TypedDict 要求所有字段都必须存在。但订单里的 ready_since（已完成了几个 tick）只在订单变成「已完成」之后才出现，新建订单时没有。所以用 total=False 表示「所有字段都是可选的」；
• 这样设计既保留了类型提示，又允许字段渐进式添加。

class Snapshot(TypedDict):
time: str
orders: list[Order]

• Snapshot 是每次广播要推给前端的完整数据：当前时间 + 所有订单列表。

2.2.4 共享状态

order_seq = count(1) # 订单号自增器
orders: dict[int, Order] = {} # order_id -> order 详情

• order_seq = count(1)：count(1) 返回一个生成器，每次 next(order_seq) 得到 1, 2, 3, …。比手动维护 id += 1 更优雅，线程安全（虽然本项目单线程）；
• orders：所有订单的字典，key 是订单号，value 是 Order。这是整个应用的「内存数据库」。

这两个是模块级变量，整个进程共享。所有 advance()、new_order()、snapshot() 都在操作它们。这也是为什么数据层要和传输层解耦——状态集中管理，谁要用谁拿。

2.2.5 时间戳函数

def now_iso() –> str:
"""ISO8601 时间戳 (Python 3.14 推荐用 UTC 别名, 避免废弃的 utcnow())."""
return datetime.now(UTC).isoformat()

• 返回形如 "2026-08-04T12:34:56.789+00:00" 的字符串；
• 前端用 new Date(iso) 解析后取时分秒显示在时钟上；
• 用 UTC 而非 utcnow()：前者返回带时区的 datetime，后者返回无时区的（已废弃）。

2.2.6 生成新订单

def new_order() –> Order:
"""随机生成一份新订单."""
m = random.choice(MENU)
return Order(
id=next(order_seq),
drink=m["name"],
base=m["base"],
price=m["price"],
sugar=random.choice(SUGAR),
ice=random.choice(ICE),
topping=random.choice(TOPPING),
status="待制作",
progress=0,
created_at=now_iso(),
)

• 随机选一款菜单、随机选糖度/冰度/加料；
• id 用 next(order_seq) 自增；
• 新订单初始状态 "待制作"、进度 0；
• Order(…) 看起来像构造函数，其实 TypedDict 只是类型提示，运行时它就是普通 dict，Order(…) 等价于 {"id": …, "drink": …, …}。

2.2.7 核心：advance() 推进逻辑

这是数据层最复杂的函数，逐段拆：

def advance() –> Snapshot:
"""推进所有未完成订单, 返回最新快照 (按 id 倒序)."""
# 1) 偶尔(20% 概率)生成新订单
if random.random() < 0.2 or not orders:
o = new_order()
orders[o["id"]] = o

• random.random() 返回 [0, 1) 的浮点数，小于 0.2 的概率约 20%；
• or not orders：如果当前一个订单都没有（刚启动），强制生成一个，避免看板空白；
• 生成后存进 orders 字典。

# 2) 推进进度
completed: list[int] = []
for oid, o in orders.items():
if o["status"] == "已取餐":
continue # 已取餐的不再推进
if o["status"] == "待制作":
o["status"] = "制作中" # 待制作 → 制作中（一次性）
if o["status"] == "制作中":
o["progress"] = min(100, o["progress"] + random.randint(8, 18))
if o["progress"] >= 100:
o["status"] = "已完成"
# 已完成 3 个 tick 后自动出队
if o["status"] == "已完成" and o["progress"] >= 100:
o["ready_since"] = o.get("ready_since", 0) + 1
if o["ready_since"] >= 3:
completed.append(oid)

状态机：

待制作 ──(首次advance)──> 制作中 ──(progress每tick +8~18)──> 已完成 ──(再3 tick)──> 已取餐

• 待制作 → 制作中：进入 advance 的第一个 tick 就转，只转一次；
• 制作中：每 tick 进度随机加 8~18，封顶 100。到 100 就 已完成；
• 已完成：用一个 ready_since 计数器，记录已完成几个 tick。满 3 个 tick 后标记为「已取餐」（放进 completed 列表，循环外统一处理）；
• o.get("ready_since", 0)：因为 ready_since 是渐进字段（新建时没有），用 .get 带默认值，避免 KeyError。

for oid in completed:
orders[oid]["status"] = "已取餐"

# 3) 队列最多保留 8 条已取餐
finished = [oid for oid, o in orders.items() if o["status"] == "已取餐"]
for oid in finished[:–8] if len(finished) > 8 else []:
orders.pop(oid, None)

• 先把 completed 列表里的订单正式改成「已取餐」；
• 然后做清理：已取餐的订单如果超过 8 条，把最老的几条从字典里删掉，防止内存无限增长；
• finished[:-8]：切片，保留最后 8 条（最新的），前面的全部 pop 掉；
• if len(finished) > 8 else []：不足 8 条时切片结果是全部，会误删，所以加这个判断；
• orders.pop(oid, None)：删除 key，不存在也不报错。

return Snapshot(time=now_iso(), orders=sorted(orders.values(), key=lambda o: –o["id"]))

• 返回当前快照：时间 + 所有订单（按 id 倒序，最新的在最前面，方便前端展示）；
• key=lambda o: -o["id"]：用负号实现倒序，等价于 reverse=True。

2.2.8 snapshot() —— 不改变状态的查询

def snapshot() –> Snapshot:
"""返回当前快照, 不改变状态."""
return Snapshot(time=now_iso(), orders=sorted(orders.values(), key=lambda o: –o["id"]))

• 和 advance() 的返回值一样，但不推进进度、不生成订单；
• 给 WebSocket 的 snapshot 命令用——前端想随时拉一次当前状态，不应该改变它。

2.3 状态机一览

• 每次调用 advance() = 一个 tick = 0.8 秒（由第 5 章广播器控制）；
• 一个订单从生成到「已取餐」大约需要：1（转制作中）+ ~7（进度到100）+ 3（等取餐）≈ 11 个 tick ≈ 9 秒。

2.4 为什么这一章没有「运行验证」？

数据层没有入口函数，单独跑看不到效果。但它会在后续章节被 broadcaster.py（每 0.8s 调 advance()）和 ws.py（响应 add/snapshot/pickup 命令）使用。

如果你想现在就验证它逻辑对不对，可以临时写个测试：

uv run python -c "
from src.models import advance, orders
import time
for _ in range(15):
snap = advance()
print(f'共 {len(snap[\\"orders\\"])} 单:', [(o['id'], o['status'], o['progress']) for o in snap['orders'][:3]])
time.sleep(0.3)
"

你会看到订单从「待制作」逐渐推进到「已取餐」，老的已取餐订单会消失。

下一章 → 03-SSE模块-sse.md：用 StreamingResponse 实现 SSE 单向推送。

第 3 章 · SSE 模块 sse.py

本章目标

实现项目第一个传输模块——SSE（Server-Sent Events，服务器单向推送）。学完本章你将理解：

• SSE 是什么、为什么它「够用又简单」；
• 用 FastAPI 的 StreamingResponse 怎么实现一个 SSE 端点；
• 每个连接独享一个 asyncio.Queue，广播器只负责往 Queue 里塞事件，发送不阻塞；
• SSE 帧的文本格式（event: / data: / 空行）；
• 三个关键响应头的作用。

3.1 先理解 SSE 是什么

SSE（Server-Sent Events）是 HTML5 标准里基于 HTTP 的服务器推送技术：

• 服务器响应头 Content-Type: text/event-stream，告诉浏览器「这是一条事件流」；

• 浏览器用 new EventSource(url) 建立连接，自动重连（断了浏览器会自动重连）；

• 服务器把数据按这个格式一段段发：

  event: 事件名
  data: 数据(一行或多行)

  每段之间用空行分隔。浏览器收到后触发 es.addEventListener("事件名", …)。

• 单向：服务器 → 浏览器。浏览器不能通过同一条连接发数据回去。

适合：股票行情、通知、日志流、我们的「订单看板」——服务器推、浏览器只看。

3.2 原始完整代码

新建文件 src/sse.py，完整照抄：

"""
教学模块 1: SSE 单向推送.

关键点
——
– 每个连接独占一个 asyncio.Queue, 后台广播任务把事件塞进每个 Queue
– StreamingResponse 的 media_type='text/event-stream' 告诉浏览器这是 SSE
– 关闭连接时 (Generator Exit / 异常) 自动从订阅集合移除, 防止内存泄漏
– 三条关键响应头:
Cache-Control: no-cache 不缓存
Connection: keep-alive 保持长连接
X-Accel-Buffering: no 关闭 nginx 缓冲, 实时推送
"""

import asyncio
import json
from typing import TypedDict

from fastapi import APIRouter
from fastapi.responses import StreamingResponse

from .models import MENU, STATUSES

class _QueueItem(TypedDict):
"""broadcaster 推入 Queue 的事件项."""
event: str
data: str

router = APIRouter()

# 所有 SSE 客户端的 Queue 集合 (由 broadcaster 写入)
sse_clients: set[asyncio.Queue[_QueueItem]] = set()

@router.get("/sse/orders")
async def sse_orders() –> StreamingResponse:
"""建立 SSE 连接, 持续接收 orders 事件."""
queue: asyncio.Queue[_QueueItem] = asyncio.Queue()
sse_clients.add(queue)

async def event_stream():
try:
# 首次发 hello, 让前端立即感知连接成功并拿到菜单
hello = {"menu": [m["name"] for m in MENU], "statuses": STATUSES}
yield f"event: hello\\ndata: {json.dumps(hello, ensure_ascii=False)}\\n\\n"

# 主循环: 从 Queue 取事件并以 SSE 帧格式写出
while True:
item = await queue.get()
yield f"event: {item['event']}\\ndata: {item['data']}\\n\\n"
finally:
# 客户端断开 / 异常 -> 退订, 避免泄漏
sse_clients.discard(queue)

return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)

3.3 逐段讲解

3.3.1 导入

import asyncio
import json
from typing import TypedDict

from fastapi import APIRouter
from fastapi.responses import StreamingResponse

from .models import MENU, STATUSES

• asyncio：用 asyncio.Queue 给每个连接当「信箱」；
• json：序列化 hello 消息；
• APIRouter：FastAPI 的路由分组工具，把 /sse/orders 单独装进 sse.router，第 6 章在 app.py 里 include_router 挂上去；
• StreamingResponse：FastAPI 的流式响应类，接收一个异步生成器，生成器 yield 什么就往客户端发什么；
• from .models import MENU, STATUSES：相对导入，拿菜单和状态列表，握手时发给前端。注意 . 表示「同包（src）内」。

3.3.2 Queue 事件项的类型

class _QueueItem(TypedDict):
"""broadcaster 推入 Queue 的事件项."""
event: str
data: str

• 广播器（第 5 章）往每个 Queue 推的是 {"event": "orders", "data": "…json字符串…"}；
• 用 TypedDict 给个类型名，下面 sse_clients 的类型标注更清晰；
• 名字带下划线 _QueueItem：约定俗成表示「模块内部用」。

3.3.3 路由对象 + 客户端集合

router = APIRouter()

# 所有 SSE 客户端的 Queue 集合 (由 broadcaster 写入)
sse_clients: set[asyncio.Queue[_QueueItem]] = set()

• router = APIRouter()：创建一个独立路由组，本模块所有路由都挂在它上面；
• sse_clients：所有当前 SSE 连接的 Queue 集合。这是一个模块级全局变量，会被第 5 章的 broadcaster.py 导入并往里塞事件；
• 为什么存的是 Queue 而不是连接对象？因为 SSE 的 StreamingResponse 没有现成的「发送方法」，我们用一个 Queue 当缓冲：广播器 queue.put(…)，连接协程 queue.get() 后 yield 出去。这样广播器不用等客户端，发送也不会阻塞广播循环。

3.3.4 SSE 端点

@router.get("/sse/orders")
async def sse_orders() –> StreamingResponse:
"""建立 SSE 连接, 持续接收 orders 事件."""
queue: asyncio.Queue[_QueueItem] = asyncio.Queue()
sse_clients.add(queue)

• 每个客户端连进来，就新建一个专属 Queue，加进 sse_clients 集合；
• 注意是 async def：因为里面要 await queue.get()。

3.3.5 事件流生成器

async def event_stream():
try:
# 首次发 hello, 让前端立即感知连接成功并拿到菜单
hello = {"menu": [m["name"] for m in MENU], "statuses": STATUSES}
yield f"event: hello\\ndata: {json.dumps(hello, ensure_ascii=False)}\\n\\n"

# 主循环: 从 Queue 取事件并以 SSE 帧格式写出
while True:
item = await queue.get()
yield f"event: {item['event']}\\ndata: {item['data']}\\n\\n"
finally:
# 客户端断开 / 异常 -> 退订, 避免泄漏
sse_clients.discard(queue)

这是 SSE 的核心，逐行看：

json.dumps(hello, ensure_ascii=False)：让中文（珍珠奶茶等）原样输出，不被转成 珍珠。

3.3.6 返回 StreamingResponse

return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)

• event_stream()：注意这里调用了生成器函数（带 ()），传进去的是生成器对象；
• media_type="text/event-stream"：SSE 的标志，浏览器看到这个才会按 SSE 解析，EventSource 才能正常工作；
• 三个响应头：
  • Cache-Control: no-cache：别缓存事件流，每次都要新的；
  • Connection: keep-alive：保持长连接，不要发完就断；
  • X-Accel-Buffering: no：给 nginx 看的。如果部署在 nginx 后面，nginx 默认会缓冲响应再批量发，这会让 SSE 失去实时性。这行告诉 nginx「别缓冲我，收到就转发」。

3.4 数据流回顾

浏览器 EventSource('/sse/orders')
│ HTTP GET, Upgrade 长连接
▼
sse_orders() → 新建 Queue 加入 sse_clients → 返回 StreamingResponse
│
│ event_stream() 生成器持续运行:
│ 1. 先 yield hello 帧
│ 2. 循环 await queue.get() → yield orders 帧
│
▼
浏览器收到 event:hello → 显示菜单
浏览器收到 event:orders → 刷新订单卡片
│
（断开时） finally: sse_clients.discard(queue) ← 防泄漏

谁往 Queue 里 put？答案是第 5 章的广播器。它每 0.8s 调 advance()，然后把结果 put 进每一个 sse_clients 的 Queue。

3.5 动手验证（需要先跳到第 6 章把 app.py 写完）

这一章的代码单独还不能跑（缺入口）。如果你已经按顺序写到了第 6 章，可以这样验证 SSE：

# 启动服务（第 6 章的命令）
uv run python -m uvicorn src.app:app –host 127.0.0.1 –port 8000 –reload

# 另开终端，用 curl 连 SSE（-N 表示不缓冲）
curl -N http://127.0.0.1:8000/sse/orders

你会持续看到：

event: hello
data: {"menu": ["珍珠奶茶", "杨枝甘露", …], "statuses": ["待制作", "制作中", "已完成", "已取餐"]}

event: orders
data: {"time": "…", "orders": [{…}, {…}]}

event: orders
data: {"time": "…", "orders": [{…}, {…}]}
…

每 0.8 秒一帧 orders，按 Ctrl+C 退出。

3.6 小结

下一章 → 04-WebSocket模块-ws.md：用 WebSocket 实现双向通道，让前端能发命令。

第 4 章 · WebSocket 模块 ws.py

本章目标

实现项目的第二个传输模块——WebSocket（双向通道）。和 SSE 的「只推不收」不同，WebSocket 让前端能主动发命令，服务器给明确回执。学完本章你将理解：

• WebSocket 在 FastAPI 里的写法（@router.websocket、accept、receive_text、send_text）；
• 命令分发模式：cmd + arg，每种命令一个分支，每条命令都有回执；
• 用 TypedDict 给每种消息定义类型，前端后端约定清晰；
• WebSocketDisconnect 异常 + finally 清理连接，防泄漏。

4.1 先理解 WebSocket 是什么

WebSocket 是 HTML5 的全双工协议：

• 握手阶段用 HTTP，之后「升级」成 ws://，连接保持，双方可随时互发消息；
• 浏览器用 new WebSocket(url) 连接（不自动重连，断了要自己重连，第 7 章前端会处理）；
• 适合有交互的场景：聊天、协作编辑、我们的「店员操作台」（发命令 + 拿回执）。

本项目里 WebSocket 支持四种命令：

4.2 原始完整代码

新建文件 src/ws.py，完整照抄：

"""教学模块 2: WebSocket 双向通道.

关键点
——
– 全双工: 客户端和服务端都可主动发送消息
– 接受连接后加入 ws_clients 集合, 广播器统一推送
– 接收客户端命令: ping / add / pickup / snapshot, 每条命令都有明确回执
– WebSocketDisconnect 异常表示客户端主动关闭, finally 清理连接
"""

import json
from datetime import UTC, datetime
from typing import TypedDict

from fastapi import APIRouter, WebSocket, WebSocketDisconnect

from .models import MENU, STATUSES, Order, Snapshot, new_order, orders, snapshot

class HelloMsg(TypedDict):
type: str
menu: list[str]
statuses: list[str]
orders: list[Order]

class PongMsg(TypedDict):
type: str
time: str

class AddedMsg(TypedDict):
type: str
order: Order

class PickedMsg(TypedDict):
type: str
order: Order

class SnapshotMsg(TypedDict):
type: str
time: str
orders: list[Order]

class ErrorMsg(TypedDict):
type: str
error: str

router = APIRouter()

# 所有已连接的 WebSocket 客户端 (由 broadcaster 写入)
ws_clients: set[WebSocket] = set()

def now_iso() –> str:
return datetime.now(UTC).isoformat()

@router.websocket("/ws/orders")
async def ws_orders(ws: WebSocket) –> None:
"""建立 WS 连接, 处理客户端命令."""
await ws.accept()
ws_clients.add(ws)

async def send(msg: dict[str, object]) –> None:
await ws.send_text(json.dumps(msg, ensure_ascii=False))

try:
# 握手成功后立即发 hello, 把菜单和当前快照给客户端
await send(HelloMsg(
type="hello",
menu=[m["name"] for m in MENU],
statuses=STATUSES,
orders=sorted(orders.values(), key=lambda o: –o["id"]),
))

# 命令分发
while True:
raw = await ws.receive_text()
try:
msg = json.loads(raw)
except json.JSONDecodeError:
await send(ErrorMsg(type="error", error="invalid json"))
continue

cmd = str(msg.get("cmd", "")).lower()
arg = msg.get("arg")

if cmd == "ping":
await send(PongMsg(type="pong", time=now_iso()))
elif cmd == "add":
o = new_order()
orders[o["id"]] = o
await send(AddedMsg(type="added", order=o))
elif cmd == "pickup":
try:
oid = int(arg)
except (TypeError, ValueError):
await send(ErrorMsg(type="error", error="arg must be int order_id"))
continue
o = orders.get(oid)
if not o:
await send(ErrorMsg(type="error", error=f"order {oid} not found"))
elif o["status"] != "已完成":
await send(ErrorMsg(type="error", error=f"order {oid} not ready"))
else:
o["status"] = "已取餐"
await send(PickedMsg(type="picked", order=o))
elif cmd == "snapshot":
snap: Snapshot = snapshot()
await send(SnapshotMsg(type="snapshot", time=snap["time"], orders=snap["orders"]))
else:
await send(ErrorMsg(type="error", error=f"unknown cmd: {cmd}"))
except WebSocketDisconnect:
pass
finally:
ws_clients.discard(ws)

4.3 逐段讲解

4.3.1 导入

import json
from datetime import UTC, datetime
from typing import TypedDict

from fastapi import APIRouter, WebSocket, WebSocketDisconnect

from .models import MENU, STATUSES, Order, Snapshot, new_order, orders, snapshot

• WebSocket：FastAPI 的 WebSocket 连接对象，函数参数注解为 WebSocket 时 FastAPI 自动注入；
• WebSocketDisconnect：Starlette 提供的异常，客户端主动断开时 receive_text() 会抛它；
• 从 models 导入一堆：菜单、状态、订单类型、new_order（加单用）、orders（取/改订单用）、snapshot（拉快照用）。

4.3.2 消息类型（前端/后端的契约）

class HelloMsg(TypedDict):
type: str
menu: list[str]
statuses: list[str]
orders: list[Order]

class PongMsg(TypedDict):
type: str
time: str

class AddedMsg(TypedDict):
type: str
order: Order

class PickedMsg(TypedDict):
type: str
order: Order

class SnapshotMsg(TypedDict):
type: str
time: str
orders: list[Order]

class ErrorMsg(TypedDict):
type: str
error: str

• 每种服务器→客户端的消息都定义一个 TypedDict，所有消息都有一个 type 字段表示消息类型；
• 这是 WebSocket 通信的常见模式：用 type 字段做消息分发，前端 ws.onmessage 里 switch(msg.type)；
• HelloMsg：握手后发的欢迎消息（菜单 + 当前所有订单）；
• PongMsg：ping 的回执；
• AddedMsg：add 的回执（新订单详情）；
• PickedMsg：pickup 成功的回执；
• SnapshotMsg：snapshot 的回执；
• ErrorMsg：任何错误的回执。

为什么用 TypedDict 而不是 dataclass 或 pydantic.BaseModel？因为这里消息直接 json.dumps 成 dict，TypedDict 零运行时开销，纯粹是类型提示，最轻量。

4.3.3 路由 + 客户端集合

router = APIRouter()

# 所有已连接的 WebSocket 客户端 (由 broadcaster 写入)
ws_clients: set[WebSocket] = set()

• 和 SSE 一样用 APIRouter；
• ws_clients：所有当前 WS 连接的集合。广播器（第 5 章）会遍历它 send_text；
• 注意这里存的是**WebSocket 对象本身**，不像 SSE 存 Queue。因为 WebSocket 原生支持异步 send_text，可以直接发。

4.3.4 接受连接 + 内部 send 辅助

@router.websocket("/ws/orders")
async def ws_orders(ws: WebSocket) –> None:
"""建立 WS 连接, 处理客户端命令."""
await ws.accept()
ws_clients.add(ws)

async def send(msg: dict[str, object]) –> None:
await ws.send_text(json.dumps(msg, ensure_ascii=False))

• @router.websocket(…)：声明这是一个 WebSocket 路由（不是 HTTP GET）；
• await ws.accept()：必须先 accept，完成 WebSocket 握手，之后才能收发；
• ws_clients.add(ws)：加入广播集合；
• 内部 send 辅助函数：把 dict 序列化成 JSON 字符串再发。后面所有命令回执都调它，避免重复写 json.dumps。

4.3.5 握手 hello

try:
# 握手成功后立即发 hello, 把菜单和当前快照给客户端
await send(HelloMsg(
type="hello",
menu=[m["name"] for m in MENU],
statuses=STATUSES,
orders=sorted(orders.values(), key=lambda o: –o["id"]),
))

• 和 SSE 一样，连上立刻发 hello，但 WS 的 hello 多带一个 orders：当前所有订单。这样前端一连上就能立刻渲染操作台的订单卡片，不用等下一次广播；
• sorted(…, key=lambda o: -o["id"])：按 id 倒序，最新在前。

4.3.6 命令分发循环

# 命令分发
while True:
raw = await ws.receive_text()
try:
msg = json.loads(raw)
except json.JSONDecodeError:
await send(ErrorMsg(type="error", error="invalid json"))
continue

cmd = str(msg.get("cmd", "")).lower()
arg = msg.get("arg")

• await ws.receive_text()：阻塞等待客户端发来的下一条文本消息；
• 前端发的格式约定为 JSON：{"cmd": "add"} 或 {"cmd": "pickup", "arg": 3}；
• json.loads 解析，失败就回 error: invalid json，continue 继续等下一条；
• cmd 取小写；arg 是可选参数（pickup 需要）。

4.3.7 四个命令分支

if cmd == "ping":
await send(PongMsg(type="pong", time=now_iso()))

• ping：最简单，回 pong + 当前时间。前端用来测往返延迟。

elif cmd == "add":
o = new_order()
orders[o["id"]] = o
await send(AddedMsg(type="added", order=o))

• add：调 new_order() 生成一杯随机订单，存进 orders，回 added + 新订单详情。前端拿到后可以高亮显示新订单。

elif cmd == "pickup":
try:
oid = int(arg)
except (TypeError, ValueError):
await send(ErrorMsg(type="error", error="arg must be int order_id"))
continue
o = orders.get(oid)
if not o:
await send(ErrorMsg(type="error", error=f"order {oid} not found"))
elif o["status"] != "已完成":
await send(ErrorMsg(type="error", error=f"order {oid} not ready"))
else:
o["status"] = "已取餐"
await send(PickedMsg(type="picked", order=o))

• pickup 最复杂，要带 arg（订单 id）：
• 先把 arg 转成 int，失败回错（arg must be int order_id）；
• 查订单存不存在，不存在回错；
• 只有 已完成 的订单能取餐，制作中 不行（回 not ready）；
• 满足条件就改成 已取餐，回 picked + 更新后的订单。
• 这种「校验 → 操作 → 回执」三段式是命令处理的通用模式。

elif cmd == "snapshot":
snap: Snapshot = snapshot()
await send(SnapshotMsg(type="snapshot", time=snap["time"], orders=snap["orders"]))

• snapshot：调 models.snapshot()（不改变状态）拿当前快照，回 snapshot + 时间 + 所有订单。前端想随时刷新用。

else:
await send(ErrorMsg(type="error", error=f"unknown cmd: {cmd}"))

• 未知命令：回错。这很重要——前端调试时打错命令能立刻看到反馈。

4.3.8 断开处理

except WebSocketDisconnect:
pass
finally:
ws_clients.discard(ws)

• WebSocketDisconnect：客户端主动关连接时，receive_text() 抛这个异常。我们 pass（不当作错误）；
• finally：无论正常退出还是异常，都把自己从 ws_clients 移除。和 SSE 的 finally discard 一样，防内存泄漏——否则广播器会一直往断开的连接发，send_text 抛异常还得处理。

对比 SSE：SSE 断开会让 yield 抛 GeneratorExit；WS 断开让 receive_text 抛 WebSocketDisconnect。处理位置不同，但目的一样——清理订阅。

4.4 数据流回顾

浏览器 new WebSocket('ws://…')
│ 握手升级
▼
ws_orders() → accept → 加入 ws_clients → 发 hello(含当前订单)
│
│ 循环 await receive_text():
│ {"cmd":"ping"} → 回 {"type":"pong",…}
│ {"cmd":"add"} → new_order + 回 {"type":"added",…}
│ {"cmd":"pickup","arg":3} → 改状态 + 回 {"type":"picked",…}
│ {"cmd":"snapshot"} → 回 {"type":"snapshot",…}
│ 非法/未知 → 回 {"type":"error",…}
│
（断开时） WebSocketDisconnect → finally: ws_clients.discard(ws)

此外，广播器（第 5 章）每 0.8s 也会遍历 ws_clients 给每个连接 send_text(payload)——那是另一条推送线，和这里的命令回执线并行。所以 WS 客户端会同时收到：① 自己命令的回执；② 广播器推的全量订单更新。

4.5 SSE vs WS 写法对比（先看一眼，第 9 章细讲）

4.6 动手验证（需写完第 6 章入口后）

WebSocket 不能用 curl 直接测（需要 ws 协议握手）。写完第 6 章后，最简单的验证是用浏览器控制台：

// 在 http://127.0.0.1:8000/ 页面的 F12 控制台执行
const ws = new WebSocket("ws://127.0.0.1:8000/ws/orders");
ws.onmessage = e => console.log(JSON.parse(e.data));
ws.onopen = () => ws.send(JSON.stringify({cmd: "ping"})); // 2 秒后会看到 {type:"pong"}
// 等 2 秒再试:
ws.send(JSON.stringify({cmd: "add"})); // 看到 {type:"added", order:{…}}
ws.send(JSON.stringify({cmd: "snapshot"})); // 看到 {type:"snapshot", …}

或者直接用项目自带的页面（第 7 章），点右侧「📡 ping / ➕ 加单 / 📸 snapshot」按钮，黑底日志区会显示收发消息。

下一章 → 05-广播器-broadcaster.md：写后台心跳，把数据同时推给 SSE 和 WS 两个集合。

第 5 章 · 广播器 broadcaster.py

本章目标

写一个后台协程「广播器」，它是 SSE 和 WS 共享的心跳源。每 0.8 秒：

学完本章你将理解：

• 后台任务为什么用协程而不是线程；
• asyncio.create_task + task.cancel() 的生命周期管理；
• 为什么 SSE 用 queue.put、WS 用 send_text 两种不同分发方式；
• 怎么做到「一处推进，两路分发」。

5.1 为什么需要广播器？

第 3、4 章各自维护了订阅者集合（sse_clients / ws_clients），但它们自己不知道何时该推数据。如果让每个连接自己去轮询 advance()，会有两个问题：

正确做法：只有一个时钟——广播器。它每 0.8s 调一次 advance()，把结果同时发给所有订阅者。所有连接看到的是同一份数据、同一时刻的更新。

5.2 原始完整代码

新建文件 src/broadcaster.py，完整照抄：

"""
后台广播任务: 每 0.8s 推进订单并分发给所有 SSE / WS 客户端.

这是 SSE 与 WS 共享的"心跳源". 各自维护自己的订阅者集合,
广播器只负责: 推进状态 -> 编码 JSON -> 投递到两个集合.
"""

import asyncio
import json

from .models import advance
from .sse import sse_clients
from .ws import ws_clients

async def broadcaster(interval: float = 0.8) –> None:
"""永久循环: 推进状态并广播; 协程被取消时自动退出."""
while True:
snapshot = advance()
payload = json.dumps(snapshot, ensure_ascii=False)

# SSE: 每个订阅者一份 Queue, 主线程非阻塞地塞入新事件
for queue in list(sse_clients):
await queue.put({"event": "orders", "data": payload})

# WS: 直接 send_text, 发送失败说明连接已断, 从集合移除
for ws in list(ws_clients):
try:
await ws.send_text(payload)
except Exception:
ws_clients.discard(ws)

await asyncio.sleep(interval)

5.3 逐段讲解

5.3.1 导入

import asyncio
import json

from .models import advance
from .sse import sse_clients
from .ws import ws_clients

• advance：每 tick 调一次，推进订单状态并返回快照；
• sse_clients：从 sse.py 导入的 Queue 集合；
• ws_clients：从 ws.py 导入的 WebSocket 集合；
• 注意 broadcaster.py 同时依赖两个传输模块——它是把两者串起来的「胶水」。

5.3.2 主循环

async def broadcaster(interval: float = 0.8) –> None:
"""永久循环: 推进状态并广播; 协程被取消时自动退出."""
while True:
snapshot = advance()
payload = json.dumps(snapshot, ensure_ascii=False)

• async def：协程。因为要 await 发送操作和 sleep；
• interval=0.8：默认 0.8 秒一 tick。调慢点（如 2.0）能看得更清楚，调快点（如 0.3）更刺激；
• while True：永久循环，直到被取消；
• advance()：同步调用（不是 await），因为它操作的是普通 dict，没有 IO；
• json.dumps(snapshot, ensure_ascii=False)：整个快照序列化成字符串，给 SSE 和 WS 共用一份（SSE 把它包进 data:，WS 直接发）。

5.3.3 分发给 SSE：queue.put

# SSE: 每个订阅者一份 Queue, 主线程非阻塞地塞入新事件
for queue in list(sse_clients):
await queue.put({"event": "orders", "data": payload})

• 遍历所有 SSE 客户端的 Queue，把 {"event": "orders", "data": payload} 塞进去；
• list(sse_clients)：先转 list 再遍历——防止遍历过程中集合被修改（连接断开时 sse.py 的 finally 会 discard，改变集合大小，直接遍历 set 会抛 RuntimeError: Set changed size during iteration）；
• await queue.put(…)：asyncio.Queue 默认无界，put 几乎不阻塞（除非设置了 maxsize）。所以广播器不会因为某个 SSE 客户端读得慢而被卡住——事件会在那个客户端的 Queue 里积压，等它慢慢消费。

这就是 SSE 用 Queue 的核心好处：生产者（广播器）和消费者（连接协程）解耦，互不阻塞。

5.3.4 分发给 WS：send_text + 异常清理

# WS: 直接 send_text, 发送失败说明连接已断, 从集合移除
for ws in list(ws_clients):
try:
await ws.send_text(payload)
except Exception:
ws_clients.discard(ws)

• 遍历所有 WS 连接，直接 await ws.send_text(payload)；
• 同样 list(ws_clients) 防遍历中集合变化；
• try/except：如果某个 WS 已经断了，send_text 会抛异常。我们捕获后把它从集合移除——这样下次广播就不会再尝试给它发；
• 为什么 SSE 不用 try/except？因为 SSE 是往 Queue 里塞，不直接接触连接，连接断了 Queue 还在（直到 SSE 协程的 finally 清理）。WS 是直接发，必须当场处理失败。

两种分发方式的差异，本质是「缓冲 vs 直发」：SSE 有 Queue 缓冲，WS 无缓冲直发。各有取舍：SSE 容错好但断连清理靠协程自己；WS 即时但失败要当场处理。

5.3.5 节拍

await asyncio.sleep(interval)

• 每轮结束睡 interval 秒，控制心跳频率；
• await asyncio.sleep 会让出事件循环，让其他协程（各连接的收发）有机会运行；
• 用 sleep 而不是 time.sleep：time.sleep 会阻塞整个事件循环，所有连接都会卡住。asyncio.sleep 是协程友好的。

5.4 广播器的一生（生命周期）

广播器不是自己启动的，它由 app.py 的 lifespan 在应用启动时创建：

应用启动 (lifespan yield 之前)
│
├─ asyncio.create_task(broadcaster()) ← 启动后台协程
│ │
│ └─ while True: advance() + 分发 + sleep ← 一直跑
│
▼ (yield: 应用运行中, 处理请求)
│
应用关闭 (lifespan yield 之后, finally)
│
└─ task.cancel() ← 取消广播器协程
│
└─ broadcaster 的 await 点抛 CancelledError, while True 退出

task.cancel() 会让广播器协程在下一次 await（queue.put / send_text / sleep）处抛 CancelledError，从而跳出 while True。广播器不需要自己写 try/except CancelledError——协程被取消是正常退出路径。

5.5 数据流总览（三个文件联动）

每 0.8 秒:
broadcaster.py
│
├─ advance() → models.py 推进订单, 返回 Snapshot
│
├─ json.dumps(snapshot) → payload 字符串
│
├─ for queue in sse_clients: queue.put({"event":"orders","data":payload})
│ │
│ └─ sse.py 各 event_stream() 的 await queue.get() 取出
│ → yield "event: orders\\ndata: payload\\n\\n" → 浏览器
│
└─ for ws in ws_clients: await ws.send_text(payload)
│
└─ 浏览器 ws.onmessage 收到 JSON, 按 type 分发(但广播没 type,前端当 orders 用)

注意：广播器推给 WS 的 payload 是纯快照 JSON（{"time":…,"orders":[…]}），没有 type 字段。前端的 ws.onmessage 会判断：如果消息没有 type 或者 type 不在已知列表里，就当作「订单更新」处理（详见第 7 章前端）。这是项目的一个小约定。

5.6 动手验证（需写完第 6 章）

写完 app.py 后启动服务，你会看到：

• 浏览器看板每 0.8 秒刷新一次（订单进度条在涨、状态在变）；
• 终端日志里 uvicorn 会持续打印 INFO: … 的连接信息，但不会有广播日志（广播器不 print）；
• 关闭浏览器，SSE/WS 协程的 finally 会清理 sse_clients/ws_clients，广播器下一轮遍历的集合就少了一个元素，不会报错。

5.7 小结

下一章 → 06-应用入口-app.md：写 app.py，把数据层、两个传输模块、广播器、静态页面组装成一个完整的 FastAPI 应用。

第 6 章 · 应用入口 app.py

本章目标

写 src/app.py，把前面所有零件组装成一个能跑的 FastAPI 应用。学完本章你将理解：

• FastAPI 的 lifespan（生命周期）机制：启动时干什么、关闭时干什么；
• 怎么用 asyncio.create_task 启动后台广播器，关闭时 cancel 它；
• CORS 中间件为什么「全开」；
• 怎么用 include_router 挂载独立模块的路由；
• 怎么用 StaticFiles 挂载静态目录、用 FileResponse 渲染首页。

写完这一章，后端就完整了，可以启动验证。

6.1 原始完整代码

新建文件 src/app.py，完整照抄：

"""fastapi-sse-ws 入口.

– lifespan: 启动后台广播任务 (broadcaster.py)
– CORS: 全开
– 路由: /sse/orders (sse.py)
/ws/orders (ws.py)
/ -> static/index.html
"""

import asyncio
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from pathlib import Path

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

from . import sse, ws
from .broadcaster import broadcaster

@asynccontextmanager
async def lifespan(_: FastAPI) –> AsyncGenerator[None]:
"""启动后台广播任务, 关闭时自动取消."""
task = asyncio.create_task(broadcaster())
try:
yield
finally:
task.cancel()

app = FastAPI(title="milk-tea-board", version="0.1.0", lifespan=lifespan)

# CORS 全开
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
allow_credentials=True,
)

# 挂载独立模块路由
app.include_router(sse.router)
app.include_router(ws.router)

# 静态页面 (位于项目根 static/)
STATIC_DIR = Path(__file__).parent.parent / "static"
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")

@app.get("/")
async def index() –> FileResponse:
"""首页: 渲染 static/index.html."""
return FileResponse(STATIC_DIR / "index.html")

if __name__ == "__main__":
uvicorn.run("src.app:app", host="127.0.0.1", port=8000, reload=True)

6.2 逐段讲解

6.2.1 导入

import asyncio
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from pathlib import Path

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

from . import sse, ws
from .broadcaster import broadcaster

• asyncio：启动后台任务用；
• AsyncGenerator / asynccontextmanager：给 lifespan 标注类型和装饰成异步上下文管理器（下面详述）；
• Path：跨平台拼静态目录路径；
• uvicorn：底部 if __name__ 用它直接启动；
• FastAPI / CORSMiddleware / FileResponse / StaticFiles：FastAPI 核心组件；
• from . import sse, ws：导入两个传输模块（用到它们的 router）；
• from .broadcaster import broadcaster：导入广播器协程。

6.2.2 lifespan：生命周期管理

@asynccontextmanager
async def lifespan(_: FastAPI) –> AsyncGenerator[None]:
"""启动后台广播任务, 关闭时自动取消."""
task = asyncio.create_task(broadcaster())
try:
yield
finally:
task.cancel()

这是本章最关键的一段，慢慢看：

• lifespan 是什么：FastAPI（底层 Starlette）提供的一个钩子，让你在应用启动前和关闭后各执行一段代码。它是一个异步上下文管理器；
• @asynccontextmanager：把一个带 yield 的 async def 变成异步上下文管理器。yield 之前 = 启动逻辑，yield 之后 = 关闭逻辑；
• yield 之前（启动）：asyncio.create_task(broadcaster()) 把广播器协程调度到后台运行，立刻返回一个 Task 对象。此时广播器已经开始 while True 循环了；
• yield：把控制权交还给 FastAPI，应用开始接收请求。广播器在后台持续跑；
• finally（关闭）：应用收到关闭信号（Ctrl+C / 停止）时，yield 之后执行。task.cancel() 取消广播器协程，它会在下一个 await 点抛 CancelledError 而退出，不会泄漏。

_ 参数：lifespan 函数签名要求接收 FastAPI 实例，但这里用不到，按惯例命名 _。

• 返回类型 AsyncGenerator[None]：asynccontextmanager 装饰后的函数实际返回异步生成器，None 表示 yield 出去的值不被使用。

6.2.3 创建 app + 绑定 lifespan

app = FastAPI(title="milk-tea-board", version="0.1.0", lifespan=lifespan)

• FastAPI(…)：创建应用实例；
• title / version：会显示在自动生成的 /docs（Swagger UI）页面；
• lifespan=lifespan：把上面定义的 lifespan 钩子绑上。FastAPI 会在启动/关闭时自动调用它。

6.2.4 CORS 中间件（全开）

# CORS 全开
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
allow_credentials=True,
)

• CORS（跨域资源共享）：浏览器有同源策略，默认禁止跨域请求。如果前端不在 127.0.0.1:8000 而在别处（如 localhost:5173 跑 Vue/React 开发服务器），就需要服务器允许跨域；
• 这里全开（教学项目）：
  • allow_origins=["*"]：允许任何来源；
  • allow_methods=["*"]：允许任何 HTTP 方法；
  • allow_headers=["*"]：允许任何请求头；
  • allow_credentials=True：允许带 cookie（本项目其实用不到，保留）；
• 生产环境不要这样全开，要精确指定允许的源。

注意：allow_origins=["*"] + allow_credentials=True 在严格浏览器里其实有冲突（CORS 规范不允许 * 配合 credentials），但 FastAPI 的 CORS 中间件会自动处理，教学场景没问题。

6.2.5 挂载路由

# 挂载独立模块路由
app.include_router(sse.router)
app.include_router(ws.router)

• include_router：把 sse.py 和 ws.py 里 APIRouter 上定义的路由（/sse/orders、/ws/orders）注册到主 app；
• 这样每个模块自己管自己的路由，app.py 只负责组装，职责清晰；
• 如果后面要加新模块（如 /api/v2/…），再写一个 xxx.py 定义 router，这里 include_router 即可。

6.2.6 静态资源挂载

# 静态页面 (位于项目根 static/)
STATIC_DIR = Path(__file__).parent.parent / "static"
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")

• Path(__file__)：app.py 自己的路径，即 src/app.py；
• .parent.parent：上一级上一级 = 项目根目录；
• / "static"：拼出 项目根/static，跨平台（Windows 用 \\ 也能正确处理）；
• app.mount("/static", StaticFiles(directory=STATIC_DIR), …)：把 static/ 目录挂到 /static/ URL 路径下。浏览器访问 /static/index.html 就能拿到文件。前端页面里如果有 <img src="/static/x.png"> 就会从这里找；
• name="static"：给这个挂载点起个名，方便反向引用。

6.2.7 首页路由

@app.get("/")
async def index() –> FileResponse:
"""首页: 渲染 static/index.html."""
return FileResponse(STATIC_DIR / "index.html")

• 访问根路径 / 时，返回 static/index.html 文件；
• FileResponse：把磁盘文件作为响应体返回，自动设置 Content-Type（html 文件会设成 text/html）；
• 这样用户访问 http://127.0.0.1:8000/ 直接看到看板页面（第 7 章写）。

6.2.8 直接运行入口

if __name__ == "__main__":
uvicorn.run("src.app:app", host="127.0.0.1", port=8000, reload=True)

• if __name__ == "__main__"：只有直接 python src/app.py 运行时才执行，被 import 时不执行；
• uvicorn.run("src.app:app", …)：用 uvicorn 启动应用。注意第一个参数是字符串 "src.app:app"（模块路径:变量名），不是 app 对象本身——因为 reload=True 需要重新导入模块，必须用字符串；
• host="127.0.0.1"：只监听本机（开发用）。要外网访问改成 0.0.0.0；
• port=8000：端口；
• reload=True：代码改动自动重启，开发神器。生产环境要关掉。

也可以不用这个 if，直接命令行：uv run python -m uvicorn src.app:app –reload，效果一样。第 8 章推荐用命令行方式。

6.3 应用的组装全景

src/app.py
│
├── lifespan ── create_task(broadcaster()) → 后台跑
│ → task.cancel() 关闭
│
├── app = FastAPI(lifespan=lifespan)
│
├── add_middleware(CORSMiddleware, …) ← 允许跨域
│
├── include_router(sse.router) ← 挂 /sse/orders
├── include_router(ws.router) ← 挂 /ws/orders
│
├── mount("/static", StaticFiles(…)) ← 挂静态资源
│
└── GET "/" → FileResponse(index.html) ← 首页

6.4 启动顺序

python -m uvicorn src.app:app –reload
│
├─ 导入 src.app 模块 → 创建 app, 注册所有路由
│
├─ uvicorn 调用 lifespan 进入启动阶段:
│ └─ create_task(broadcaster()) ← 广播器开始后台运行
│
├─ lifespan yield → 应用进入「运行中」, 开始接收请求
│ │
│ ├─ GET /sse/orders → sse_orders() 建长连接
│ ├─ WS /ws/orders → ws_orders() 建长连接
│ ├─ GET / → 返回 index.html
│ └─ (广播器每 0.8s 推数据给所有连接)
│
└─ Ctrl+C → lifespan finally → task.cancel() → 广播器退出 → 应用关闭

6.5 动手验证

现在后端已经完整（但前端 static/index.html 还没写，第 7 章写）。先验证后端能启动：

# 在项目根目录
uv run python -m uvicorn src.app:app –host 127.0.0.1 –port 8000 –reload

期望看到：

INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [xxxx] using StatReload
INFO: Started server process [xxxx]
INFO: Waiting for application startup.
INFO: Application startup complete.

Application startup complete. 说明 lifespan 的启动阶段跑完了，广播器已经在后台运行。

另开一个终端测试 SSE：

curl -N http://127.0.0.1:8000/sse/orders

应该持续收到 event: hello 和 event: orders 帧（按 Ctrl+C 退出）。

测试根路径：

curl -s http://127.0.0.1:8000/

此时会报 404（因为 static/index.html 还没写）。第 7 章写完前端后，这里会返回 HTML 页面。

下一章 → 07-前端看板-index.md：写 static/index.html，把 SSE 和 WS 的数据渲染成实时看板。

第 7 章 · 前端看板 static/index.html

本章目标

写一个纯 HTML + CSS + JS（无框架）的前端页面，它同时：

• 用 EventSource 连 /sse/orders 接收 SSE 推送，渲染左侧订单看板；
• 用 WebSocket 连 /ws/orders 接收广播 + 发命令（ping/add/snapshot），渲染右侧操作台和通信日志；
• 断线自动重连；
• 把订单渲染成带进度条、状态色块的卡片，顶部显示统计和实时时钟。

写完这一章，整个项目就齐了，可以跑起来看效果。

本章代码很长（300 多行），但全是原始代码，照抄即可。讲解分「CSS / HTML / JS」三块，可以抄完再看。

7.1 原始完整代码

新建文件 static/index.html，完整照抄（这一份是项目里唯一的前端文件）：

<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>奶茶店实时订单看板</title>
<style>
/* ———- 设计变量 ———- */
:root {
–bg: #faf3e8;
–bg-soft: #fff8ef;
–card-bg: #ffffff;
–card-border: #f0dcc4;
–card-shadow: 0 4px 16px rgba(180, 120, 60, .08);
–text: #3d2c1e;
–text-soft: #8d6e63;
–text-mute: #bca995;
–accent: #ff7043;
–accent-dark: #e64a19;
–accent-soft: #ffe0b2;
–tea: #ff8a65;
–tea-2: #ffb74d;
–green: #66bb6a;
–green-bg: #e8f5e9;
–blue: #42a5f5;
–blue-bg: #e3f2fd;
–gray-bg: #eceff1;
–radius: 14px;
–radius-sm: 8px;
}

* { box-sizing: border-box; margin: 0; padding: 0; }
html, body { height: 100%; }
body {
font-family: -apple-system, "PingFang SC", "Microsoft YaHei", system-ui, sans-serif;
background: var(–bg); color: var(–text); min-height: 100vh;
-webkit-font-smoothing: antialiased;
}

/* ———- 顶部导航 ———- */
header {
padding: 0 28px; height: 64px;
background: linear-gradient(135deg, var(–tea) 0%, var(–tea-2) 100%);
color: #fff; display: flex; gap: 20px; align-items: center;
box-shadow: 0 2px 12px rgba(255, 138, 101, .25);
position: sticky; top: 0; z-index: 100;
}
header h1 { font-size: 18px; font-weight: 700; letter-spacing: .5px; white-space: nowrap; }
.header-right { margin-left: auto; display: flex; gap: 10px; align-items: center; }
.badge {
font-size: 12px; padding: 5px 14px; border-radius: 999px;
background: rgba(255,255,255,.2); backdrop-filter: blur(4px);
display: flex; align-items: center; gap: 6px; white-space: nowrap;
transition: background .3s;
}
.badge .dot { width: 7px; height: 7px; border-radius: 50%; background: #c8e6c9; }
.badge .dot.off { background: #ef5350; }
.badge .dot.live { background: #c8e6c9; animation: pulse 1.5s infinite; }
@keyframes pulse { 0%,100%{opacity:1} 50%{opacity:.3} }

/* ———- 统计条 ———- */
.stats-bar {
display: flex; gap: 12px; padding: 20px 28px 0;
}
.stat-chip {
flex: 1; background: var(–card-bg); border: 1px solid var(–card-border);
border-radius: var(–radius-sm); padding: 12px 16px;
display: flex; flex-direction: column; gap: 2px;
box-shadow: 0 2px 8px rgba(180,120,60,.06);
}
.stat-chip .num { font-size: 22px; font-weight: 700; color: var(–accent-dark); line-height: 1.2; }
.stat-chip .lbl { font-size: 11px; color: var(–text-soft); }
.stat-chip.green .num { color: var(–green); }
.stat-chip.blue .num { color: var(–blue); }
.stat-chip.gray .num { color: var(–text-soft); }

/* ———- 主体布局 ———- */
main { padding: 20px 28px 28px; display: grid; grid-template-columns: 1.4fr 1fr; gap: 20px; align-items: start; }
@media (max-width: 880px) { main { grid-template-columns: 1fr; } .stats-bar { flex-wrap: wrap; } .stat-chip { min-width: 45%; } }
section {
border: 1px solid var(–card-border); border-radius: var(–radius); padding: 18px;
background: var(–card-bg); box-shadow: var(–card-shadow);
}
section h2 { font-size: 14px; color: var(–accent-dark); margin-bottom: 14px; display: flex; align-items: center; gap: 6px; }
section h2 .tag {
font-size: 10px; font-weight: 600; padding: 2px 8px; border-radius: 4px;
background: var(–accent-soft); color: var(–accent-dark);
}

/* ———- 订单卡片 ———- */
.orders { display: grid; grid-template-columns: repeat(auto-fill, minmax(210px, 1fr)); gap: 12px; }
.card {
border: 1px solid var(–accent-soft); border-radius: 10px; padding: 14px;
background: linear-gradient(180deg, var(–bg-soft) 0%, var(–card-bg) 100%);
position: relative; transition: transform .2s, box-shadow .2s;
overflow: hidden;
}
.card:hover { transform: translateY(-2px); box-shadow: 0 6px 20px rgba(255,138,101,.15); }
.card::before {
content: ''; position: absolute; left: 0; top: 0; bottom: 0; width: 4px;
background: var(–accent); border-radius: 4px 0 0 4px;
}
.card.s-done::before { background: var(–green); }
.card.s-picked::before { background: var(–text-mute); }
.card .id { position: absolute; top: 10px; right: 12px; font-size: 11px; color: var(–text-mute); font-weight: 600; }
.card h3 { margin-bottom: 4px; font-size: 15px; color: var(–accent-dark); font-weight: 700; }
.card .meta { font-size: 11px; color: var(–text-soft); margin-bottom: 10px; line-height: 1.6; }
.card .meta span { display: inline-block; padding: 1px 6px; background: var(–bg); border-radius: 4px; margin: 2px 2px 0 0; }
.bar-wrap { display: flex; align-items: center; gap: 8px; margin-bottom: 8px; }
.bar { flex: 1; height: 6px; background: var(–accent-soft); border-radius: 3px; overflow: hidden; }
.bar > div { height: 100%; background: linear-gradient(90deg, var(–accent), var(–tea-2)); transition: width .5s cubic-bezier(.4,0,.2,1); border-radius: 3px; }
.bar-wrap .pct { font-size: 11px; color: var(–text-soft); font-weight: 600; min-width: 32px; text-align: right; }
.status { display: inline-flex; align-items: center; gap: 4px; padding: 3px 10px; border-radius: 999px; font-size: 11px; font-weight: 600; }
.status::before { content:''; width:5px; height:5px; border-radius:50%; background: currentColor; }
.s-待制作 { background: #fff3e0; color: #ef6c00; }
.s-制作中 { background: #ffe0b2; color: #e65100; }
.s-制作中::before { animation: pulse 1.2s infinite; }
.s-已完成 { background: var(–green-bg); color: #2e7d32; }
.s-已取餐 { background: var(–gray-bg); color: #607d8b; }

/* ———- 控制面板 ———- */
.panel { display: flex; flex-direction: column; gap: 12px; }
input {
width: 100%; padding: 10px 14px; background: var(–bg-soft);
border: 1px solid var(–card-border); border-radius: var(–radius-sm);
font: inherit; color: var(–text); transition: border-color .2s;
}
input:focus { outline: none; border-color: var(–accent); }
.btn-row { display: flex; gap: 8px; flex-wrap: wrap; }
button {
padding: 9px 18px; background: var(–accent); border: 0; border-radius: var(–radius-sm);
color: #fff; cursor: pointer; font: inherit; font-weight: 600; font-size: 13px;
transition: all .2s; display: flex; align-items: center; gap: 4px;
}
button:hover { background: var(–accent-dark); transform: translateY(-1px); box-shadow: 0 3px 10px rgba(255,112,67,.3); }
button:active { transform: translateY(0); }
button.ghost { background: var(–accent-soft); color: var(–accent-dark); }
button.ghost:hover { background: #ffcc80; box-shadow: 0 3px 10px rgba(255,224,178,.5); }

/* ———- WS 日志 ———- */
.log-header { display: flex; justify-content: space-between; align-items: center; }
.log-header span { font-size: 11px; color: var(–text-mute); }
pre {
background: #2d2018; color: #ffe0b2; padding: 12px; border-radius: var(–radius-sm);
overflow: auto; max-height: 220px; font-size: 11px; line-height: 1.6;
font-family: "SF Mono", "Cascadia Code", Consolas, monospace;
}
pre .arrow-in { color: #66bb6a; }
pre .arrow-out { color: #ffb74d; }
pre .type { color: #42a5f5; font-weight: 600; }

/* ———- 空状态 ———- */
.empty {
text-align: center; padding: 32px 16px; color: var(–text-mute); font-size: 13px;
}
.empty .icon { font-size: 28px; display: block; margin-bottom: 8px; opacity: .5; }

.meta-line { font-size: 11px; color: var(–text-soft); margin-top: 10px; padding-top: 10px; border-top: 1px dashed var(–card-border); }

/* ———- 卡片入场动画 ———- */
@keyframes cardIn { from { opacity:0; transform: translateY(8px); } to { opacity:1; transform: translateY(0); } }
.card { animation: cardIn .3s ease-out; }
</style>
</head>
<body>
<header>
<h1>🧋 奶茶店实时订单看板</h1>
<div class="header-right">
<span class="badge"><span class="dot off" id="sse-dot"></span><span id="sse-status">SSE: …</span></span>
<span class="badge"><span class="dot off" id="ws-dot"></span><span id="ws-status">WS: …</span></span>
<span class="badge" id="clock">–:–:–</span>
</div>
</header>

<div class="stats-bar">
<div class="stat-chip"><span class="num" id="stat-total">0</span><span class="lbl">总订单</span></div>
<div class="stat-chip"><span class="num" id="stat-making">0</span><span class="lbl">制作中</span></div>
<div class="stat-chip green"><span class="num" id="stat-done">0</span><span class="lbl">已完成</span></div>
<div class="stat-chip gray"><span class="num" id="stat-picked">0</span><span class="lbl">已取餐</span></div>
</div>

<main>
<section>
<h2>📋 实时订单 <span class="tag">SSE 单向推送</span></h2>
<div class="orders" id="sse-orders">
<div class="empty"><span class="icon">🧋</span>等待新订单…</div>
</div>
<div class="meta-line" id="sse-meta">尚未连接</div>
</section>
<section>
<h2>🎛️ 店员操作台 <span class="tag">WebSocket 双向通道</span></h2>
<div class="orders" id="ws-orders" style="margin-bottom:14px">
<div class="empty"><span class="icon">🎛️</span>等待新订单…</div>
</div>
<div class="panel">
<input id="cmd" placeholder='{"cmd":"pickup","arg":3} 或 {"cmd":"add"}' />
<div class="btn-row">
<button onclick='sendCmd({cmd:"ping"})'>📡 ping</button>
<button onclick='sendCmd({cmd:"add"})'>➕ 加单</button>
<button class="ghost" onclick='sendCmd({cmd:"snapshot"})'>📸 snapshot</button>
</div>
<div class="log-header"><span>WS 通信日志</span><span id="ws-log-count">0 条</span></div>
<pre id="ws-log"></pre>
</div>
</section>
</main>

<script>
const STATUS_CLASS = {'待制作':'s-待制作','制作中':'s-制作中','已完成':'s-已完成','已取餐':'s-已取餐'};
const STATUS_CARD = {'已完成':'s-done','已取餐':'s-picked'};

const fmtTime = (iso) => {
const d = new Date(iso); return d.toLocaleTimeString('zh-CN', {hour12: false});
};

// 更新统计条
function updateStats(list) {
const s = { total:0, making:0, done:0, picked:0 };
for (const o of (list || [])) {
s.total++;
if (o.status === '制作中' || o.status === '待制作') s.making++;
else if (o.status === '已完成') s.done++;
else if (o.status === '已取餐') s.picked++;
}
document.getElementById('stat-total').textContent = s.total;
document.getElementById('stat-making').textContent = s.making;
document.getElementById('stat-done').textContent = s.done;
document.getElementById('stat-picked').textContent = s.picked;
}

const renderOrders = (el, list) => {
if (!list || !list.length) {
el.innerHTML = '<div class="empty"><span class="icon">🧋</span>暂无订单</div>';
updateStats([]);
return;
}
el.innerHTML = list.map(o => `
<div class="card ${STATUS_CARD[o.status] || ''}">
<span class="id">#${o.id}</span>
<h3>${o.drink}</h3>
<div class="meta">
<span>${o.base}</span><span>${o.sugar}</span><span>${o.ice}</span><span>${o.topping}</span><span>¥${o.price}</span>
</div>
<div class="bar-wrap">
<div class="bar"><div style="width:${o.progress}%"></div></div>
<span class="pct">${o.progress}%</span>
</div>
<span class="status ${STATUS_CLASS[o.status] || ''}">${o.status}</span>
<span class="meta-line" style="margin-left:8px;border:0;padding:0">${fmtTime(o.created_at)}</span>
</div>
`).join('');
updateStats(list);
};

// ———- SSE ———-
function connectSSE() {
const es = new EventSource('/sse/orders');
const status = document.getElementById('sse-status');
const dot = document.getElementById('sse-dot');
const meta = document.getElementById('sse-meta');
const box = document.getElementById('sse-orders');
es.addEventListener('hello', (e) => {
status.textContent = 'SSE: 已连接';
dot.className = 'dot live';
const info = JSON.parse(e.data);
meta.textContent = '菜单: ' + info.menu.join(' / ');
});
es.addEventListener('orders', (e) => {
const { time, orders } = JSON.parse(e.data);
renderOrders(box, orders);
document.getElementById('clock').textContent = fmtTime(time);
});
es.onerror = () => {
status.textContent = 'SSE: 断开, 3s 后重连';
dot.className = 'dot off';
es.close(); setTimeout(connectSSE, 3000);
};
}
connectSSE();

// ———- WebSocket ———-
const log = document.getElementById('ws-log');
let logCount = 0;
const wsLog = (m, dir) => {
const line = (typeof m === 'string' ? m : JSON.stringify(m));
const arrow = dir === 'out' ? '<span class="arrow-out">→</span>' : '<span class="arrow-in">←</span>';
// 高亮 type 字段
const colored = line.replace(/"type":"([^"]+)"/g, '"type":"<span class="type">$1</span>"');
log.innerHTML = `${arrow} ${colored}\\n` + log.innerHTML.split('\\n').slice(0, 40).join('\\n');
logCount++;
document.getElementById('ws-log-count').textContent = logCount + ' 条';
};
function connectWS() {
const proto = location.protocol === 'https:' ? 'wss' : 'ws';
const ws = new WebSocket(`${proto}://${location.host}/ws/orders`);
const status = document.getElementById('ws-status');
const dot = document.getElementById('ws-dot');
const box = document.getElementById('ws-orders');
ws.onopen = () => { status.textContent = 'WS: 已连接'; dot.className = 'dot live'; };
ws.onmessage = (e) => {
const msg = JSON.parse(e.data);
wsLog(msg, 'in');
if (msg.type === 'hello' || msg.type === 'snapshot') {
renderOrders(box, msg.orders || []);
}
};
ws.onclose = () => { status.textContent = 'WS: 断开, 3s 后重连'; dot.className = 'dot off'; setTimeout(connectWS, 3000); };
window._ws = ws;
}
connectWS();
function sendCmd(obj) {
const ws = window._ws;
if (!ws || ws.readyState !== 1) return;
ws.send(JSON.stringify(obj));
wsLog(obj, 'out');
}
document.getElementById('cmd').addEventListener('keydown', (e) => {
if (e.key === 'Enter') { try { sendCmd(JSON.parse(e.target.value)); } catch (_) {} }
});
</script>
</body>
</html>

7.2 代码结构总览

index.html
├── <head>
│ ├── <meta> 编码/视口
│ └── <style> 全部 CSS（设计变量 + 各区块样式）
├── <body>
│ ├── <header> 标题 + SSE/WS 状态灯 + 时钟
│ ├── <div class="stats-bar"> 4 个统计 chip
│ ├── <main>
│ │ ├── <section> 左：SSE 实时订单（卡片网格）
│ │ └── <section> 右：WS 操作台（卡片 + 输入框 + 按钮 + 日志）
│ └── <script> 全部 JS（渲染 + SSE 连接 + WS 连接 + 命令发送）

7.3 CSS 讲解（设计变量 + 关键样式）

7.3.1 设计变量 :root

:root {
–bg: #faf3e8;
–accent: #ff7043;
–tea: #ff8a65;
…
}

• 把所有颜色、圆角抽成 CSS 变量，统一管理「奶茶暖色调」主题；
• 改主题只要动 :root 里的值，不用全局搜替换。

7.3.2 关键样式

• .badge .dot.live：连接成功时小绿点 + pulse 动画（透明度闪烁），表示「活着」；
• .card::before：每张订单卡片左侧 4px 色条，颜色随状态变（默认橙、s-done 绿、s-picked 灰）；
• .s-制作中::before：制作中的状态点也有 pulse 动画，视觉上「在动」；
• .bar > div：进度条，transition: width .5s 让进度变化有平滑过渡；
• @keyframes cardIn：订单卡片入场动画（淡入 + 上移），新增订单有「冒出来」的感觉；
• @media (max-width: 880px)：窄屏时主体变单列、统计条换行，响应式。

7.4 HTML 讲解（关键 id）

页面里很多元素带 id，JS 靠这些 id 操作它们：

三个按钮用内联 onclick='sendCmd({cmd:"ping"})' 直接调 JS 函数，省事。

7.5 JS 讲解（核心逻辑）

7.5.1 渲染函数 renderOrders

const renderOrders = (el, list) => {
if (!list || !list.length) { el.innerHTML = '…暂无订单…'; updateStats([]); return; }
el.innerHTML = list.map(o => `
<div class="card ${STATUS_CARD[o.status] || ''}">
<span class="id">#${o.id}</span>
<h3>${o.drink}</h3>
…
<div class="bar"><div style="width:${o.progress}%"></div></div>
<span class="status ${STATUS_CLASS[o.status] || ''}">${o.status}</span>
</div>
`).join('');
updateStats(list);
};

• 接收一个 DOM 元素和订单数组，用 map 把每个订单拼成卡片 HTML 字符串，最后 join('') 一次性赋给 innerHTML；
• STATUS_CLASS / STATUS_CARD 是状态→CSS 类的映射表，给卡片加状态色；
• 每次渲染都调 updateStats 更新顶部统计。

这种「每次全量重渲染」对几十个订单完全没问题，简单直接。订单量大时可改成 diff 更新。

7.5.2 SSE 连接 connectSSE

function connectSSE() {
const es = new EventSource('/sse/orders');
es.addEventListener('hello', (e) => {
document.getElementById('sse-status').textContent = 'SSE: 已连接';
document.getElementById('sse-dot').className = 'dot live';
const info = JSON.parse(e.data);
document.getElementById('sse-meta').textContent = '菜单: ' + info.menu.join(' / ');
});
es.addEventListener('orders', (e) => {
const { time, orders } = JSON.parse(e.data);
renderOrders(box, orders);
document.getElementById('clock').textContent = fmtTime(time);
});
es.onerror = () => {
document.getElementById('sse-status').textContent = 'SSE: 断开, 3s 后重连';
document.getElementById('sse-dot').className = 'dot off';
es.close(); setTimeout(connectSSE, 3000);
};
}
connectSSE();

• new EventSource('/sse/orders')：建立 SSE 连接。浏览器自动处理重连，但这里我们手动重连——为了能在断开时更新状态灯；
• addEventListener('hello', …)：监听服务器发的 hello 事件（event: hello 帧），收到就把状态灯变绿、显示菜单；
• addEventListener('orders', …)：监听 orders 事件，每次收到就渲染订单卡片 + 更新时钟；
• onerror：出错时手动 close + 3 秒后重连，并把状态灯变红。

注意：EventSource 本身有自动重连，但配合手动 close + setTimeout 能更好控制状态显示。这里两者结合。

7.5.3 WS 连接 connectWS

function connectWS() {
const proto = location.protocol === 'https:' ? 'wss' : 'ws';
const ws = new WebSocket(`${proto}://${location.host}/ws/orders`);
ws.onopen = () => { status.textContent = 'WS: 已连接'; dot.className = 'dot live'; };
ws.onmessage = (e) => {
const msg = JSON.parse(e.data);
wsLog(msg, 'in');
if (msg.type === 'hello' || msg.type === 'snapshot') {
renderOrders(box, msg.orders || []);
}
};
ws.onclose = () => { status.textContent = 'WS: 断开, 3s 后重连'; dot.className = 'dot off'; setTimeout(connectWS, 3000); };
window._ws = ws;
}
connectWS();

• location.protocol 判断当前是 http 还是 https，对应用 ws:// 或 wss://（加密）；
• onopen：连上后状态灯变绿；
• onmessage：每收到一条消息：① 写进日志 wsLog(msg, 'in')；② 如果是 hello 或 snapshot（带 orders），渲染右侧订单卡片；
• onclose：WebSocket 不自动重连，必须自己处理——3 秒后重连；
• window._ws = ws：把连接对象存到全局，sendCmd 函数要用。