简介：RBI_Thingsboard 是一份基于 C++ 开发的物联网平台源码包，面向希望深入理解设备接入、数据存储与可视化实现的开发者和嵌入式爱好者，也适合作为课程设计或项目实训参考。压缩包内共 83 个文件，包含 28 个 cpp 源文件、18 个 h 头文件、22 个 o 编译中间文件，以及 Qt 工程文件、Makefile、配置说明和 PDF 文档，整体大小约 1016KB，结构清晰便于检索。平台后端以 C++ 系统级编程直接操作硬件资源，借助模板和面向对象特性降低维护成本，并利用 Boost、Poco 等库强化网络通信与并发处理；内容预览覆盖启停入口、MQTT 接入、LoRa 收发、数据库管理与图表展示等关键模块，可帮助读者完整梳理从设备端到云端的数据链路，同时观察 Qt 界面与底层通信模块的协作方式。已有 190 人学习下载，适合具备 C++ 基础、希望将高性能语言落地到 IoT 场景的开发者，通过研读源码可掌握跨平台物联网平台的架构划分、并发连接优化及二次开发思路，也可为进一步定制网关功能打下基础。
1. RBI_Thingsboard：先搞清楚这套 IoT 集成方案到底在解决什么
RBI_Thingsboard 这个标题，字面拆开就是 RBI 加 ThingsBoard。我在实际工程里更愿意把 RBI 理解成 Rule-Based Integration：基于规则做设备接入、数据处理和业务联动的集成方案。它解决的是 IoT 项目里最常被低估的一环——设备数据进来后，不只是存起来画几条曲线，还要在几秒内完成解析、判断、告警，再回传给业务系统。这套方案适合正在选型 IoT 平台的团队，也适合已经跑起 ThingsBoard、却还只拿它当数据展示工具的开发者。接下来从核心机制、最小数据链路、规则链配置、高频故障排查到生产化技巧，一路讲到底，保证每个环节都能照着落地。
2. 拆解 ThingsBoard 的核心机制：接入、数据模型与规则引擎
2.1 设备接入与数据模型：device、asset 与遥测/属性
ThingsBoard 的接入层相当透明，MQTT、HTTP、CoAP 三条路都支持。生产环境里绝大多数设备走 MQTT，这个选择不是跟风：MQTT 长连接开销小，高频率上报时不必反复握手；ThingsBoard 的 MQTT 认证直接用设备的 access token 作为用户名，topic 路径固定，调试成本最低。HTTP 上报适合低频脚本或网关侧聚合成批的数据，CoAP 适合电池供电的受限节点，但 CoAP 的调试工具少、抓包麻烦，第一版不建议上，等技术方案稳定后再补。
数据模型先分清两组容易混的概念：device 与 asset，遥测与属性。device 对应真实设备，比如一台温控器；asset 对应逻辑资产，比如一条产线、一座泵站。两者可以建立关联关系，告警能从 device 上报后自动向 asset 和租户层级传播。这一点在 RBI 方案里价值很大：现场只看产线级告警，维修才下钻到具体设备，规则链里配一个 propagate 就能实现。
遥测（telemetry）是带时间戳的时序值，属性（attribute）是稳定的键值对。两者分开存储是刻意的设计：遥测数据量大，查询要按时间窗口走时序索引；属性变化频率低，适合放在普通键值存储里。不知道这个区分，后面写规则链时就容易把属性当遥测处理，导致落库混乱。属性还按作用域分成三类：
| 属性作用域 | 写入方 | 典型用途 |
|---|---|---|
| client scope | 设备端 | 上报设备本地状态、运行模式 |
| shared scope | 服务端，可下发 | 下发远程参数配置给设备 |
| server scope | 服务端私有 | 设备密钥、内部标记，不上报设备 |
开发 RBI 时我习惯把采集周期、上报阈值这类参数放进 shared scope，这样改参数不用重新烧固件。设备每次连着 MQTT 的 v1/devices/me/attributes topic，启动时拉一次就能拿到最新配置。服务器私有信息放 server scope，设备永远读不到。
2.2 规则引擎的节点式处理：从 message 到 action
规则引擎是 RBI 方案真正的核心。设备消息进入平台后先落在传输层，再由规则引擎按规则链逐节点处理。一条消息在链里流动时始终带着三样东西：msg 是原始负载，通常是 JSON；metadata 是设备名、设备类型、客户端地址这些上下文；msgType 标识消息类型，比如 POST_TELEMETRY_REQUEST 表示设备上报遥测。弄清楚这三样东西，后面写脚本 filter 才知道去哪里取数。
规则链由节点串成，节点按职责分四类：filter 负责判断走哪条分支，比如 script filter、message type switch；transform 负责改写消息内容，比如抽字段、补时间戳；action 负责真正干活，如保存遥测、建立告警、调用外部 REST 接口；external 负责把消息推到 Kafka、RabbitMQ 这类外部队列。RBI 最常见的链就是 filter 判断 -> transform 清洗 -> action 落库/告警/回调。
规则引擎不是黑匣子。每个节点都能开 debugMode，开启后消息经过节点时会记录输入和输出结构。我一般先在测试面板里用模拟消息把链路跑通，确认每个节点的调试日志符合预期后再关掉 debugMode。别贪方便一直开着，生产流量一大，调试日志会撑爆存储。
选型上要回答一个问题：判断逻辑为什么放在规则引擎，而不是业务后端？因为设备消息量大、判断逻辑调整频繁。全塞进业务后端意味着接入、存储、计算三件事同时耦合在一个服务里，每一次改告警阈值都要重新发布。规则链把这些问题拆到平台内部，节点间低耦合，改一条判断逻辑只是在界面上改一个脚本。这就是 Rule-Based Integration 比「后端硬编码处理」更适合 IoT 场景的核心原因。
还要注意规则链与 device profile 的绑定关系。每个设备默认挂在租户的根规则链上，但更规范的做法是：在设备 profile 里指定专用规则链，这样同类设备统一走同一条处理逻辑。这是第 6 章进阶内容的地基，先在这里留个印象。开发时遇到「改来改去不生效」，十有八九是改错了链，或者设备根本没挂在你以为的那条链上。
3. 用 Docker 拉起 ThingsBoard 并跑通第一条数据链路
3.1 Docker Compose 部署与初始化检查
本地验证 RBI 方案最快的路径是 docker compose 起一套单机环境。我给的示例省略了数据库外部依赖，直接用官方 tb-postgres 镜像，适合功能验证。注意把数据目录挂出来，否则容器一删全没了。
services: tb: image: thingsboard/tb-postgres:latest restart: always ports: - "9090:9090" - "1883:1883" - "7070:7070" environment: TB_QUEUE_TYPE: in-memory volumes: - tb-data:/data - tb-logs:/var/log/thingsboard volumes: tb-data: tb-logs:保存成 docker-compose.yml 后执行 docker compose up -d 启动。第一次启动要做数据库初始化和索引构建，日志里会持续滚动安装信息。判断初始化完成的标志是 9090 端口能打开登录页，而不是容器状态变成 running。我遇到过容器一直是 running 但 Web 端口迟迟不通的情况，那是初始化还没结束，多等几分钟。
启动参数有几个点值得说。TB_QUEUE_TYPE 设为 in-memory 表示规则引擎消息队列走内存，单机验证足够；生产环境要换成 kafka 或 rabbitmq，否则重启实例会丢排队消息。端口上 9090 是 Web UI 和 REST API 共用端口，1883 是 MQTT 默认端口，7070 是设备边缘 RPC 用的端口，这几项按需映射。用 latest 标签图省事，正式环境一定要把版本固定到具体 tag，避免镜像更新带来行为变化。
3.2 MQTT 上报遥测的最小命令与参数
连接前先去设备列表创建一台设备，在设备详情页复制 access token。这个 token 就是 MQTT 连接时的用户名，密码任意或留空，这是 ThingsBoard 和其它 MQTT broker 最不一样的地方，很多人拿 password 去填 token，导致认证失败。
最简单的是用 mosquitto_pub 验证链路：
export ACCESS_TOKEN="你的设备访问令牌" mosquitto_pub \ -h localhost \ -p 1883 \ -u "$ACCESS_TOKEN" \ -t "v1/devices/me/telemetry" \ -m '{"temperature": 41.2, "humidity": 68}' \ -q 1topic 必须是 v1/devices/me/telemetry，payload 必须是合法 JSON。qos 用 0 或 1 都可以，ThingsBoard 在传输层不做复杂确认，业务确认要靠规则链里的 action 节点去实现。命令执行完没有输出是正常现象，去设备的「最新遥测」页签看有没有 temperature 和 humidity 两条数据。
生产脚本里更多用 Python 的 paho-mqtt 库，好处是可以做批量上报和异常捕获：
import time import paho.mqtt.client as mqtt broker = "localhost" port = 1883 access_token = "你的设备访问令牌" client = mqtt.Client() client.username_pw_set(access_token) # ThingsBoard 用 username 承载设备凭证 client.connect(broker, port, 60) ts = int(time.time() * 1000) payload = [ {"ts": ts, "values": {"temperature": 41.2, "humidity": 68}} ] client.publish("v1/devices/me/telemetry", str(payload), qos=1) client.disconnect()这段代码的关键在 payload 结构。不带 ts 时，ThingsBoard 默认取服务器接收时间；带 ts 时可以补传历史数据或对齐设备本地时间戳。数组格式允许一次上报多个时间点的多组数值，适合网关缓存补报场景。认证参数、topic、时间戳这三个点是最容易翻车的位置，后面避坑章节还会展开。
3.3 REST API 回读数据：token、设备列表与遥测查询
数据进去之后要验证是否落库，REST API 是最直接的检查方式。先用租户管理员账号登录拿到 JWT，后续请求都带上它。
JWT=$(curl -s -X POST http://localhost:9090/api/auth/login \ -H "Content-Type: application/json" \ -d '{"username":"tenant@thingsboard.org","password":"tenant"}' \ | python3 -c "import sys,json; print(json.load(sys.stdin)['token'])") curl -s -X GET "http://localhost:9090/api/tenant/devices?pageSize=10&page=0" \ -H "X-Authorization: Bearer $JWT"登录接口返回的 JSON 里 token 字段就是 JWT，把它塞进后续请求的 X-Authorization 头，前缀记得带 Bearer 和一个空格。设备列表接口返回的是一个分页对象，里面 data 数组的每一项都有 id.id 字段，这个设备内部 ID 在查遥测时要用到。
查遥测时把设备 ID 拼进 URL：
DEVICE_ID="上一步返回的设备ID" curl -s -X GET "http://localhost:9090/api/plugins/telemetry/DEVICE/${DEVICE_ID}/values/timeseries?keys=temperature&limit=10" \ -H "X-Authorization: Bearer $JWT"keys 参数指定要查的遥测字段，limit 控制返回条数。查询条件里还可以加 startTs 和 endTs 限定时间窗口，单位是毫秒时间戳。必须强调一个习惯：回读遥测永远带上时间窗口，别一次性拉全量。单机验证没问题，数据量上来后全量拉取会直接拖垮查询接口，这是第 5 章要讲的一个坑。到这里，一条「设备 -> MQTT -> 平台 -> REST 回读」的最小闭环就成立了。
4. 用规则链实现告警与第三方联动：RBI 的关键配置
4.1 把遥测消息分流到告警链
数据能进能出之后，开始给 RBI 加「脑子」。先说设计原则：不要在根规则链里堆几十个节点，一条链只干一件事。我常用的结构是根链做分流，子链做具体业务。根链上加一个 message type switch 节点，按 msgType 区分消息：POST_TELEMETRY_REQUEST 走遥测处理分支，POST_ATTRIBUTES_REQUEST 走属性处理分支。选 switch 而不是 script filter 的原因是：switch 按消息类型路由，语义清晰，后续加新设备类型不用改脚本；script filter 适合业务条件判断，放在子链里更灵活。
分流之后把遥测消息推到专门的告警处理链。这里的 push 不是自动发生的，需要显式加一个 push to chain 节点，指定目标链名字。很多新手以为建了新链就能生效，结果消息一路走到末端都没进过新链，原因就是缺了这个 push 动作。规则链的流转是「点对点」的，不会自动广播到所有链。
告警处理链的第一步通常是 script filter，把真正越限的样本筛出来：
// 过滤出有温度字段且超过 50 度的遥测消息 if (msg.temperature === undefined) { return false; } if (typeof msg.temperature === 'string') { msg.temperature = Number(msg.temperature); } return msg.temperature > 50;这个脚本里做了两件事：判断字段是否存在，避免把没有温度的设备消息也拉进来；把字符串类型的数值转成 number，因为很多设备固件会把数字序列化成字符串。类型转换不做，后面的阈值比较会得到错误结果。filter 节点返回 true 表示消息继续往下游走，false 表示该消息在这条分支上终止。
4.2 create alarm 与去重参数
筛选出的越限消息进入 create alarm 节点。这个节点的配置决定了告警是「一条」还是「一堆」，也是 RBI 方案里最容易被忽视的参数。核心配置如下：
{ "alarmType": "HIGH_TEMPERATURE", "severity": "CRITICAL", "propagate": true, "useAlarmDeduplication": true, "alarmDeduplicationPeriod": 60000 }alarmType 是告警类型标识，在同一台设备上会按类型区分不同的告警；severity 支持 CRITICAL、MAJOR、MINOR 等几个级别；propagate 决定告警是否向上传播到关联的 asset，实际项目中建议开启，这样产线大屏能看到产线级告警。useAlarmDeduplication 是关键开关，开启后同一个设备、同一种 alarmType 的告警在去重窗口内不会重复创建，而是更新已有告警的详情；alarmDeduplicationPeriod 配置窗口时长，单位毫秒。
窗口时长怎么定？我一般取「业务认为同一次故障的最长持续时间」。温度越限这类连续性故障，1 到 5 分钟的窗口比较合理；如果是发电机启动信号这类瞬态事件，窗口就要设短，否则会吞掉连续多次的启动告警。窗口太长漏报，窗口太短冗余告警，这两个方向在实际项目里都翻过车，多嘴提醒一句。
4.3 REST API Call 节点对接业务系统与失败消息处理
告警建好之后，RBI 最后一步是把告警同步给外部业务系统。常见的对接方式是 REST API Call 节点，直接向业务系统的接口发起 HTTP 请求：
{ "restEndpointUrl": "http://business-system.lan/api/alarms", "requestMethod": "POST", "headers": "{\"Content-Type\":\"application/json\"}", "body": "{\"deviceName\":\"${metadata.deviceName}\",\"alarmType\":\"HIGH_TEMPERATURE\",\"temperature\":${msg.temperature},\"ts\":${metadata.ts}}" }body 里用 ${metadata.deviceName} 和 ${msg.temperature} 这类模板变量，把设备名、数值动态填进请求体。这样外部系统拿到的不只是一条「有告警」的通知，还带上设备维度、数值维度的上下文，可以直接触发工单或联动控制逻辑。模板变量在节点配置里随处可见，看清楚 metadata 和 msg 的区别就不会填错位置。
注意：REST 节点有默认行为，目标接口超时或不可达时，消息会被直接丢弃，规则链不会自动重试。
生产环境必须处理失败分支。我的做法是给 REST 节点后面挂一个失败分支，把失败消息推到另一个队列链，由专门的链做定时重放；或者在业务系统侧保证接口幂等，用告警 ID 做去重键。接口幂等和失败重试要同时做，只做一个都不算闭环。告警已经写进平台数据库了，外部系统没收到，重放只是时间问题；接口不幂等，重放就会造成重复工单。这两个设计占了告警集成项目里一半的血泪经验。
5. ThingsBoard 集成避坑指南：五个高频故障与排查路径
5.1 设备已上线，遥测却查不到
现象：MQTT 客户端显示连接成功、publish 返回成功，设备页也显示在线，但「最新遥测」里就是没有数据。
原因按出现频率排：第一，把设备名称当成了 access token 用，或 token 复制时多了空格；第二，topic 写错，把 telemetry 写成了 attributes；第三，payload 不是合法 JSON，或者 JSON 是合法但 key 与查询时不一致；第四，消息进了规则链但被脚本 filter 悄悄过滤掉了。
解决路径从传输层往业务层查。先用官方 REST 接口直接上报一次，排除客户端问题：curl -X POST http://localhost:9090/api/v1/$ACCESS_TOKEN/telemetry --data '{"temperature":41.2}'。如果接口能查到数据，问题一定在客户端或规则链。接着看 docker compose logs -f tb，把日志级别调到 DEBUG，重点搜设备 token 关键字，传输层会打印收到的消息和校验结果。最后查看消息有没有进入规则链，打开规则链 debugMode，用同样的 payload 在测试面板里模拟一遍，看消息流到哪个节点断了。
5.2 同一条越限消息反复产生告警
现象：一次温度越限，告警列表里出现几十条 HIGH_TEMPERATURE，全部是同一个设备同一个时间段。
原因：create alarm 节点没有开启去重，或者去重窗口设得太短。默认情况下，每一条通过 filter 的遥测消息都会触发一次 create alarm 的「创建」动作。遥测上报频率是每分钟一次，去重窗口设了 30 秒，那窗口内第二条消息就会另起炉灶。
解决：把 useAlarmDeduplication 设为 true，按故障持续时长设 alarmDeduplicationPeriod。这里还要配合业务理解：去重只抑制「重复创建」，不会自动更新告警详情。想要告警详情关联最新一次越限数值，需要在前置脚本里把当前数值写进告警详情字段，create alarm 节点会保留这些字段。排查时先用 API 查一下同设备同告警的创建时间间隔，如果间隔稳定等于上报周期，基本就是去重没生效。
5.3 规则链改了不生效
现象：在界面里改了脚本阈值，保存后旧逻辑还在跑，新值完全没影响。
原因有三个常见来源：设备挂在设备 profile 的默认链上，而你改的是租户根链或另一条自定义链；规则链确实改了，但保存的是草稿没点确认发布；浏览器端调试面板和实际运行的节点版本不一致。第三个最容易让人误判，因为调试面板里跑的是你当前编辑的这版脚本，而线上流量跑的是上次保存的版本。
解决：先确认设备详情页的 profile 信息，再进对应 profile 查看绑定的规则链名称，保证改的是同一条链。修改完成后保存，回规则链列表看版本号和更新时间是否变化。调试时用「测试」按钮模拟消息之前，先看一眼节点的实际配置，别拿编辑区脚本去猜线上行为。实在排查不清，重启一次 tb 容器清掉规则引擎缓存，但这只是排查手段，别当成常态操作。
5.4 时序数据回读越来越慢
现象：同样一条查遥测的 REST 请求，刚开始 100 毫秒返回，三个月后要好几秒，数据库磁盘占用也持续上涨。
原因：ThingsBoard 默认不会无限期保留遥测，但也不会自动按业务需求清理，数据量过千万行后，单表查询性能会肉眼可见地退化。尤其在纯 PostgreSQL 模式下，时序数据与业务数据共用存储，膨胀速度比混合模式更快。
解决：从两个方向着手。一是查询侧，所有回读请求都带 startTs/endTs 时间窗口，并且按需加 limit，不写全量拉取的代码。二是存储侧，定期清理过期数据，用官方 REST 删除接口按时间窗口批量删除；数据量预期超过几十亿行时，迁移到 Cassandra 混合模式会更扛时序写入。我的习惯是先把保留周期写进方案文档，比如「遥测保留 6 个月，属性保留 2 年」，然后让定时任务按这个策略执行，而不是等慢到报警再处理。
5.5 外部回调偶发失败导致丢失告警
现象：业务系统偶尔收不到告警，规则链日志里 REST API Call 节点报超时或连接拒绝，但平台里告警记录是存在的。
原因：REST 节点默认失败即丢，加上目标系统在发布窗口或瞬时高负载时响应慢，超时一过消息就没了。这是集成场景的高频事故，而且不重试的话很难发现，因为平台侧不会主动报错。
解决：给 REST 节点配置合理的超时时间，并利用失败分支。常见做法是 REST 节点后面连一个失败分支，失败消息推到独立的「重试队列」规则链，这条链延迟重放，重试三次后仍失败就写入本地日志文件或生成一条内部告警。同时要求业务系统接口做幂等，用平台告警 ID 作为去重键收消息。判断幂等实现得对不对，就看同一告警 ID 连续投递两次，业务系统会不会生成两条工单。这一条放到最后讲，因为它是从「能跑」到「可靠」的分水岭，也是 RBI 方案在生产环境里最容易被验收方追问的细节。
6. 进阶技巧：把 RBI 从「演示可用」做到「生产可用」
6.1 用设备 profile 统一规则链与告警配置
设备多了之后，别给每台设备单独挂规则链。在设备 profile 里把规则链、告警配置、传输配置固化成模板，同类设备创建时自动继承。这样新增 100 台设备不用重配，改阈值也只需要动一处，不容易出现「两台设备一套逻辑各改各的」的问题。
6.2 告警恢复闭环：clear alarm 自动收尾
生产环境要的是「告警 -> 恢复 -> 通知」完整闭环。在告警链的末端加一个分支：当温度回落小于 50 时走 clear alarm 节点，清掉对应设备同类型的 active 告警，再调一次 REST 接口通知业务系统「告警已恢复」。闭环后工单系统才能自动关闭，否则告警会一直挂着，运维很快对告警列表失去信任。
6.3 链路自检脚本与日常验证习惯
最后写一个小的 Python 自检脚本，把登录、查设备、查遥测、查告警串起来，作为每次改完配置后的回归检查。脚本不需要复杂，能验证「数据进得来、告警建得起、结果查得到」三个核心链路就够。
import requests base = "http://localhost:9090" jwt = requests.post(base + "/api/auth/login", json={"username": "tenant@thingsboard.org", "password": "tenant"}).json()["token"] headers = {"X-Authorization": f"Bearer {jwt}"} devices = requests.get(base + "/api/tenant/devices?pageSize=1&page=0", headers=headers).json()["data"] assert devices, "no device" device_id = devices[0]["id"]["id"] resp = requests.get( base + f"/api/plugins/telemetry/DEVICE/{device_id}/values/timeseries?keys=temperature&limit=1", headers=headers).json() assert "temperature" in resp, "telemetry missing" print("RBI check passed")我最早做这套方案时，图省事把判断逻辑全塞在根规则链里，几十个节点纠缠在一起，每次改动都要胆战心惊地确认会不会影响其它设备。后来养成的习惯是：一条链只干一件事，每改一个节点先在测试面板用模拟消息过一遍再保存，最后跑一遍自检脚本才敢说「完成了」。这套节奏看着慢，实际省掉的是半夜爬起来查告警丢失的时间。希望帮到你。
本文还有配套的精品资源，点击获取
