ThingsBoard集成TDengine：从规则引擎直写到Kafka管道实践-平芜编程栈

你有没有遇到过这种局面：ThingsBoard控制台上设备数据刷得飞快，但查询历史曲线时页面转圈，数据库磁盘三天涨了一大截，PostgreSQL的CPU常年飘在70%以上。我最初接ThingsBoard的时候，觉得它自带的那套存储方案够用，设备量小，一切安好；等设备从几千台扩到几万台，消息频率一上来，问题堆着来找你。后面我把设备遥测数据从ThingsBoard解耦出去，落到了TDengine这套时序数据库里，才算把这个结解开。

这篇文章不是讲“把TDengine装好”这种入门教程，而是讲怎么真正把TDengine集成进ThingsBoard的规则链路里：用规则引擎直接把遥测写入TDengine、用Kafka管道做异步削峰、在RPC下发之前先查TDengine的状态，以及我实际部署时踩过的那些坑，包括那个让人头疼的0x83a: query denied by license: external query is restricted错误。适合正在做ThingsBoard二次开发、或者打算把时序数据单独剥离出来的团队参考。

1. 为什么我会选择“ThingsBoard + TDengine”这套组合

1.1 ThingsBoard自带存储的边界在哪里

ThingsBoard默认的存储方案，CE版本基本都是PostgreSQL，有的部署会叠加Cassandra做混合存储。PostgreSQL本身没问题，但它不是为“海量时序写入”设计的。ThingsBoard每秒钟要写入大量设备属性、遥测数据，还要频繁做历史查询，PostgreSQL的MVCC机制、行式存储结构，在这种场景下会逐渐拖垮性能。

我遇到过几个具体的现象：

设备上报频率是10秒一条，6万个设备，每天产生大概5亿条遥测，PostgreSQL单表从几百万行开始，历史查询就明显变慢。
页面打开仪表盘，选择最近24小时曲线，SQL要扫近千万行，聚合计算全部压在数据库上。
磁盘占用快速膨胀，因为PostgreSQL对时序数据没有专门的压缩策略，清理过期数据也要额外写定时任务。

如果只用ThingsBoard自带存储，后期要不停地加索引、做分区、甚至上Cassandra集群。问题是Cassandra那套运维成本并不低，查询能力也不像时序数据库那么贴合物联网场景。

1.2 TDengine能补齐哪块短板

TDengine把遥测数据按时间维度和设备标签做了专门的处理，有几件事天生就比PostgreSQL合适：

一张超级表对应一类设备，子表按设备ID建立，写入时不需要人为管理分表。
数据按时间分区，老数据到时间自动滚动，删除历史数据几乎零成本。
列式存储加压缩，磁盘占用通常能省不少。
提供了LAST_ROW、AVG、COUNT、时间窗口聚合这类时序语义，查询语句比SQL从PostgreSQL里硬算要简洁得多。

最关键的是，TDengine支持标准SQL，还有REST接口，这意味着ThingsBoard规则引擎不需要写很复杂的自定义插件，用现成的HTTP节点就能对接。集成成本比想象中低。

1.3 集成前先想清楚：是同步写还是异步写

不要一上来就开始配置，先想清楚数据链路要同步还是异步。

同步写的意思是：ThingsBoard收到设备上报后，在规则链里当场调用TDengine的REST接口写入。优点是简单、实时、数据不经过中间件，缺点也很明显：如果TDengine抖动或者网络延迟，会让ThingsBoard的消息处理链路变慢，甚至积压。

异步写的意思是：ThingsBoard规则引擎先把遥测丢到Kafka，再由TDengine的taosAdapter消费后写入。优点是削峰填谷，ThingsBoard的响应速度不受TDengine影响；缺点是多了一套Kafka组件，数据链路更长，定位问题也麻烦一点。

我当时的判断标准很简单：

