409 lines
13 KiB
Markdown
409 lines
13 KiB
Markdown
# Redis Stream Realtime Event Architecture Design
|
||
|
||
> 日期: 2026-05-27
|
||
> 范围: PolyWeather 网站图表实时观测事件层
|
||
> 目标服务器: 2 vCPU / 8 GB RAM / 50 GB 系统盘
|
||
> 实施状态: 2026-05-28 已落地为 `v1.8.1`
|
||
|
||
## 背景
|
||
|
||
当前实时层已经具备生产化雏形：
|
||
|
||
- 后端已有 `city_observation_patch.v1` schema。
|
||
- `web/realtime_event_store.py` 使用 SQLite 保存 `observation_patch_events`，支持 `since_revision` replay。
|
||
- `/api/events?cities=...&since_revision=...&replay_limit=...` 已经是前端 SSE 入口。
|
||
- 前端 `frontend/hooks/use-sse-patches.ts` 使用 `EventSource`，并按当前可见城市订阅。
|
||
|
||
这解决了单进程、单实例下的实时推送和短窗口 replay，但还不够“一步到位”：
|
||
|
||
- 多 worker / 多实例时，进程内广播不能共享。
|
||
- SQLite replay log 可以补数据，但不适合作为跨实例 live fanout。
|
||
- 服务重启、浏览器后台恢复、SSE 断线后，需要更稳定的事件源。
|
||
- 后续如果把采集器拆成独立进程，需要一个明确的后端事件总线。
|
||
|
||
## 目标
|
||
|
||
升级为“Redis Stream 主事件日志 + SSE 浏览器传输 + SQLite fallback”的架构：
|
||
|
||
- Redis Stream 是生产环境实时事件主通道。
|
||
- SSE 仍然是浏览器唯一实时入口。
|
||
- 前端协议保持 `city_observation_patch.v1`，不让浏览器感知 Redis。
|
||
- 继续使用 numeric `revision`，兼容现有前端 `since_revision` 逻辑。
|
||
- 支持按城市订阅、断线 replay、多 worker fanout。
|
||
- 事件只保留短窗口；当前默认按 Redis Stream `MAXLEN` 控制，建议以约 24 小时 replay 为目标，不做 72 小时保留。
|
||
|
||
## 非目标
|
||
|
||
这次不做以下事情：
|
||
|
||
- 不引入 Kafka、RabbitMQ 或复杂 consumer group。
|
||
- 不把天气历史数据全部 event sourcing。
|
||
- 不让浏览器直接连接 Redis。
|
||
- 不把 DEB 预测、模型曲线、概率分布都改成事件流。
|
||
- 不做永久行情归档；超过 replay 窗口时继续走 HTTP full detail resync。
|
||
|
||
## 推荐架构
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
Collector["WeatherDataCollector / city refresh"] --> Ingest["POST /api/internal/collector-patch"]
|
||
Ingest --> Normalize["city_observation_patch.v1 normalizer"]
|
||
Normalize --> Store["RealtimeEventStore interface"]
|
||
Store --> Redis["Redis Stream: stream:city_observation"]
|
||
Store --> SQLite["SQLite fallback: observation_patch_events"]
|
||
Redis --> WorkerSub["Redis XREAD subscriber per API worker"]
|
||
WorkerSub --> LocalFanout["Local SSE fanout"]
|
||
SQLite --> ReplayFallback["SQLite replay fallback"]
|
||
SSE["GET /api/events?cities=...&since_revision=..."] --> Browser["Frontend EventSource"]
|
||
LocalFanout --> SSE
|
||
Redis --> SSE
|
||
ReplayFallback --> SSE
|
||
Browser --> Chart["Chart applies patch"]
|
||
Browser -->|resync_required| Detail["HTTP /api/city/{city}/detail"]
|
||
```
|
||
|
||
关键原则：
|
||
|
||
- Redis Stream 负责后端事件保存、短窗口 replay、多实例共享。
|
||
- SSE 负责服务端到浏览器的单向推送。
|
||
- SQLite 只做 fallback 和本地开发，不再作为生产主事件总线。
|
||
- 前端只认 patch event，不关心事件来自 Redis 还是 SQLite。
|
||
|
||
## 事件模型
|
||
|
||
继续使用现有 schema：
|
||
|
||
```json
|
||
{
|
||
"type": "city_observation_patch.v1",
|
||
"revision": 12345,
|
||
"city": "taipei",
|
||
"source": "cwa",
|
||
"obs_time": "2026-05-27T10:00:00+08:00",
|
||
"observed_at_utc": "2026-05-27T02:00:00Z",
|
||
"observed_at_local": "2026-05-27T10:00:00+08:00",
|
||
"city_local_date": "2026-05-27",
|
||
"city_timezone": "Asia/Taipei",
|
||
"source_cadence_sec": 600,
|
||
"ts": 1780000000000,
|
||
"payload": {
|
||
"temp": 34.2,
|
||
"max_so_far": 34.2,
|
||
"station_code": "466920",
|
||
"station_label": "中央气象署台北站",
|
||
"series_key": "settlement",
|
||
"unit": "celsius"
|
||
}
|
||
}
|
||
```
|
||
|
||
`revision` 仍然是前端公开排序键，必须是全局单调递增整数。
|
||
|
||
Redis Stream 自身的 stream ID，例如 `1780000000000-0`，只作为后端内部定位信息，不暴露给前端作为主 revision。
|
||
|
||
## Redis Stream 设计
|
||
|
||
Stream key:
|
||
|
||
```text
|
||
stream:city_observation
|
||
```
|
||
|
||
Revision key:
|
||
|
||
```text
|
||
counter:city_observation_revision
|
||
```
|
||
|
||
每条 stream entry 字段：
|
||
|
||
- `revision`: 全局递增整数，公开给 SSE/前端。
|
||
- `type`: `city_observation_patch.v1`
|
||
- `city`: normalized city key。
|
||
- `source`: source code，例如 `cwa`, `metar`, `amos`, `amsc_awos`。
|
||
- `obs_time`: 原始观测时间。
|
||
- `payload_json`: compact JSON。
|
||
- `created_at_ms`: 写入 Redis 的 UTC epoch ms。
|
||
- `producer_id`: API worker / collector instance id，便于排查。
|
||
|
||
写入必须通过 Lua 脚本原子完成：
|
||
|
||
1. `INCR counter:city_observation_revision`
|
||
2. `XADD stream:city_observation MAXLEN ~ {maxlen} * revision ... payload_json ...`
|
||
3. 返回 `{revision, stream_id}`
|
||
|
||
原因：
|
||
|
||
- 避免 `INCR` 成功但 `XADD` 失败后造成非必要 gap。
|
||
- 让 revision 分配和事件入流成为一个原子操作。
|
||
- 保持现有前端 numeric `since_revision` 不变。
|
||
|
||
## Replay 策略
|
||
|
||
`GET /api/events` 收到 `since_revision` 后：
|
||
|
||
1. 读取 Redis Stream 最旧 entry，拿到 oldest revision。
|
||
2. 如果 `since_revision > 0` 且 `since_revision < oldest_revision - 1`，发送 `resync_required`。
|
||
3. 从 Redis Stream 顺序扫描事件。
|
||
4. 过滤 `revision > since_revision`。
|
||
5. 过滤 `city IN subscribed_cities`。
|
||
6. 返回最多 `replay_limit + 1` 条。
|
||
7. 如果超过 `replay_limit`，发送 `resync_required`，让前端 HTTP resync。
|
||
|
||
当前事件量很小，单个 stream 顺序扫描 24 小时事件是可以接受的：
|
||
|
||
- 30 城市 * 1 条/分钟 * 24 小时 = 43,200 条。
|
||
- replay 通常只发生在断线重连，不是高频查询。
|
||
|
||
如果后续城市量扩大，或者确实需要超过 24 小时的 replay，再加 per-city stream 或 Redis sorted index。
|
||
|
||
## Live Fanout
|
||
|
||
每个 API worker 启动一个 Redis subscriber loop：
|
||
|
||
```text
|
||
XREAD BLOCK 5000 STREAMS stream:city_observation {last_seen_stream_id}
|
||
```
|
||
|
||
读取到事件后：
|
||
|
||
1. 解析 `city_observation_patch.v1`。
|
||
2. 按本进程 SSE 连接的 `cities` subscription 过滤。
|
||
3. 推送给本进程匹配的浏览器连接。
|
||
4. 更新 worker 内存中的 `last_seen_stream_id`。
|
||
|
||
ingest endpoint 写入 Redis 后不直接广播本地连接，由 subscriber loop 统一 fanout。这样避免同一个 worker 上“写入后本地广播一次、subscriber 又广播一次”的重复事件。
|
||
|
||
前端仍然用 revision 去重。即使极端情况下收到重复 revision，也应忽略旧 revision。
|
||
|
||
## SQLite Fallback
|
||
|
||
保留当前 SQLite store，但角色调整为：
|
||
|
||
- 本地开发无需 Redis 时使用。
|
||
- Redis 不可用且允许降级时使用。
|
||
- 单实例临时运行时可用。
|
||
|
||
新增环境变量：
|
||
|
||
```text
|
||
POLYWEATHER_EVENT_STORE=redis
|
||
POLYWEATHER_EVENT_STORE_FALLBACK=sqlite
|
||
POLYWEATHER_REDIS_URL=redis://127.0.0.1:6379/0
|
||
POLYWEATHER_REDIS_STREAM_KEY=stream:city_observation
|
||
POLYWEATHER_REDIS_STREAM_MAXLEN=50000
|
||
POLYWEATHER_PATCH_EVENT_RETENTION_HOURS=24
|
||
POLYWEATHER_REDIS_REQUIRED=true
|
||
```
|
||
|
||
推荐生产策略：
|
||
|
||
- `POLYWEATHER_EVENT_STORE=redis`
|
||
- `POLYWEATHER_REDIS_REQUIRED=true`
|
||
- Redis 写入失败时 ingest 返回 503，不广播不可 replay 的事件。
|
||
|
||
推荐本地策略：
|
||
|
||
- 不配置 Redis，自动使用 SQLite。
|
||
- 或 `POLYWEATHER_EVENT_STORE=sqlite`。
|
||
|
||
## Redis 容量与配置
|
||
|
||
当前 VPS 可以跑本机 Redis。
|
||
|
||
建议：
|
||
|
||
```text
|
||
bind 127.0.0.1
|
||
protected-mode yes
|
||
appendonly yes
|
||
appendfsync everysec
|
||
maxmemory 512mb
|
||
maxmemory-policy noeviction
|
||
```
|
||
|
||
默认 `MAXLEN ~ 50000` 足够约 24 小时事件：
|
||
|
||
- 30 城市 * 1/min * 24h = 43,200 条。
|
||
- payload 约 1-3 KB，实际内存取决于 Redis Stream overhead。
|
||
- 512 MB Redis maxmemory 对当前事件量足够。
|
||
|
||
注意：这个事件流不是永久数据仓库。超过 24 小时窗口后，前端应 HTTP resync 当前城市 detail。
|
||
|
||
## 模块边界
|
||
|
||
新增/调整后端接口：
|
||
|
||
### `web/realtime_event_store.py`
|
||
|
||
定义统一接口：
|
||
|
||
```python
|
||
class RealtimeEventStore:
|
||
def append_event(self, event: dict) -> dict: ...
|
||
def replay_events(self, *, cities: set[str] | None, since_revision: int, limit: int) -> list[dict]: ...
|
||
def replay_requires_resync(self, *, cities: set[str] | None, since_revision: int, replay_count: int, limit: int) -> bool: ...
|
||
def latest_revision(self) -> int: ...
|
||
```
|
||
|
||
保留现有 SQLite 实现，新增 Redis 实现。
|
||
|
||
### `web/redis_realtime_event_store.py`
|
||
|
||
职责：
|
||
|
||
- Redis 连接与健康检查。
|
||
- Lua append script。
|
||
- Redis Stream replay。
|
||
- maxlen trim。
|
||
- entry 到 SSE event 的反序列化。
|
||
|
||
### `web/sse_manager.py`
|
||
|
||
职责：
|
||
|
||
- 管理本进程 SSE 连接。
|
||
- 记录每个连接订阅城市集合。
|
||
- 将 Redis subscriber 读到的 event fanout 给匹配连接。
|
||
|
||
### `web/routers/sse_router.py`
|
||
|
||
职责：
|
||
|
||
- `/api/internal/collector-patch` 规范化并写入 event store。
|
||
- `/api/events` 处理 connected、replay、heartbeat、live stream。
|
||
- 当 replay 不完整时发送 `resync_required`。
|
||
|
||
### 前端
|
||
|
||
前端原则上不需要大改：
|
||
|
||
- `use-sse-patches.ts` 继续使用 `EventSource`。
|
||
- `lastRevision` 仍然是 number。
|
||
- `resync_required` 继续触发当前可见城市 HTTP detail refresh。
|
||
- 继续按可见城市列表构建 `cities` 参数。
|
||
|
||
## 运行流程
|
||
|
||
### 首屏
|
||
|
||
1. 前端 HTTP 加载 terminal rows。
|
||
2. 可见图表 HTTP 加载 full detail。
|
||
3. 前端连接 `/api/events?cities=...&since_revision=...&replay_limit=500`。
|
||
4. 后端先 replay Redis Stream 中缺失事件，再进入 live stream。
|
||
|
||
### 新观测
|
||
|
||
1. 采集器产生 `city_patch` 或 v1 patch。
|
||
2. ingest endpoint 规范化成 `city_observation_patch.v1`。
|
||
3. Redis Lua script 写入 stream 并生成 revision。
|
||
4. 每个 API worker 的 subscriber loop 读到事件。
|
||
5. worker fanout 给订阅对应城市的 SSE 连接。
|
||
6. 前端 apply patch，图表无痛追加点。
|
||
|
||
### 浏览器后台恢复
|
||
|
||
1. EventSource 如果断线，前端按指数退避重连。
|
||
2. URL 带 `since_revision=lastRevision`。
|
||
3. Redis replay 补齐后台期间错过的事件。
|
||
4. 如果 revision 太旧或 replay 超限，前端收到 `resync_required`，HTTP 重拉 full detail。
|
||
|
||
## 错误处理
|
||
|
||
- Redis append 失败且 `POLYWEATHER_REDIS_REQUIRED=true`：ingest 返回 503，不广播。
|
||
- Redis append 失败且允许 fallback：写入 SQLite，并在 health 状态标记 degraded。
|
||
- Redis replay 失败：SSE 发送 `resync_required`，然后继续尝试 live stream。
|
||
- Redis subscriber loop 断开：指数退避重连；期间 SSE heartbeat 仍可维持连接，但不会有 live patch。
|
||
- 前端收到 revision 倒退或重复：忽略。
|
||
- 前端收到未知 event type：忽略并记录 debug。
|
||
|
||
## 监控与诊断
|
||
|
||
新增健康指标：
|
||
|
||
- 当前 event store mode: `redis` / `sqlite` / `degraded_sqlite`
|
||
- Redis ping latency
|
||
- latest revision
|
||
- stream length
|
||
- oldest revision
|
||
- subscriber connected
|
||
- SSE active connection count
|
||
- dropped/resync_required count
|
||
|
||
建议暴露到现有 health endpoint 或日志：
|
||
|
||
```json
|
||
{
|
||
"realtime": {
|
||
"store": "redis",
|
||
"redis_connected": true,
|
||
"stream_len": 43210,
|
||
"latest_revision": 123456,
|
||
"oldest_revision": 80200,
|
||
"subscriber_connected": true,
|
||
"sse_connections": 9
|
||
}
|
||
}
|
||
```
|
||
|
||
## 测试策略
|
||
|
||
后端：
|
||
|
||
- Redis store append 会生成全局递增 numeric revision。
|
||
- Redis replay 支持 city 过滤。
|
||
- Redis replay 支持 `since_revision`。
|
||
- Redis replay 超出 retention/limit 返回 resync required。
|
||
- Redis 不可用时，根据 env 严格失败或 fallback SQLite。
|
||
- SQLite store 现有测试保持通过。
|
||
- SSE router 在 Redis store 下仍发送 connected、replay、heartbeat。
|
||
|
||
前端：
|
||
|
||
- `use-sse-patches.ts` 保持 `revision` numeric。
|
||
- 重连 URL 保留 `since_revision`。
|
||
- 收到 `resync_required` 会触发 visible city refresh。
|
||
- 重复 revision 不重复追加图表点。
|
||
|
||
验收命令：
|
||
|
||
```powershell
|
||
python -m pytest tests/test_realtime_patch_schema.py tests/test_realtime_event_store.py tests/test_sse_replay.py
|
||
python -m pytest tests/test_redis_realtime_event_store.py
|
||
cd frontend
|
||
npm run test:business
|
||
npm run typecheck
|
||
npm run build
|
||
```
|
||
|
||
## 迁移步骤
|
||
|
||
1. 保留现有 SQLite implementation，抽出 event store factory。
|
||
2. 新增 Redis implementation 和单元测试。
|
||
3. 新增 Redis subscriber loop，但先在本地开发环境跑。
|
||
4. `/api/events` 接入 event store factory。
|
||
5. 线上安装 Redis，只监听 `127.0.0.1`。
|
||
6. 先用 `POLYWEATHER_EVENT_STORE=redis`、`POLYWEATHER_REDIS_REQUIRED=false` 灰度。
|
||
7. 观察 stream length、latest revision、SSE error、resync count。
|
||
8. 稳定后切换 `POLYWEATHER_REDIS_REQUIRED=true`。
|
||
9. 保留 SQLite fallback 代码，但生产不主动降级，避免多实例状态分裂。
|
||
|
||
## 验收标准
|
||
|
||
- 断开浏览器网络 1-5 分钟后恢复，图表自动补齐期间 patch。
|
||
- 后台挂页面再回来，图表不需要等待 full loading 才更新当前点。
|
||
- 多 worker 下，任意 worker 写入的 patch 都能推到其他 worker 的 SSE 连接。
|
||
- `/api/events?cities=taipei,shanghai&since_revision=...` 只 replay 指定城市。
|
||
- Redis 重启后，前端收到 `resync_required` 并 HTTP 重建画面。
|
||
- 前端无需知道 Redis 存在。
|
||
- 现有 `city_observation_patch.v1` schema 不破坏。
|
||
|
||
## 结论
|
||
|
||
这是一种“一步到位但不过度设计”的方案：
|
||
|
||
- 一步到位的是事件契约、replay、跨实例 fanout 和生产部署边界。
|
||
- 不过度设计的是不用 Kafka、不做永久行情库、不重写前端协议。
|
||
|
||
对当前产品最关键的是：图表像股票行情一样无痛刷新，断线可补，后台恢复可追上，多实例以后不用再推倒实时层。
|