实时数据管道

Phase 06 — Real-time · 知识点 08–09：Concurrent Connections · Real-time Data Pipeline

1. 学习目标

完成本知识点后，你应该能够：

评估并优化 WebSocket 服务的 Concurrent Connections 承载能力
设计从数据源到 Three.js 前端的 Real-time Data Pipeline
使用 goroutine、channel、Worker Pool 解耦采集、处理与推送
识别并避免 goroutine 泄漏、锁竞争、内存暴涨
实现消息合并（coalesce）与采样，降低无效推送
为 Project 03 数字孪生后端奠定管道架构基础

2. 为什么需要

当仓库内 AGV 从 10 台增至 500 台、在线 WebSocket 客户端从个位数增至数百时，「每个连接一个无限循环 + 每帧全量广播」的 naive 实现会迅速耗尽 CPU 和内存。

Concurrent Connections 不仅是「能连多少」，更是在连接数增长时如何保持延迟稳定。Real-time Data Pipeline 把「数据从哪来、如何加工、推给谁」拆成清晰阶段，便于测试、扩容和与 PostgreSQL / Redis 集成。

3. 核心概念

3.1 Concurrent Connections（高并发连接）

瓶颈	常见原因	应对
文件描述符	ulimit 过低	调高 `nofile`（Phase 08）
内存	每连接缓冲过大	限制 `SetReadLimit`，合理 channel buffer
CPU	频繁 JSON 序列化	预序列化、批量、二进制协议
锁竞争	Hub 全局锁	分 Room 锁、shard Hub
goroutine 泄漏	未 unregister	defer + context 取消

经验参考（单机、纯推送、优化后）：

4 核 8G：数千～万级 idle 连接可行
实际以压测为准（go test benchmark + wrk/自定义 WS 客户端）

3.2 Real-time Data Pipeline

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ Source   │───▶│ Ingest   │───▶│ Process  │───▶│ Dispatch │
│ 模拟器   │    │ 接收解析  │    │ 过滤聚合  │    │ Room推送 │
└──────────┘    └──────────┘    └──────────┘    └──────────┘
                                      │
                                      ▼
                               ┌──────────┐
                               │ Store    │
                               │ PG/Redis │
                               └──────────┘

阶段说明：

阶段	职责
Source	PLC、MQTT、内部模拟器、HTTP webhook
Ingest	读入原始字节，校验，转领域对象
Process	去重、告警规则、坐标变换、节流
Dispatch	按 room/device 路由到 WebSocket Hub
Store	持久化快照、历史轨迹（可选）

3.3 并发模式

// 固定 Worker 处理 ingest，避免无界 goroutine
jobs := make(chan RawEvent, 1024)
for i := 0; i < workerCount; i++ {
    go func() {
        for ev := range jobs {
            processed := process(ev)
            dispatcher.Publish(processed)
        }
    }()
}

4. 基础语法

4.1 带 Context 的连接生命周期

func (s *Server) HandleWS(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        return
    }
    ctx, cancel := context.WithCancel(r.Context())
    defer cancel()
    defer conn.Close()

    client := &Client{conn: conn, send: make(chan []byte, 256)}
    s.hub.Register(client)

    go client.writePump(ctx)
    client.readPump(ctx) // 阻塞直到断开
    s.hub.Unregister(client)
}

4.2 Coalesce（合并最新状态）

type Coalescer struct {
    mu    sync.Mutex
    latest map[string][]byte // deviceId -> last payload
    notify chan struct{}
}

func (c *Coalescer) Update(deviceID string, payload []byte) {
    c.mu.Lock()
    c.latest[deviceID] = payload
    c.mu.Unlock()
    select {
    case c.notify <- struct{}{}:
    default:
    }
}

// 单独 goroutine：每 100ms 批量 flush latest 到 Dispatch

同一设备 100ms 内多次位置更新只推送最后一次，减轻 Three.js 渲染压力。

4.3 Pipeline 与 HTTP 快照 API 协作

// GET /api/v1/warehouses/{id}/devices — 全量快照
// WS /ws — 增量 agv.position / agv.status

Pipeline 的 Process 阶段同时更新内存 Snapshot Store，HTTP 接口读同一 Store，保证快照与增量字段一致。

4.4 简单压测思路

func BenchmarkMarshalAGV(b *testing.B) {
    msg := AGVPositionMsg{Type: "agv.position", DeviceID: "AGV-001", X: 1, Y: 2}
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        json.Marshal(msg)
    }
}

配合 -race 检测 Hub 注册/注销竞态。

5. 代码解析

make(chan []byte, 256)：每客户端发送队列。无缓冲会导致 Hub 广播被单个慢客户端阻塞；过大则占用内存 = 连接数 × buffer × 消息大小。

Context 取消：HTTP 请求结束或进程 shutdown 时，cancel() 通知 writePump 退出。

Coalescer：数字孪生场景下，渲染帧率 60fps 但 AGV 定位可能 10Hz，合并可显著降带宽。

Pipeline 解耦：Ingest 阻塞不影响 Dispatch；Store 写失败不应阻塞推送（异步或降级）。

6. JavaScript / TypeScript 对比