如果设备量不大，消息频率在每秒几百条以内，同步直写完全够用。
如果消息频率高、峰值明显，或者团队对Kafka本身熟悉，优先走异步管道。

后面两章我会把这两条路线都展开，并给出我的选型建议。

2. 两条主流的集成路线，我最终选了哪条

这一章先把两条路线的技术骨架讲清楚，再给一个对比，方便你按自己的场景做决定。

2.1 路线A：规则引擎REST API直写TDengine

ThingsBoard规则引擎里有一个现成的节点叫REST API Call，可以发起HTTP请求。TDengine的taosAdapter默认监听6041端口，对外提供REST SQL接口。所以我们可以在规则链里拿到设备遥测后，拼一条INSERT语句，发给TDengine执行。

这条链路的流程是：设备上报 -> TB消息进入规则链 -> 通过REST API Call节点POST到http://tdengine-host:6041/rest/sql-> TDengine执行SQL完成写入。

优点是组件最少，不需要额外部署Kafka，规则链里一个节点就能搞定。缺点是每条消息独立写一次，吞吐量取决于HTTP请求的耗时。如果要提升性能，得在TDengine侧做批量写入，或者在规则链里做消息聚合，这需要额外开发。

2.2 路线B：Kafka管道 + taosAdapter自动落库

如果走异步，更结构化的方案是：ThingsBoard规则引擎里用Kafka节点把遥测发布到指定的topic，TDengine这边用taosAdapter或者官方Kafka Connector订阅这个topic，解析JSON后自动写入超级表。

ThingsBoard的Kafka节点本身支持指定topic和消息体，消息体可以直接放原始JSON。TDengine侧要做的，是配置好topic对应到哪张超级表、JSON里的字段和表字段如何映射。这个方案的好处是解耦和削峰，消息峰值再高，中间有Kafka挡一层，TDengine按自己的节奏消费写入。

缺点是需要多维护一套Kafka，如果你的团队没有额外精力运维中间件，这条路的成本会超过直写。另外，taosAdapter对JSON的解析策略比较直接，字段名最好和超级表字段保持一致，如果设备上报的JSON嵌套层次很多，还得在规则链里先用脚本展开。

2.3 两条路线的对比与选型建议

对比项	规则引擎直写	Kafka管道写
组件数量	少，只需要TB和TDengine	多，需要Kafka
实时性	高，消息处理完立即写入	略延迟，取决于Kafka消费速度
吞吐上限	受单条HTTP写入限制	高，能扛突发流量
故障影响	TDengine异常会影响TB消息处理	TDengine异常不阻塞TB，消息先落在Kafka
运维复杂度	低	高
适合场景	中小规模、开发团队小	设备量大、峰值明显、有Kafka经验

从我的实际经验来看，如果业务还处于快速迭代期，先走路线A是最省事的。你不需要引入新中间件，就能把TDengine用起来。等确实遇到写入瓶颈，再引入Kafka管道做改造，规则链节点的复用度也高。

我当时的项目最终选了“混合”方案：一般遥测数据通过规则引擎直写TDengine，重要事件和突发性强的数据另外扔到Kafka，再由taosAdapter消费。这样既保证关键链路的及时性，又不至于让直写把TDengine压垮。

3. 环境准备：版本搭配、驱动和需要提前避开的License坑

3.1 版本组合建议

TDengine目前常见的是2.x和3.x两个大版本。2.x生态稳定，资料多；3.x在存算分离、流式计算上有不少改进。从ThingsBoard集成角度，两者差别不大，taosAdapter和REST接口都是标配。

ThingsBoard CE建议用3.4以上版本，规则引擎的REST API Call节点对模板变量的支持更完整。如果用的是2.x老版本，功能也够，但节点配置界面略旧。

这里提醒一点：不要只看ThingsBoard版本，还要确认TDengine的taosAdapter是不是启动状态。很多人在服务器上装了TDengine，发现6041端口不通，基本都是taosAdapter没起来。

