实时管线：TCP 输入、sequence、handler、队列和 Socket.IO

背景问题

呼吸信号是连续流。后端不能等用户刷新页面才处理，也不能让前端每次请求时再去读传感器。系统必须在后台持续接收、处理和缓存数据，并把最新结果推给所有连接的客户端。

这就带来一个设计问题：数据从 TCP 进入后，究竟应该如何分发给滤波、记录、质量分析和前端推送？如果全部写在一个 while True 里，短期能跑，长期会变成难以测试的实时泥团。

设计方案对比

方案	优点	问题	适合场景
一个大循环处理全部逻辑	代码路径短，调试直观	滤波、记录、发送互相耦合，任何一步变慢都会影响采集	小型实验脚本
直接函数调用链	比大循环清晰，调用关系明确	下游一多，调用顺序和异常处理变复杂	模块数量少的服务
事件总线	解耦强，扩展性好	对当前项目偏重，排查顺序成本高	大型多模块系统
handler + asyncio 队列	边界清楚，复杂度可控，可批量推送	需要维护队列上限和 dropped 统计	当前采用

RespiraScope 选择 handler 和队列，是因为它需要可解释的实时链路，而不是追求最高抽象。DataReceiver 分发 raw 点，SignalProcessor 分发 filtered 点，DataQueueManager 把数据放进不同队列，DataSender 固定节奏批量推送。

当前方案

DataReceiver 连接 AsyncSocketClient，每读到一个传感器值就调用 data_handler()。这个函数做两件关键事：更新最近数据时间，给点分配递增 sequence_number，然后把 (sequence, value) 发给所有 data handler。

BreathProcessSystem._setup_connections() 把后续关系接好：

• DataReceiver 输出给 DataQueueManager、RecordManager、SignalQualityAnalyzer、SignalProcessor。
• SignalProcessor 输出给 DataQueueManager、RecordManager、PeakValleyDetector。
• PeakValleyDetector 输出给 PeakValleyHandler 和 RecordManager。
• PeakValleyHandler 把 peak、valley、metrics 写入队列。
• DataSender 周期性读取队列并通过 Socket.IO 发送。

为什么这样选

sequence 是这条管线的核心。时间戳能描述“什么时候收到”，但不能稳定描述“第几个采样点”。滤波、峰谷、记录、scan 都需要一个可排序、可对齐、可判断 gap 的主键。sequence 正好满足这个需求。

队列的价值在于把采集和发送解耦。传感器输入是小步快跑，浏览器推送适合批量。DataSender 每 40ms 发送一批 raw、filtered、peak、valley、metrics，减少 Socket.IO 消息数量，也让前端图表更新更平稳。

最近快照也是一个实际设计。DataQueueManager 除了实时队列，还维护 recent_raw、recent_filtered、recent_peaks、recent_valleys 等 deque。新浏览器连接后，socket_io_service 会调用 snapshot_provider 补发最近数据。这样用户刷新页面不会看到一张空图，而是能立即获得上下文。

代价和边界

handler 设计要求每个模块的输入接口保持简单。比如 raw handler 约定是 handle_data(sequence_number, sensor_val)，filtered handler 约定是 handle_filtered_data(sequences, values)。如果以后加入复杂事件对象，可能需要升级成显式事件类型或消息总线。

队列也不是无限缓冲。DataQueueManager 给每个队列设置 maxsize，并记录 dropped count。这个选择很明确：实时展示宁可丢旧数据，也不能让内存无限增长。它适合实时监测，不适合要求每个采样点都必须可靠入库的审计系统。

关键函数表

函数或类	作用	设计点
asyncio.open_connection()	连接真实或模拟 TCP server	后端作为 TCP client 主动读传感器
AsyncSocketClient.start_receiving()	循环读取 4 字节采样包	断开后按 reconnect_delay 重连
DataReceiver.data_handler()	给采样点编号并分发	sequence 是全链路对齐主键
asyncio.Queue(maxsize=...)	保存待推送批次	限制内存，记录 dropped
collections.deque(maxlen=...)	保存最近快照	新客户端连接后补发上下文
DataSender.start_sending()	每 40ms 批量推送	降低推送频率和前端更新压力
sio.emit("breath", msg)	Socket.IO 广播实时消息	Monitor 和外部系统可同时消费

读源码建议

先从 data_receive.py 看 raw 点如何产生，再看 data_manager.py 如何同时维护实时队列和最近快照。随后读 send_to_client.py，确认队列是如何变成 { type, data } 消息的。最后看 socket_io_service.py，理解为什么新 client 能收到最近快照。