概念	Three.js 前端	Go Pipeline
数据入口	`ws.onmessage`	Ingest channel
渲染节流	`requestAnimationFrame`	Coalesce / 采样
状态存储	Scene Graph / Store	内存 Snapshot + DB
并发	单线程 Event Loop	goroutine + channel

前端用 rAF 合并渲染；后端用 Coalescer 合并推送——同一思想，不同层级。

7. 常见错误

错误 1：每事件一个 goroutine

AGV 1000 台 × 10Hz = 每秒 10000 goroutine 创建，应使用 Worker Pool。

错误 2：Hub Broadcast 全量序列化 N 次

同一消息应序列化一次，再 broadcast 字节切片。

错误 3：Pipeline 反向压力缺失

Ingest 无界 channel 在 Source 爆发时 OOM，需 select default 丢弃或阻塞策略。

错误 4：忽略 graceful shutdown

SIGTERM 时应停止接受新连接、等待 Hub 排空、关闭 DB。

错误 5：快照与增量字段不一致

Three.js 解析失败或显示跳动，需共享 struct 定义或 OpenAPI 契约。

8. 实际应用

Project 03 — 数字孪生完整管道

AGV Simulator (Go ticker)
    → Ingest: 解析 DeviceEvent
    → Process: 边界检测、低电量告警
    → Store: PostgreSQL 写历史 + Redis 缓存最新状态
    → Dispatch: room[warehouseId].Broadcast
    → Three.js: 更新 AGV Mesh + 告警面板

Concurrent Connections 目标（练习/项目）：

100 并发 WS 客户端稳定推送
50 台 AGV 10Hz 更新，服务端 CPU < 50%（开发机参考）

监控指标：

当前连接数、每 room 连接数
推送 QPS、丢弃/合并计数
goroutine 数量（runtime.NumGoroutine()）

9. 深入理解

9.1 C10K 问题

Go 的 goroutine + epoll 使 C10K 较易达成，但应用层设计仍是瓶颈。

9.2 水平扩展

多实例 + Redis Pub/Sub：每个实例维护本机 WS 连接，Dispatch 发布到 Redis，各实例订阅后广播给本地 Hub。

9.3 背压与 SLA

定义「可丢弃」与「必达」消息：位置可合并；告警必达且需 ACK（可选扩展）。

10. 练习

详细练习见 exercises/phase-06-realtime/03-realtime-pipeline.md。

Level 1 — 基础

练习 1.1：统计当前 WebSocket 连接数，暴露 GET /metrics/connections。

练习 1.2：实现 3 个 Worker 的 Ingest channel 处理。

Level 2 — 应用

练习 2.1：Coalescer：100ms 窗口合并同 deviceId 消息。

练习 2.2：序列化一次、广播多次 refactor。

Level 3 — 综合

练习 3.1：模拟 50 AGV + 20 客户端压测脚本，记录 CPU/内存。

练习 3.2：Pipeline + HTTP 快照 API 字段一致性测试。

Level 4 — 项目实践

练习 4.1：Project 03 Phase 06 里程碑 — 完整实时管道 + Three.js 联调。

11. 学习检查

影响 WebSocket 并发连接数的主要因素有哪些？
Pipeline 各阶段职责如何划分？
Coalesce 适用哪些消息，不适用哪些？
如何实现 graceful shutdown？
多实例部署时 Dispatch 如何扩展？

12. 下一步

已完成	下一知识点	关系
Concurrent Connections · Real-time Data Pipeline	Phase 07 — Project Architecture	实时功能完成 → 工程化重构

Phase 06 完成后，你将具备 Project 02 / Project 03 的实时通信能力。进入 Phase 07，把实验性代码整理为可测试、可容器化的工程项目。

学习导航

上一篇：WebSocket 可靠性与广播 · 对应练习 · 下一篇：项目架构与工程基础

实时数据管道 ​

1. 学习目标 ​

2. 为什么需要 ​

3. 核心概念 ​

3.1 Concurrent Connections（高并发连接） ​

3.2 Real-time Data Pipeline ​

3.3 并发模式 ​

4. 基础语法 ​

4.1 带 Context 的连接生命周期 ​

4.2 Coalesce（合并最新状态） ​

4.3 Pipeline 与 HTTP 快照 API 协作 ​

4.4 简单压测思路 ​

5. 代码解析 ​

6. JavaScript / TypeScript 对比 ​

7. 常见错误 ​

错误 1：每事件一个 goroutine ​

错误 2：Hub Broadcast 全量序列化 N 次 ​

错误 3：Pipeline 反向压力缺失 ​

错误 4：忽略 graceful shutdown ​

错误 5：快照与增量字段不一致 ​

8. 实际应用 ​

Project 03 — 数字孪生完整管道 ​

9. 深入理解 ​

9.1 C10K 问题 ​

9.2 水平扩展 ​

9.3 背压与 SLA ​

10. 练习 ​

Level 1 — 基础 ​

Level 2 — 应用 ​

Level 3 — 综合 ​

Level 4 — 项目实践 ​

11. 学习检查 ​

12. 下一步 ​

学习导航 ​