3.2 开通并验证REST API

TDengine的REST接口默认由taosAdapter提供服务。检查方式很简单：

curl -u root:taosdata -d "select server_version()" http://localhost:6041/rest/sql

返回结果如果是类似：

{ "code": 0, "column_meta": [["server_version()", "VARCHAR", 64]], "data": [["3.0.7.0"]], "rows": 1 }

说明REST接口可用。注意root:taosdata是默认账号密码，生产环境务必改掉，并通过防火墙限制6041端口只允许ThingsBoard服务器访问。

3.3 常见错误0x83a的排查思路

网上关于“tdengine error (0x83a): query denied by license: external query is restricted”的讨论不少，我也遇到过。这个错误从字面看是授权层面拒绝了外部查询，实际排查时不能只看错误码，要按链路一步步定位。

我建议按这个顺序排查：

第一步，看请求是从哪里发起的。如果你是用taosCLI、JDBC、REST API分别执行同一条SQL，只有某一个渠道报0x83a，那问题基本出在渠道对应的服务或配置上。
第二步，看taosAdapter日志。日志里会记录请求来源和具体拒绝原因，能帮你判断是taosAdapter拒绝，还是taosd返回的license限制。
第三步，用最小化SQL复现。有些时候不是所有查询都被限制，而是某个特定SQL触发了“外部查询”的限制，比如跨节点查询、UDF查询等。把SQL简化后逐个排除。
第四步，检查当前部署的TDengine版本是社区版还是企业版，以及license授权里是否包含外部查询能力。如果确实是授权限制，联系官方调整，或者改用taosAdapter的WebSocket/原生连接方式作为替代。

这个错误很容易让人误以为是自己配置错了，其实大部分时候是版本授权和服务网格问题。只要日志里能看到请求到达，方向就对了，剩下的是规则层面的排查。

4. 实操：通过规则引擎把设备遥测写入TDengine

这一章是整篇文章最核心的部分。我按照我自己的配置过程，从建表到规则链节点，一步步写出来。

4.1 先建好库、超级表和子表

不要在ThingsBoard里直接配置完规则链再去建表，SQL可以提前在TDengine里准备好。TDengine的建模思路是：一个设备对应一张子表，一类设备对应一张超级表，子表动态创建，写入时带上标签即可。

我通常先建一个独立的数据库，默认保留90天数据：

CREATE DATABASE IF NOT EXISTS thingsboard KEEP 90 DURATION 10 BUFFER 32 WAL_LEVEL 1; USE thingsboard;

接着创建一张超级表，用来存所有设备上报的温度湿度数据：

CREATE STABLE IF NOT EXISTS device_telemetry ( ts TIMESTAMP, temperature FLOAT, humidity FLOAT, device_name NCHAR(64) ) TAGS ( device_id NCHAR(64) );

注意TDengine要求第一列必须是时间戳，标签列不能和普通列重名。设备名放在普通列还是标签列，取决于你的查询方式。如果经常按设备名过滤，放在标签列更合适；如果只是作为展示字段，放在普通列也无所谓。

4.2 配置REST API Call节点

在ThingsBoard的规则链里，从库里拖一个REST API Call节点到画布，连在Post telemetry消息处理链路上。节点配置如下：

Request URL：http://tdengine-host:6041/rest/sql
Request Method：POST
Headers：添加Authorization，值为Basic加空格加base64(root:taosdata)，也可以直接把账号密码写成root:taosdata由节点自动处理
Body：填要执行的SQL

HTTP请求头里还要加Content-Type: text/plain，不然TDengine可能不认这个请求体。

4.3 用Script节点拼SQL，避免模板变量解析问题

刚接触ThingsBoard规则引擎的人，容易在REST API Call节点的Body里直接写：

INSERT INTO device_telemetry VALUES (${metadata.ts}, '${metadata.deviceName}', ${msg.temperature}, ${msg.humidity})

