最近在调一个设备监控平台的实时数据通道，前端大屏要求毫秒级看到几万个监测点的电压电流变化。最初方案很朴素：前端每两秒轮询一次REST接口，后端再到TDengine里查一次最新数据。单用户看还好，多用户一上来就顶不住了，数据库压力大、告警延迟肉眼可见。后来把整个链路改成了基于WebSocket的时序数据库数据订阅模式，才算把实时性真正做起来。这篇文章就把这次改造的完整思路、实现细节、实测数据和踩坑经过一起写出来，给同样被HTTP轮询困扰的朋友做个参考。

1. 为什么轮询会成为瓶颈：从表面延迟到深层资源消耗

1.1 短轮询的三层固定成本

先说清楚HTTP短轮询为什么在时序数据场景里这么难受。它不只是一个“慢”的问题，而是每一层都有固定开销，叠加起来就很可怕。

第一层是连接成本。HTTP请求走的是“请求-响应”模型，每次轮询都要重新建立TCP连接，如果是HTTPS还要多做一次或两次TLS握手。在局域网里这个成本也许可以忽略，但到了线上环境，客户端跨公网访问，握手一趟下来几十毫秒到几百毫秒就没了。即便用Keep-Alive复用连接，请求头和响应头的解析开销也依然存在。

第二层是请求内容本身的成本。哪怕你只想问“有没有新数据”，也要老老实实带上一整套HTTP协议头，Cookie、User-Agent、Accept这些字段一样都省不了。对于一个变化频率可能只有几秒一次的时序数据来说，大量请求其实是在做无用功。

第三层是数据库查询成本。这是最容易被低估的。以前端两秒轮询一次为例，后端每次都要去TDengine里执行一条聚合查询，比如查最近一分钟内所有设备的电压最大值。这种查询看起来简单，但涉及扫描、排序、聚合，在数据量大时并不便宜。更麻烦的是，轮询间隔一缩短，查询次数就成倍上涨，数据库负载直接跟着起飞。

1.2 实时性本身是不可调的

短轮询还有一个结构性难题：轮询间隔和实时性永远在打架。

间隔设得长，服务器压力小，但数据从写入到前端看到，最多可能要等一个完整间隔。假设每两秒轮询一次，一条记录在第0.1秒写入，下一次轮询在第2秒才发起，用户看到的延迟就是1.9秒。如果中间再叠加HTTP建连时间和查询时间，延迟可能到两三秒。对于监控告警这种场景，两三秒的延迟往往已经足够让工单被用户投诉。

间隔设得短，延迟确实能降下来，但代价是更高的请求频次。从两秒改到一秒，QPS直接翻倍；想做到500毫秒以内，数据库和后台都得硬扛。可真正需要实时推送的数据，变化可能只发生在其中的某几秒，剩下90%的轮询请求全是空转。这个矛盾用HTTP是解不开的，它天生就不是为“服务端主动推送”设计的。

顺带说一下长轮询和SSE。长轮询是把请求挂住，等到有新数据才返回，能在一定程度上降低空转比例，但服务端挂着一堆未完成的请求，连接管理复杂度上来了，一旦中间网络抖动，恢复成本很高。SSE只支持单向服务端推送，而且名字里就带着“Server-Sent Events”，是个半标准的东西，在代理和浏览器兼容方面坑不少。真正能同时解决“长连接”“低开销”“双向通信”这三个问题的，还是WebSocket。

2. 重新认识TDengine订阅：不是消息队列，但非常适合增量消费

2.1 TDengine订阅模型：游标加SQL

在动手之前，我对TDengine的订阅功能也有过误解，一直以为它是像Kafka那样的消息队列。用了一段时间才搞明白，它的订阅模型本质上还是“SQL查询”，只是加了一个游标机制：你用一条SELECT语句定义“想订阅什么”，数据库端记录一个消费进度，订阅连接每次去拉取游标之后新写入的数据。

举个例子，监控平台有张超级表meters，每个设备一条子表，字段包括ts、device_id、voltage、current。我想订阅所有电压超过380V的记录，就可以写这样一条订阅SQL：

select ts, device_id, voltage from meters where voltage > 380

数据库会把这个查询当作一个持续任务，每次执行时只返回从上次游标位置以来的新增数据。这个设计非常巧妙——它绕开了“每次查询全表再筛选”的笨办法，把增量读取的成本降到了一个很低的水平。

