源码分析 ThingsBoard 规则节点

正文与配图以原稿为主；原稿仅列标题或未展开的节点，按 ThingsBoard 官方 Rule Nodes 补全说明，文中标为「辅」。
已单独成篇的节点可对照：check alarm status、check existence fields、check relation、定时 Generator。

规则节点是规则链的积木：每次处理一条消息，再经命名连线（Success / Failure / True / False 或自定义标签）交给下游。常见六类：过滤器、属性集（Enrichment）、变换、动作、外部、流。

---

一、过滤器节点（Filter）

不改写数据，只按条件分流。

1、asset profile switch

按发起者资产的 Asset Profile 名称把消息路由到同名连线。发起者必须是 ASSET，否则走 Failure。节点本身几乎无需配置，连线标签要与资产配置名称完全一致。

案例准备：

新建设备配置 / 新建资产配置：

新建设备 / 新建资产：

资产与设备绑定关系：

新建规则链（含 asset profile switch）：

规则链底层查询（源码侧按资产配置名匹配出口）：

验证结果：

2、check alarm status

该节点主要在告警类消息处理时有意义。结合源码，常见会跟 Alarm Acknowledged、Alarm Cleared 等消息类型一起用。

场景： 智慧农业老板下有很多大棚，大棚分给管理员；某大棚温度异常触发告警，推送给老板和管理员。管理员点击 Alarm Acknowledged 后，可用本节点过滤，再自定义短信等通知老板「已有人在处理」。

四种状态与完整跟做见：规则链：check alarm status。

3、check existence fields

字面意思：检查字段是否存在，常用于设备上报完整性校验（msg / metadata）。

专题：规则链：check existence fields。

4、check relation

检查实体与实体之间的关联是否存在。

场景（大棚蔬菜）： 每个大棚是资产，棚内温度传感器是设备；关系「大棚 → 温度传感器」。大棚为 From，设备为 To。设备上报后若要把数据报到大棚遥测，可先 check relation 确认设备与大棚有关联，再继续业务。方向是谁就找谁。

专题：规则链：check relation。

5、device profile switch

辅（官方）： 按发起者设备的 Device Profile 名称路由到同名连线。发起者必须是 DEVICE；资产/客户等走 Failure。无配置项——连线标签须与设备配置名完全一致。适合同一规则链内按设备类型分叉处理。

6、entity type（entity type filter）

辅（官方）： 勾选一种或多种发起者实体类型（Device、Asset、Customer…）。匹配 → True，否则 → False。适合放在「只处理设备」类节点前面挡掉资产/客户消息。

7、entity type switch

辅（官方）： 无配置。按发起者类型的可读名路由，如 Device、Asset、Customer、Tenant、Entity View、Edge 等。与 filter 的 True/False 不同，可直接为各实体类型接不同下游。

8、gps geofencing filter

辅（官方）： 从 msg（或回退到 metadata）读取经纬度，判断点是否在围栏内。围栏支持 Polygon / Circle；也可「从 metadata 拉取围栏定义」实现按设备不同区域。在内 → True，在外 → False；坐标缺失或围栏非法 → Failure。

9、message type

辅（官方）： 勾选允许的消息类型（如 POST_TELEMETRY_REQUEST、POST_ATTRIBUTES_REQUEST）。类型在集合内 → True，否则 → False。常放在规则链入口后做最简单分流。

10、message type switch

辅（官方）： 按消息类型名路由到同名连线（遥测、属性、RPC、告警确认等），无需脚本。根规则链里常见用法。

11、script（过滤器）

辅（官方）： 用 TBEL/JS 写返回布尔值的函数，出口为 True / False。入参一般为 msg、metadata、msgType。适合官方节点盖不住的复杂条件。

12、switch

和 script 节点执行逻辑是一套的，但返回的是一个或多个连线名称（字符串数组），用于自定义多路分支，而不是单纯 True/False。

辅（官方）： 脚本返回连接标签列表，消息可复制到多条出口。

---

二、属性集节点（Enrichment / metadata 包）

向消息 metadata（或 data）补充属性、最新遥测、实体字段等，供下游使用。

13、calculate delta（计算增量）

辅（官方）： 计算当前值与上一值的差值，可选带上时间间隔。常用于脉冲计数、累计电量差分等。

14、customer attributes（查询客户属性或最新遥测）

找到消息发起者所属客户，查询客户的属性或最新遥测，写入消息元数据。

15、customer details（查询客户基础信息字段）

辅（官方）： 把所属客户的 name、email、address 等详情字段写入 metadata，便于告警模板、邮件正文引用。

16、fetch device credentials（查询设备凭证）

辅（官方）： 将发起者设备的凭证类型与凭证值写入 metadata（注意脱敏与权限；生产环境慎把密钥打到外部系统）。

17、originator attributes（查询发起者属性或最新遥测）