这个写法在某些版本下能生效，但很依赖消息里有没有metadata.ts、metadata.deviceName这些字段。如果设备上报的数据字段经过数据转换后变了名字，或者没有ts字段，模板会解析失败，请求体里出现空值。

我更推荐在REST API Call节点前面放一个Script节点，先用脚本把SQL拼好，再交给REST节点发送。脚本大致这样：

var sql = "INSERT INTO device_telemetry VALUES (" + metadata.ts + ", '" + metadata.deviceName + "', " + msg.temperature + ", " + msg.humidity + ")"; return { msg: { sql: sql }, metadata: metadata };

然后在REST API Call节点的Body里直接写：

${msg.sql}

这样做的好处是逻辑更清晰，也方便在脚本里统一处理字符串转义、空值过滤、单位换算等问题。设备名称、字符串字段里的引号，建议在脚本里做一次转义，否则SQL可能被截断。

4.4 验证写入和失败排查

配置完成后，用模拟设备发一条遥测数据。到TDengine里执行：

SELECT last_row(ts, temperature, humidity) FROM device_telemetry WHERE device_id = 'device001';

如果有数据返回，说明链路通了。如果没数据，先看ThingsBoard规则链里节点的Success和Failure分支计数，通过右上角的“节点统计”可以快速定位是哪一步失败了。

我踩过的一个坑是ts字段精度不一致。ThingsBoard消息里的ts一般是毫秒时间戳，13位数字。TDengine默认支持毫秒、微秒、纳秒，但如果你建表时没有指定精度，默认是毫秒。如果消息里的ts是秒级或者微秒级，写入后查询会偏移。最稳妥的方式是在建库时明确指定：

CREATE DATABASE IF NOT EXISTS thingsboard PRECISION 'ms' KEEP 90;

数据从源头统一成毫秒，后面查询才不会出现边界时间少一秒多一秒的问题。

5. 进阶玩法：用TDengine的数据反过来影响RPC下发

ThingsBoard不止是接收遥测。项目中经常需要“根据时序数据判断再下发命令”，比如设备温度连续3分钟超过阈值，才触发断电指令。这个场景特别适合把TDengine作为状态查询源。

5.1 RPC下发的基本套路

ThingsBoard下发RPC命令，是利用规则引擎里的RPC Call Request节点，向设备发布v1/devices/me/rpc/request/+主题。节点需要配置Timeout、设备名称、请求方法等字段。

基本流程是先构造RPC参数，比如：

{ "method": "setThreshold", "params": { "value": 80 } }

然后交给RPC Call Request节点发送，设备侧订阅RPC主题后会收到这个请求，执行后返回响应。

5.2 在RPC前先查TDengine的时序状态

要结合TDengine，可以先在规则链里增加一个REST API Call节点，POST一条查询SQL给TDengine，再从结果里判断是否触发RPC。

举个例子，我想查某个设备最近5条温度数据：

SELECT LAST_ROW(temperature) FROM device_telemetry WHERE device_id = 'device001';

在REST API Call节点里配置好URL和认证后，TDengine返回的结果是一个JSON数组，需要再经过一个Script节点解析。比如把响应里的温度值取出来，和阈值比较：

var result = JSON.parse(msg.sqlResult); if (result.code === 0 && result.data && result.data.length > 0) { var currentTemp = result.data[0][0]; if (currentTemp > 80) { msg.shouldRpc = true; } else { msg.shouldRpc = false; } } return { msg: msg, metadata: metadata };

这里的msg.sqlResult是REST API Call节点的响应体，具体字段名要看节点配置里怎么保存返回值。我在实际项目里习惯把响应体完整存到一个自定义字段里，后面想调试也方便。

解析出shouldRpc之后，规则链里放一个Switch节点，满足条件就走到RPC Call Request节点，不满足就走结束分支。这样就把“TDengine的时序判断”和“ThingsBoard的命令下发”串起来了。