这和Kafka式的消息队列有几个本质区别，我用一张表来梳理：

理解了这个差异，就不会把订阅当成一个万能的消息通道。它最适合的场景，恰恰就是“数据已经在库里、我需要持续拿到新写入的结果”这种典型的时序数据处理场景。

2.2 从HTTP REST到WebSocket连接

TDengine本身对外提供RESTful接口，默认走HTTP。用HTTP做查询，好处是标准、简单，任何语言都能对接，但坏处就是我前面分析的那套轮询成本一样不少。尤其是监控大屏这类需要频繁交互的应用，每个操作都走一次HTTP，连接开销会被放大到很难看。

好在TDengine从一版适配器开始也支持WebSocket协议连接，也就是说，客户端可以建立一条WebSocket长连接，在这条连接上反复执行SQL或订阅操作。这相当于把“每次查询都重新握手”变成了“一个连接吃到底”，协议开销大幅下降。

我在改造时做的第一件事，就是把后端查询数据库的方式从HTTP REST换成WebSocket。这一步改动其实不大，连接管理轻了，后续做订阅推送也就顺理成章了。具体版本功能有差异，但总体思路在较新的版本里都成立。

2.3 一个具体的订阅实现示例

下面用Python给出一个订阅的示意代码，重点在于理解思路，具体API签名以你使用的连接器版本为准：

import taos import threading import time def on_data(result): # result 是本次拉取到的增量数据集 rows = result.fetch_all() for row in rows: # 这里把每条增量数据交给下游模块处理 publish_to_ws(row) def publish_to_ws(row): # 这里通过 WebSocket 推送到在线客户端 pass conn = taos.connect( host="127.0.0.1", user="root", password="taosdata", database="power" ) conn.subscribe( topic="sub_voltage_warn", sql="select ts, device_id, voltage from meters where voltage > 380", interval=500, # 每500毫秒拉取一次增量 handler=on_data ) # 保持主线程不退出 try: while True: time.sleep(1) except KeyboardInterrupt: conn.close()

代码里我特意把interval设成了500毫秒，这是当时反复调出来的一个折中值。太短会增加数据库端拉取压力，但收益不明显；太长又会把延迟抬上去。这个值在后面的压测里还会再提到。

关键点是on_data这个回调是在连接器的内部线程里被调用的，它和WebSocket服务端所在的线程不是同一个。实际实现时，我会把增量数据塞进一个队列，由事件循环统一消费，避免跨线程操作WebSocket连接。这个细节很多人第一次写会踩坑，后面专门讲。

3. WebSocket订阅服务实现细节：从订阅到推送的完整链路

3.1 架构整体一览

完整的链路是这样的：

• 采集端设备持续向TDengine写入时序数据；
• 一个订阅服务通过TDengine连接器建立订阅，以一定时间粒度拉取增量数据；
• 订阅服务将增量数据转换为统一的JSON消息结构；
• 通过WebSocket长连接把消息推送给所有在线的客户端；
• 前端收到消息后直接更新界面，不再主动轮询。

消息格式我个人习惯设计成下面这种:

{ "type": "data", "topic": "voltage_warn", "data": { "ts": "2024-06-01 12:00:00.123", "device_id": "dev-1024", "voltage": 398.5 }, "seq": 10086 }

seq字段非常重要，它是消息序号，客户端拿它做去重和断线补偿。没有这个字段，重连之后补发消息时会分不清哪些是重复的，后端也不好实现“只补最近N条”的逻辑。

3.2 服务端实现：用Python做一个轻量推送网关

我用的是一套非常轻量级的组合：Python的websockets库做WebSocket服务端，TDengine官方连接器做订阅。整个网关代码量不大，但逻辑很完整。下面是我最初跑通Demo版本的核心代码：