辅（官方）： 从发起者拉取 client / shared / server 属性及最新遥测。属性键常带 cs_ / shared_ / ss_ 前缀；可写入 data 或 metadata。

18、originator fields（查询发起者实体字段）

辅（官方）： 取发起者实体字段（name、type、label、profile 等）进 metadata。定时 RPC、换发起者后校验时常与 Generator 调度 联用。

19、originator telemetry（查询发起者遥测）

辅（官方）： 按时间窗口拉取发起者历史时序，可聚合（AVG/MIN/MAX 等）。与「最新遥测」节点不同，这里面向区间历史。

20、related attributes（关系查询关联实体属性或最新遥测）

辅（官方）： 新版文档中常对应 Related entity data：按关系查询找到关联实体，再取其属性 / 最新遥测 / 字段。需配置方向、关联类型、最大层级等。

21、related device attributes

辅（官方）： 专门沿关系找到关联设备，再取该设备的 client/shared/server 属性或最新遥测。可按设备配置过滤；多设备匹配时通常只用第一个。属性写入时常带 cs_ / shared_ / ss_ 前缀。找不到关联设备 → Failure。

22、tenant attributes

辅（官方）： 取发起者所属租户的服务端属性或最新遥测写入消息，适合租户级阈值、开关等配置下沉到规则链。

23、tenant details

辅（官方）： 取租户详情字段（名称、邮箱、地址等）写入 metadata，用法类似 customer details，作用域是租户。

---

三、变换节点（Transformation）

改消息内容、结构或发起者。

change originator

辅（官方）： 把消息发起者改成另一实体后再存数、建告警。可选：所属 Customer、Tenant、按关系查到的实体、告警发起实体、按名称模式查找实体（支持 ${metadataKey}）。网关代报、设备数据报到资产等场景常用。找不到目标 → Failure。

copy keys

辅（官方）： 在 msg data ↔ metadata 之间按键复制。告警、发邮件前常把遥测字段拷到 metadata。

delete keys

辅（官方）： 从 data 或 metadata 删除指定键，用于脱敏或瘦身后再外发。

json path

JSONPath 语法可参考：https://blog.csdn.net/weixin_45503499/article/details/119739658

用表达式从内嵌 JSON 抽出片段，变成新的 payload。

基本操作符：

符号	含义
$	根节点（整个 JSON）
@	当前节点（多用于过滤子表达式）
. 或 []	连接父子节点
..	深层扫描
*	通配符
[n (, m)]	数组下标
[start:end]	数组切片
[?(expr)]	过滤表达式
.<name>	Dot 取子节点
['a','b']	Bracket 一次取多个键

设备上传示例（节选结构）：

{
  "msg": "操作成功",
  "code": 200,
  "data": [
    {
      "original_price": "0.01",
      "code": 678913,
      "category_name": "平板-进口",
      "name": "示例商品",
      "price": "0.01"
    }
  ]
}

取值示例：

1. $.data — 取根下 data
2. $["msg","code"] — 一次拼多个 key 得到新 JSON 子串

rename keys

更换 data、metadata 中的字段名为自己指定的 key。

script（变换）

使用自定义 JavaScript（或 TBEL）更改消息负载、元数据或消息类型。

函数常见三个入参：

• msg — 消息负载
• metadata — 元数据
• msgType — 消息类型

可返回修改后的对象；部分版本还支持一次输出多条消息。

split array msg

将数组消息拆成多条消息：从 msg 取数组，元素逐个下发。传入必须是数组结构。

演示案例：

{
  "array": [
    "JSON是什么？",
    "JSON语法规则",
    "JSON数据类型"
  ]
}

节点处理：

to email（构建发送邮件的模板）

该节点需配合外部节点 send email 使用：先把消息变成 SEND_EMAIL 形态（发件人、收件人、主题、正文模板），再由 send email 真正走 SMTP。

参见：配置发送邮件、如何发送告警邮件。

---

四、动作节点（Action）

有副作用：落库、告警、RPC、改关系等。

assign to customer

辅（官方）： 将消息发起者实体分配给指定客户（可按名称模式查找客户）。

clear alarm

辅（官方）： 清除发起者上处于激活状态的告警（可按告警类型等配置）。

copy to view

辅（官方）： 把设备/资产上的属性变更复制到关联的 Entity View，便于视图侧展示与权限隔离。

create relation

辅（官方）： 在发起者与配置的目标实体之间创建关联（类型、方向、目标实体等可配）。

deduplication

根据指定的重复数据删除策略，在可配置时间窗内做去重。

可选策略：

策略	含义
FIRST	返回窗口内到达的第一条消息
LAST	返回窗口内到达的最后一条消息
ALL	把窗口内消息合并成一个 JSON 数组（元素含 msg 与 metadata）

delay（已弃用）

在可配置时间段内延迟消息。

注意：该节点会确认当前队列中的消息（消息从队列删除）。已弃用原因：已确认的消息仍留在内存中等待延迟，进程重启会丢；即使选了「重试失败和超时」也不能保证一定被处理。新版请优先用 Checkpoint 等队列方案。

