TDengine 与 Node-RED 深度集成实战指南基于 node-red-node-tdengine 实现工业 IoT 数据采集、实时查询与告警订阅【免费下载链接】tdengineTDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.项目地址: https://gitcode.com/taosdata/tdengineTDengine 官方为 Node-RED 提供了node-red-node-tdengine插件通过拖拽式的可视化编程即可完成从设备数据采集、写入、实时查询到事件订阅告警的完整链路。本文以智能电表车间监控为实战场景完整讲解插件的两个核心节点tdengine-operatorSQL 写/查与tdengine-consumer数据订阅消费的配置方法、数据建模、流程搭建与错误处理并深入当前仓库的 Node.js 连接器与 taosAdapter 源码文档揭示连接串格式、订阅 Topic 机制等底层原理。Node-RED 与 TDengine 的集成架构Node-RED 是 IBM 基于 Node.js 开发的开源可视化编程工具用户通过图形化界面将各类节点拖拽、连线、组合即可快速搭建 IoT 设备、API 与在线服务之间的数据流。它支持多种协议、跨平台运行、社区活跃适合智能家居、工业自动化等事件驱动型应用开发核心优势是低代码与可视化。TDengine 与 Node-RED 的深度集成为工业 IoT 场景提供了一体化方案通过 Node-RED 的 MQTT / OPC UA / Modbus 协议节点可以从 PLC、传感器等设备采集数据毫秒级延迟TDengine 中的实时查询结果可以触发继电器动作、阀门切换等物理控制实现即时指令下发。node-red-node-tdengine是 TDengine 官方为 Node-RED 开发的插件由两个节点组成tdengine-operator提供 SQL 执行能力用于数据写入、查询和元数据管理。tdengine-consumer提供数据订阅与消费能力从指定的订阅服务器和 Topic 中拉取数据。从组件架构看浏览器中的 Node-RED 前端负责流程编排node-red-node-tdengine通过 JavaScript 调用 TDengine 官方的 Node.js 连接器tdengine/websocket连接器经 WebSocket 协议连接 taosAdapter 服务再由 taosAdapter 通过原生协议与taosd核心存储服务通信。组件与版本前提在动手前需要准备以下环境组件组件版本要求说明TDengine 集群3.3.2.0 及以上企业版 / 社区版 / 云服务均支持taosAdapter随版本部署参考 taosAdapter Reference所有语言连接器都通过其 WebSocket 接口与 TDengine 通信Node-RED3.0.0 及以上参考 Node-RED 官方安装文档Node.js 连接器tdengine/websocket3.1.8 及以上官方 npm 包node-red-node-tdengine 插件最新版官方 npm 包为什么必须依赖 taosAdapter从 taosAdapter 文档 可知taosAdapter 是 TDengine 集群与应用之间的桥接/适配组件通过其 WebSocket 接口各语言连接器才能实现 SQL 执行、无模式写入、参数绑定和数据订阅功能。Node-RED 插件正是通过 WebSocket 与 taosAdapter 建立连接因此 taosAdapter 必须处于运行状态。插件数据源配置插件的数据源在节点属性中通过 Node.js 连接器格式配置参考 Node.js 连接器文档启动 Node-RED 服务在浏览器中访问 Node-RED 首页。从左侧节点面板中将 tdengine-operator 或 tdengine-consumer 节点拖拽到工作区画布上。双击画布上的节点打开属性设置tdengine-operator 节点连接格式ws://user:passwordhost:port例如ws://root:taosdatawww.example.com:6041。tdengine-consumer 节点连接格式ws://host:port。点击右上区域的字典图标可查看详细文档。配置完成后点击右上角 Deploy部署按钮节点变为绿色即表示连接成功。从源码文档看这个连接串遵循tdengine/websocket的 DSN 规范见 Node.js 连接器 API 参考[protocol]://[username:password][host1:port1[,...hostN:portN]][/database][?p1v1[p2v2]]protocol使用 WebSocket 协议如ws://localhost:6041。username/password数据库用户名和密码。addresses支持一个或多个host:port地址逗号分隔如node1:6041,node2:6041可用于负载均衡与故障转移场景IPv6 地址需用方括号包裹如ws://root:taosdata[::1]:6041。database数据库名。params如token云服务认证、bearer_tokenTDengine TSDB 认证优先级高于用户名密码、adapter_hataosAdapter 高可用需 TDengine 3.4.2.0 及以上。实战场景智能电表车间监控场景概述一个车间部署了多台智能电表每台电表每秒产生一条数据记录数据存储在 TDengine 中并每分钟执行一次实时计算展示每台电表的平均电流/电压。电耗。当电流 25A 或电压 230V 时触发告警。实现采用 Node-RED TDengineInject function 节点模拟设备。tdengine-operator 写入数据。通过 tdengine-operator 实时查询。通过 tdengine-consumer 订阅实现过载告警。场景假设TDengine 服务器www.example.com。WebSocket 端口6041。使用默认凭据。模拟设备d0、d1、d2。数据建模使用 taos-CLI 手动创建数据模型超级表meters子表 d0、d1、d2。create database test; create stable test.meters (ts timestamp, current float, voltage int, phase float) tags (groupid int, location varchar(24)); create table test.d0 using test.meters tags(1, workshop1); create table test.d1 using test.meters tags(2, workshop1); create table test.d2 using test.meters tags(2, workshop2);这里用到的是 TDengine 的**超级表STABLE**机制meters定义了统一的表结构时间戳 电流、电压、相位三个数据列groupid、location作为标签tags描述设备属性每个子表通过USING ... TAGS(...)继承超级表结构并携带各自的标签值。这样既保证了同类设备的数据结构统一又允许按标签高效过滤查询。业务处理流程搭建数据采集tdengine-operator 写入本示例使用随机数模拟真实设备数据。tdengine-operator 节点配置 TDengine 数据源连接信息将数据写入 TDengine并通过 debug 节点在界面上监控成功写入的记录数。添加写入节点 td-writer在节点面板中选择 tdengine-operator 节点并拖到画布。双击节点打开属性设置名称填写td-writer点击 database 字段右侧的 图标。在弹出的窗口中Namedb-server。Connection typeConnection string连接串。输入ws://root:taosdatawww.example.com:6041点击 Add 返回。模拟设备数据function 节点从面板选择 function 节点拖到画布上td-writer之前。双击节点Namewrite d0。选择 Function 标签页输入// Generate random values const current Math.floor(Math.random() * (30 - 5 1)) 5; // 5-30A const voltage Math.floor(Math.random() * (240 - 198 1)) 198; // 198-240V const phase Math.floor(Math.random() * (3 - 1 1)) 1; // 1-3 // Create SQL msg.topic insert into test.d0 values (now, ${current}, ${voltage}, ${phase});; return msg;在write d0之前拖入一个 inject 节点。配置 inject 节点Nameinject1。触发方式Repeat重复。间隔1 秒。对其他设备d1、d2重复步骤 1-4。添加输出监控debug 节点在td-writer之后拖入一个 debug 节点。配置Namedebug1。Node status勾选。在下拉列表中选择Message count消息计数。所有节点添加完成后按顺序连接形成流水线点击 Deploy 发布。运行成功时td-writer变绿debug1显示数据计数。写入成功输出示例{ topic: insert into test.d1 values (now, 20, 203, 2);, isQuery: false, payload: { affectRows: 1, totalTime: 2, timing: 961982 } }关于上述返回结构affectRows表示受影响行数totalTime与timing分别是总响应时间与执行耗时。这与 Node.js 连接器中exec()接口返回的TaosResult结构affectRows/timing/totalTime完全对应说明 tdengine-operator 底层执行 SQL 写操作时直接复用了连接器的 SQL 执行能力。实时数据查询tdengine-operator 查询数据查询工作流由三个节点inject / tdengine-operator / debug组成用于计算每台智能电表每分钟的平均电流、平均电压和电耗。inject 节点每分钟触发一次查询请求结果发送到下游 debug 节点显示成功执行的查询次数。步骤拖入一个 inject 节点Namequery。设置 msg.topic 为select tbname, avg(current), avg(voltage), sum(p) from ( select tbname,current,voltage,current*voltage/60 as p from test.meters where ts now-60s partition by tbname) group by tbname;拖入 tdengine-operator 节点Database选择已有的db-server连接。保存并返回。拖入 debug 节点并配置Namedebug2。Node status勾选。在下拉列表中选择Message count。按顺序连接节点并点击 Deploy。流程成功启动后td-reader节点变绿debug 节点显示结果计数。td-reader的输出失败时抛出异常{ topic: select tbname,avg(current) ..., isQuery: true, payload: [ { tbname: d2, avg(current): 26.7, avg(voltage): 235, sum(p): 6329 }, { tbname: d0, avg(current): 16.5, avg(voltage): 222, sum(p): 121 }, { tbname: d1, avg(current): 29, avg(voltage): 202, sum(p): 5833 } ] }查询语句要点解析partition by tbname对超级表meters按子表设备分区保证各设备的电流、电压数据独立参与计算current*voltage/60 as p基于功率 电压 × 电流估算电耗除以 60 得到每分钟千瓦时量级外层group by tbname与avg()、sum()聚合得到每台设备每分钟的统计指标返回的payload是数组结构与连接器query()接口返回的数据集一致isQuery: true表明这是一次查询而非写入。数据订阅tdengine-consumer 消费告警数据订阅工作流由两个节点tdengine-consumer / debug组成提供设备过载告警功能。debug 节点直观显示下游推送的订阅消息计数生产环境可将其替换为功能节点来处理订阅数据。步骤使用 taos-CLI 手动创建订阅 Topictopic_overloadcreate topic topic_overload as select tbname,* from test.meters where current 25 or voltage 230;拖入 tdengine-consumer 节点Nametd-consumer。Subscription Serverws://www.example.com:6041。Usernameroot。Passwordtaosdata。Topicstopic_overload。Initial Offsetlatest。其他设置默认。拖入 debug 节点并配置Namedebug3。Node status勾选。在下拉列表中选择Message count。按顺序连接节点并点击 Deploy。运行后td-consumer节点变绿debug 节点显示消费计数。td-consumer的告警输出{ topic: topic_overload, payload: [ { tbname: d1, ts: 1750140456777, current: 31, voltage: 217, phase: 2, groupid: 4, location: California.MountainView } ], database: test, vgroup_id: 4, precision: 0 }Topic 机制底层说明从仓库的 Topic 语法文档 可知从 TDengine v3.0.0.0 起用户可以用 SQL 直接创建 Topic 并通过连接器 API、taos shell 或 MQTT 客户端消费。create topic ... as select ...创建的是查询类型 Topic它订阅一段 SELECT 语句定义的数据流可包含过滤条件和标量函数但不支持聚合函数、时间窗口聚合以及DISTINCT、GROUP BY、ORDER BY、PARTITION BY、LIMIT/SLIMIT等子句。创建后订阅数据的结构即被固定。本示例中select tbname,* from test.meters where current 25 or voltage 230将超级表meters中所有电流超限或电压超限的记录实时推送给消费者因而天然实现了过载告警的事件驱动能力。消费端Initial Offset设置为latest与 Node.js 连接器订阅常量AUTO_OFFSET_RESETlatest表示只消费新数据earliest表示包含旧数据语义一致vgroup_id对应 TDengine 存储引擎中该数据所在的虚拟节点组precision为时间精度标志。错误处理catch 节点数据采集、查询和订阅流程中的错误在 Node-RED 中会被路由到 catch 节点统一处理。实现错误监控拖入一个 catch 节点。配置节点Namecatch all except。ScopeAll nodes所有节点。拖入 debug 节点并配置Namedebug4。Node status勾选。在下拉列表中选择Message count。连接节点并部署。当发生错误时debug 节点显示错误计数详细错误信息可在 Node-RED 日志中查看。运行时总览部署后的完整工作流总览如下写、查、消费、错误捕获四路并行体现了生产可用的参考架构生产化要点连接信息安全示例使用默认凭据root:taosdata生产环境务必修改默认密码并可通过连接串参数使用token/bearer_token进行云服务或 TSDB 认证。写入性能高吞吐场景下建议采用参数绑定stmt批量写入而非逐条 SQL insertNode.js 连接器的stmtInit/prepare/bind/batch/exec链路即为该能力可参考连接器文档中的 stmt 示例。消费可靠性tdengine-consumer 的消费进度可通过连接器的commit/committed/seek/seekToBeginning/seekToEnd等接口管理生产环境可结合ENABLE_AUTO_COMMIT、AUTO_COMMIT_INTERVAL_MS等订阅属性控制提交策略。高可用TDengine 3.4.2.0 及以上可启用adapter_ha连接器会自动发现集群内其他 taosAdapter 端点并加入地址池实现负载均衡与故障转移配合多地址 DSNnode1:6041,node2:6041进一步提升可用性。错误治理务必为写/查/订阅流程接入 catch 节点并落盘日志避免数据链路静默失败。总结通过工业监控场景本文完整演示了 Node-RED 与 TDengine 的三种集成模式数据采集tdengine-operator 写入inject function 模拟设备周期写入超级表子表。实时查询tdengine-operator 查询按设备分区聚合输出每分钟平均电流、电压与电耗。事件驱动架构tdengine-consumer 订阅基于 SQL Topic 实时推送过载告警。同时覆盖了完整的错误处理机制以及可直接参考的生产化部署架构。本文章节以示例为主线更完整的节点说明可参阅 node-red-node-tdengine 节点的在线文档连接层细节可继续阅读仓库中的 Node.js 连接器参考、taosAdapter 参考 与 Topic 语法。【免费下载链接】tdengineTDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.项目地址: https://gitcode.com/taosdata/tdengine创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