import asyncio import json import websockets connected_clients = set() message_queue = asyncio.Queue(maxsize=10000) # ---------- TDengine 订阅回调 ---------- def td_subscribe_callback(result): rows = result.fetch_all() for row in rows: payload = { "ts": row[0], "device_id": row[1], "voltage": row[2], } # 注意：这里运行在数据库线程里，不能直接操作 asyncio 对象 asyncio.get_event_loop().call_soon_threadsafe( lambda p=payload: message_queue.put_nowait(p) ) # ---------- WebSocket 广播协程 ---------- async def broadcast_worker(): while True: payload = await message_queue.get() msg = json.dumps({ "type": "data", "topic": "voltage_warn", "data": payload }, ensure_ascii=False, default=str) if connected_clients: await asyncio.gather( *[client.send(msg) for client in connected_clients], return_exceptions=True ) # ---------- WebSocket 客户端处理 ---------- async def handle_client(websocket): # 上线时加入集合，下线时移除 await websocket.send("connected") connected_clients.add(websocket) try: # 用 wait_closed 等待客户端断开，不需要处理复杂消息 await websocket.wait_closed() finally: connected_clients.discard(websocket) # ---------- 主函数 ---------- async def main(): async with websockets.serve(handle_client, "0.0.0.0", 8765): await broadcast_worker() if __name__ == "__main__": asyncio.run(main())

这段代码最核心的心得是把“订阅拉取”和“广播”解耦：订阅回调只负责把数据塞进队列，广播协程统一从队列取消息再发给所有客户端。这样即使某个客户端发送很慢，也不会卡住数据库订阅线程。队列容量限制是防缓冲堆积用的，真堆满了就说明下游消费能力不够，该去查客户端了。

call_soon_threadsafe的作用是把数据库线程里的数据安全地转交给asyncio事件循环。这一步如果不做，直接在回调里调用asyncio对象的方法，很容易遇到“事件循环未运行在当前线程”的报错或偶发崩溃。这是多线程+asyncio组合的典型坑位。

3.3 客户端接入与断线重连

服务端搞定之后，前端接入就比较简单了。但简单归简单，还是有几个点必须处理好，首当其冲的是重连策略。

直接在浏览器里用原生的WebSocket对象：