delete attributes

辅（官方）： 删除发起者上指定的属性键（需选 scope）。

delete relation

辅（官方）： 删除发起者与配置实体之间的关联（可按方向、类型、是否删到特定实体等配置）。

device profile（较复杂）

根据设备配置文件中的设置处理设备消息：按设备配置里定义的告警规则创建和清除告警。

输出关系类型包括：告警创建、告警更新、告警级别更新、告警清除；若不影响告警则走 Success。

辅（官方）： 部分新版本将「设备配置告警处理」标为 deprecated，能力逐步由其它告警/计算字段机制承接；读源码时注意你所用分支的节点类名与默认根链是否仍挂此节点。

generator

定期生成消息；用 JS/TBEL 定义载荷。可配置周期与 Originator。CE 上常当「伪调度器」用。

专题：实现定时任务或调度器。

gps geofencing events

从消息 data/metadata 提取经纬度，对照配置的围栏，产生进入 / 离开 / 停留等事件并路由（与仅 True/False 的 filter 不同，本节点偏「事件动作」）。

log（简单提一下）

用脚本格式化后写入 ThingsBoard 日志，调试规则链最常用。

math function（简单讲）

对来自 msg / metadata / 属性 / 时序的数值做运算，结果写回上述任一目标。

message count

辅（官方）： 在可配置时间间隔内统计流入消息数，并按服务节点等维度以遥测形式吐出计数，用于流量观察。

push to edge

原稿：不讲。
辅： 云端实例把消息推到 Edge；Edge 上回云用 push to cloud。仅混合 Edge 部署需要。

rpc call reply

设备向服务端发 RPC，服务端应答。会走到 tb-core。

rpc call request

服务端向设备发送 RPC 请求，设备响应可作为出站消息 data。

save attributes

辅（官方）： 将消息 data 持久化为发起者的属性（需选 client/shared/server 等 scope）。

save timeseries

辅（官方）： 将消息 data 持久化为发起者时序（遥测）。默认根链里「上报遥测 → save timeseries」最常见。

save to custom table（简单讲一下）

将传入 Message payload 写入 Cassandra 自定义表。

要点（原稿）：

• 表应带预定义前缀 cs_tb_，避免误写入 TB 公共表
• 仅 Cassandra 可用
• 管理员配置表名时不要带前缀 cs_tb_（由系统加）
• 可配置消息字段名 ↔ 表列名映射
• 映射键为 $entity_id 时，写入消息发起者 ID
• 字段不存在或不是 JSON 原始类型 → Failure，否则 → Success

初始化 / 封装参数 / 执行插入：

synchronization start / end（简单提一下）

synchronization start 现已弃用，请用 Checkpoint 代替。

原设计：与 synchronization end 成对使用；按消息发起者 id 入队，同一发起者后续消息在前一条处理完或超时前不会继续。每发起者队列大小与超时可在系统级配置。

unassign from customer

按客户名称模式找到客户，再将发起者实体从该客户取消分配。

---

五、外部节点（External）

把数据发到第三方系统。原稿列了常用几个，说明按官方补全：

节点	说明（辅）
kafka	发布到 Apache Kafka Topic
mqtt	发布到外部 MQTT Broker
rabbitmq	发布到 RabbitMQ Exchange
rest api call	向外部 HTTP 端点发请求，响应体可写回消息
send email	通过 SMTP 发送；需上游 to email 先备好 SEND_EMAIL 消息

另有 AWS SNS/SQS、GCP Pub/Sub、Send SMS、Slack 等，按版本在节点面板中查看。自定义短信节点见：华为云 SMS、阿里云 SMS。

---

六、流节点（Flow）

控制消息在规则链 / 队列之间如何流动：

节点	说明（辅）
acknowledge	确认队列中的消息并继续转发给下游
checkpoint	把消息转到指定队列，便于隔离或串行处理（替代旧 synchronization）
rule chain	把消息转发到另一条规则链（子链）
output	子链处理结束后，把结果返回给调用方的 Rule Chain 节点

---

读源码时怎么对照

1. 节点 UI 名称 ↔ Java 类：多在 rule-engine 相关模块，类名常带 Tb…Node。
2. 先看 init / 配置类，再看 onMsg 里如何 ctx.tellSuccess / tellNext / tellFailure。
3. 过滤器多是 True/False 或「连线标签 = 配置名/类型名」；属性集改 metadata；动作为副作用；外部为出站 IO。
4. 标注 deprecated 的节点（delay、synchronization、部分 device profile 路径）以你检出的分支源码与官方当前文档为准。

相关文章

• 规则链：check alarm status
• 规则链：check existence fields
• 规则链：check relation
• 实现定时任务或调度器
• ThingsBoard 如何发送告警邮件
• 专栏导读

官方索引：Rule Nodes · Filter · Enrichment · Transformation · Action · External · Flow