5.3 子设备RPC下发的注意事项

ThingsBoard里的设备分两种：直连设备和网关设备。如果传感器是挂在网关下的子设备，RPC下发不是直接发给子设备，而是先发给网关设备，由网关翻译后转发给子设备。

在规则链里做子设备RPC时，容易出现目标设备选错的问题。RPC Call Request节点的目标应该是网关设备，而子设备的标识要放到RPC的params里，比如：

{ "method": "childCommand", "params": { "targetDevice": "sensor_01", "command": "restart" } }

网关收到这个命令后，再通过自己的协议去和子设备通信。所以集成TDengine的时候，如果要按子设备查询历史数据，建议在TDengine的标签里把子设备ID也存进去，否则RPC下发前查询状态会找不到目标设备。

6. 上线后我遇到的高频问题和调优心得

6.1 批量写入与连接复用

如果选择规则引擎直写TDengine，最容易遇到的瓶颈是单条HTTP请求太多。一台设备10秒上报一次，几千台设备就相当于每秒几百个请求。TDengine本身能扛住，但网络开销和taosAdapter的连接压力不小。

我后来在规则链里加了一个聚合思路：不是每条消息都调一次REST，而是先把多条消息暂存在规则引擎的内存队列里，攒到一定数量后拼成一条多行SQL再写入。TDengine的INSERT语句支持一条语句插入多行：

INSERT INTO device_telemetry VALUES (1700000000000, 'device001', 23.5, 60), (1700000001000, 'device002', 24.1, 62);

一次HTTP请求写入几百行，吞吐量提升非常明显。如果不想自己写聚合逻辑，也可以引入Kafka管道，把批量问题交给taosAdapter处理。

6.2 时间戳与字段精度

TDengine对时间戳精度特别敏感。如果你的建库语句没有指定PRECISION，默认是毫秒。但设备上报的可能是秒级时间戳，也可能是微秒级的，不统一就乱了。

我建议所有设备在上报前统一转换为毫秒时间戳，不要在TDengine端再去做单位换算。时间戳是第一列，排序和分区都依赖它，一旦有脏数据，查询结果会对不上。

另外，TDengine的浮点字段类型是FLOAT和DOUBLE。如果设备上报的数值本身是整数，但可能在后面版本里扩展精度，建议直接用DOUBLE，省得后期改表结构。

6.3 数据保留策略

时序数据的特点是越老的数据价值越低，但占用磁盘空间不减。TDengine的KEEP参数直接控制数据保留天数，比自己在PostgreSQL里写删除任务优雅得多。

我常用的配置：

CREATE DATABASE IF NOT EXISTS thingsboard KEEP 90 DURATION 10;

这里DURATION 10表示每10天一个分区，过期分区会被自动丢弃。如果有特殊设备的数据需要长期保留，可以把这些设备的子表放到另一个保留周期更长的库里，避免全局策略一刀切。

6.4 监控与告警建议

集成跑起来之后，不要只看ThingsBoard的仪表盘，TDengine这一侧也要盯几个关键指标：

taosAdapter的HTTP请求量和错误率。
TDengine的写入QPS和查询响应时间。
磁盘空间使用率，尤其是分区文件增长情况。
规则链节点的Failure计数，一旦有异常马上能看到。

我常用的方式是给TDengine的监控数据也接入一套告警，比如连续5分钟查询响应超过500ms就告警。这样能在用户反馈之前发现链路问题。

整个方案跑下来，我最大的体会是：TDengine和ThingsBoard集成并不难，难的是想清楚数据流的方向和边界。ThingsBoard负责设备接入和业务规则，TDengine负责时序数据的存储和分析，各管一段，不要混在一起。如果你正准备做这个集成，我建议先按直写方案跑通最小链路，等数据量上来再演进到Kafka管道。先把链路跑通，再考虑优化，这是我认为最稳妥的节奏。