let ws; let retryCount = 0; let lastSeq = 0; function connect() { ws = new WebSocket(`wss://your-domain.example/ws?token=${authToken}`); ws.onopen = () => { retryCount = 0; // 连接建立后，把本地记录的最后一条消息序号发给服务端 ws.send(JSON.stringify({ type: "sync", lastSeq })); }; ws.onmessage = (event) => { const msg = JSON.parse(event.data); if (msg.type === "ping") { ws.send("pong"); return; } if (msg.seq <= lastSeq) { return; // 去重 } lastSeq = msg.seq; renderTick(msg.data); }; ws.onclose = () => { // 指数退避重连，最大间隔30秒 const delay = Math.min(30000, 1000 * Math.pow(2, retryCount)); retryCount++; setTimeout(connect, delay); }; ws.onerror = () => { ws.close(); }; } connect();

指数退避是一定要做的。我见过不少人图省事，断线后固定一秒重连，结果服务端发布时所有客户端同时断线、同时蜂拥重连，把网关打崩。指数退避加上一个随机抖动，就能让重连请求错开，这种细节在故障恢复时非常关键。

另一个容易忽略的点是同步消息序号。重连之后，客户端把lastSeq发给服务端，服务端从消息历史里把序号大于这个值的消息补发过来。这相当于一个轻量级的断点续传。当然，前提是服务端得保留最近的消息历史。我当时用一个环形缓冲存了最新2000条消息，足够覆盖绝大多数断线场景。

4. 实测对比：延迟、连接开销和查询负载的变化

4.1 测试场景设置

为了直观说明这个改造到底值不值，我搭了一套对比测试。场景是这样的：模拟1000台设备，每5秒向TDengine写入一条状态数据，同时有100个在线客户端在查看实时界面。

协议对比分三组：

• 组A：前端每2秒轮询一次REST接口，后端每次查询最近1分钟所有设备的电压平均和最大值；
• 组B：前端每1秒轮询一次REST接口，查询逻辑同上；
• 组C：后端使用TDengine订阅，每500毫秒拉取一次增量，通过WebSocket推送。

压测工具用的是日常用的那套自研脚本，模拟客户端并发连接、统计消息接收延迟。每轮跑10分钟，取P50和P95延迟。

4.2 数据解读：延迟、QPS与数据库查询次数

三组数据的大致情况如下：

这里有几个数字值得解释一下。

延迟方面：组A的P50是1200ms，意味着半数数据的显示延迟超过一秒。组B翻倍请求后延迟降到700ms左右，但代价是数据库查询次数从50次/秒涨到100次/秒。组C把订阅粒度设为500ms后，P95能做到650ms，P50只有450ms，这个延迟已经接近实时。

数据库查询次数这个指标最容易被人忽略。组A和组B每次轮询都要执行一次带GROUP BY的聚合查询，即便窗口内没有新数据也要查。组C的订阅连接每500毫秒拉一次增量，但只查询游标之后的新数据，复杂度完全不一样。最终数据库侧的负载差距是几十倍，不是几倍。这在数据规模更大时会更加明显。

连接数方面：组A的100个客户端每2秒各发起一次请求，网关每秒要处理50次建连和握手；组C的100个客户端全程只有100条长连接。连接资源占用完全不是一个量级。

4.3 结果解读与进一步优化

第一轮测试已经足够说明问题。随后我又把订阅粒度从500毫秒调到100毫秒试了一次，P95延迟降到了大概300毫秒，数据库查询次数从2次/秒涨到10次/秒，但相比HTTP轮询的100次/秒依然是质的改善。这个结果让我对订阅模式的定位有了更明确的认识：它不是包治百病，但很擅长把“定期拉取”变成“按需推送”。

优化空间还有一块：把订阅SQL改成只返回真正需要的字段，不要select *。有一次我在订阅SQL里多带了一个文本备注字段，消息体积大了三倍，广播时带宽和客户端解析时间都有明显上升。时序数据往往是数值型为主，字段越少，WebSocket传输效率越高。

5. 落地过程中最需要小心的几个坑

5.1 别把订阅服务当无限容量的消息队列

这是我在整个改造里踩过最深的一个坑。一开始我想得比较简单：既然订阅能拉增量，WebSocket能推送，那直接把订阅回调里拿到的数据全都塞进队列再广播，不就成了一个实时消息平台吗？结果上线没多久就出问题了。

问题出在消费速度不匹配。前端有100个客户端，其中几个网络状况很差或页面切到了后台标签页，WebSocket发送速度变得极慢。服务端的广播协程一次send可能要等好几秒才返回，而订阅线程还在不断往队列里丢新数据。队列从几百条迅速涨到几千条，内存占用飙升，最终把进程拖崩。

后来我把方案改成了：每个客户端一个独立的发送队列，而不是全局共享一个队列；如果某个客户端队列超过阈值，直接把它标记为“慢消费者”，服务端主动断开连接，让客户端重连后走补发逻辑。这样就把单点瓶颈变成了可管理的背压机制。

处理背压的经验：慢客户端不能让所有正常客户端陪它一起等。宁可断开它，也不能拖垮整个网关。

5.2 不要在订阅SQL里做重聚合

订阅SQL写得越简单越好。我当时有一版订阅语句带了GROUP BY device_id和几个聚合函数，想着直接在订阅阶段就把统计数据算好，推给前端就能直接展示。运行了没多久就发现，订阅拉取的间隔稍长一点，增量窗口内没有新数据，聚合结果根本不会变化；而且数据库端要维护复杂的状态，消耗的资源反而比普通查询多。

正确的拆分方式是：TDengine订阅只负责把原始增量数据拿出来，统计聚合放到前端或查询接口做。客户端收到的消息越“原始”，后续扩展越灵活。我之前把订阅里查出来的字段先做了一层聚合再推送，结果产品经理过来说想换个维度统计，我只好重新改订阅SQL重新推消息，白白折腾了一整天。

5.3 连接管理和token过期的隐性故障

WebSocket长连接一旦建立，就可能长时间不释放。这对用户来说是好事，但对服务端来说意味着连接状态必须被妥善管理。

第一个坑是代理层的空闲超时。很多云负载均衡默认对空闲连接有个超时时间，比如60秒或300秒。如果客户端半天不发消息，代理就会把连接悄悄断开，而客户端未必能立刻感知。解决方案很常规：WebSocket服务端定期发ping，客户端回pong，既保活，也顺带验证链路通畅。

第二个坑是token过期。HTTP轮询时每次请求都带token，过期了客户端重新登录即可。但WebSocket连接是常驻的，token如果有效期为半小时，连接建立后半小时一到，服务端再推送就会失败。我当时的处理是：在ping帧里带上token的过期时间，服务端发现即将过期时主动断开连接，客户端重连时带上新token。或者干脆把长连接的有效期设计成和token一致，过期就重新握手。

5.4 订阅游标丢失与补偿

TDengine订阅的这个游标，是在连接器内部维护的。如果订阅服务进程异常退出或者网络断开，重新订阅时是从头开始还是从上次位置继续，不同版本的行为可能不完全一样。

为避免数据丢失，我做了个妥协式方案：订阅服务重启后，先主动从now - 5秒的时间点开始拉取一次，把最近5秒内的数据都读一遍，再进入正常增量模式。这5秒的窗口会有少量重复数据，所以客户端一定要做去重。去重用什么？就是我前面提到的seq消息序号，时序数据的ts字段也可以作为自然去重键，但ts精度不够时还是seq更可靠。

从这个坑能看出，订阅模式虽然比手动管理“上次查询时间”要省事，但它不是说完全不用管数据连续性。任何分布式系统里，断点和补偿都是绕不开的功课。

6. 从Demo走向生产：稳定性、可观测性与扩展建议

6.1 稳定性建设：确认机制与死信处理

上面的方案跑通之后，我们已经能做到“数据一变，前端就变”。但要真正上生产，还得回答一个问题：如果推送出去的消息客户端没收到，或者收到了没处理成功，怎么办？

我加了一个轻量级的确认机制。客户端每收到一批消息，就向服务器发一个ack，把最后一条消息的seq带上。服务端维护每个客户端最近的lastAckSeq，如果某个客户端在N秒内没有推进lastAckSeq，就把它的连接断开，让重连逻辑去补数据。这样做的好处是客户端崩溃与否不会被遗漏，坏处是服务端多了不少状态。

那些多次重连都补不上的消息，最终落到一张死信表里，定时任务每天扫一遍，发现异常直接告警。这个设计看着重，但时序数据监控的可靠性往往比想象中的更重要——数据没推上去，业务侧可能完全无法察觉，等用户自己发现问题时，影响已经扩散了。

6.2 可观测性：发布订阅链路如何监控

改造之前，系统是HTTP轮询模型，日志里全是请求记录，出了问题用链路追踪工具一把梭。改成WebSocket订阅后，没有“请求-响应”这个天然监控点，排查问题难度一下子上来了。我花了不少时间才把监控体系补齐。

现在保留这样几类指标：

• 订阅拉取延迟：从数据写入TDengine到订阅回调收到数据的时间差；
• 广播耗时：消息从队列取出到所有客户端都发送完成的时间；
• 连接状态：在线连接数、最近一小时重连次数；
• 队列积压：消息队列长度变化，这是背压快溢出的前兆。

日志方面，重点记录“拉取到推送”的耗时分布，分P50/P95/P99三个档。有了这些指标，至少能在出问题时快速区分是数据库慢、订阅慢，还是WebSocket广播慢。有一次线上告警延迟突然升高，靠的就是这条指标链定位到一个慢客户端拖累了asyncio.gather的等待，促成了慢客户端断开机制的落地。

6.3 再往前走一步的扩展思路

如果后续设备量从1000涨到10万，单台订阅服务大概率扛不住。我的扩展思路是按设备分片，而不是盲目堆机器。

具体做法是：把设备ID按哈希分到多个订阅服务上，每个服务只订阅自己负责的那段设备的数据。TDengine的超级表天然支持按子表范围查询，所以每个订阅服务可以只用一条带条件的SQL定义自己的订阅边界，互不干扰。前端客户端按区域或业务线连接不同的订阅网关，数据路由交给上层的接入层处理。

如果以后还需要从平台下发控制指令给设备，WebSocket本身就是双向的，直接扩展一条command消息类型就好，不需要再造一套通道。这算是我这次改造里比较得意的地方，一条连接把上行数据和下行指令都解决了。

按我这次改造的经验，第一周先把订阅拉通，第二周补心跳和断线重连，第三周再处理背压和监控，节奏刚刚好。别一上来就想用重型消息框架，订阅模式的核心优势恰恰就在于轻——它让你用最小的改动，拿到接近实时消息队列的效果。如果你也被HTTP轮询折磨过，不妨从TDengine订阅加WebSocket这条路开始试，验证成本很低，回报却非常直接。