简介RBI_Thingsboard是一套基于C实现的物联网平台完整源码面向具备一定编程基础、希望深入理解IoT平台内部机制的开发者和学习者。项目围绕设备注册管理、数据收集、实时监控与历史分析等核心模块展开清晰展示了C语言在应对高并发连接和高速数据流时的性能优势。压缩包内共收录83个文件主体为28个cpp源文件和18个h头文件并配套pro/pri工程配置、Makefile构建脚本以及编译生成的o目标文件方便直接编译学习另有qcustomplot可视化组件、串口通信相关代码和两份PDF说明文档可对照理解界面与功能设计。资源包整体约1016KB轻量而完整。当前已有191人学习下载。通过通读源码可以掌握C系统级编程、模板与面向对象设计在模块化架构中的应用同时学习qextserialport串口通信、MQTT协议接入等典型物联网场景的具体实现还能体会跨平台部署在多样化硬件环境中的处理思路对提升C实践能力和物联网平台设计认知都很有价值。1. 项目定位设备接入层的C高吞吐网关某次在产线上调设备接入Java 写的接入网关在千台设备同时上报时 CPU 直接被打满消息积压越来越严重而底层 ThingsBoard 平台本身并没有问题——瓶颈全在协议解析和会话管理上。后来拿到这份 RBI_Thingsboard 源码包思路一下就打开了它把设备接入这一层单独拆出来用 C 重写只负责「收报文、解协议、过滤规则、转发给 ThingsBoard」平台侧的数据存储和可视化完全不动。RBI 在这里可以理解为规则接入层Rule-Based Integration源码里没有专门文档解释缩写但整个工程的核心就是围绕「规则过滤 协议接入」展开的。适合谁凡是设备端协议杂、上报频率高、Java 网关扛不住吞吐的团队都可以把这份工程当作接入层的替换方案来读。2. 架构设计C17 接入层与 ThingsBoard 桥接的边界2.1 整体分层接入层、规则层、转发层这份工程没有去动 ThingsBoard 本身而是把它当作一个「数据下游」。整套代码按功能可以切成三个层次每一层只依赖下一层暴露的接口不跨层调用。接入层负责跟设备打交道监听 TCP/UDP 端口、解析 MQTT/CoAP 报文、维护设备会话状态。规则层是 RBI 的核心含义所在——对解析出来的数据进行关键字匹配、阈值过滤、格式整形决定这条消息值不值得往上送。转发层最薄只做一件事把通过规则的数据用 MQTT 协议发布到 ThingsBoard 的 broker 上。设备端MQTT/私有TCP ↓ 接入层 session 管理、协议解析 ↓ 规则层 关键字过滤、阈值判断、字段重命名 ↓ 转发层 MQTT publish → ThingsBoard这个分层的价值在于如果设备协议变了只需要换接入层的解析器如果平台从 ThingsBoard 换成了别的物联网平台只需要换转发层的客户端实现规则层完全不用动。我在改这个项目时最满意的也是这一点——它把「协议适配」和「业务规则」两件事彻底解耦了。2.2 技术选型C17 Boost.Asio MQTT为什么这么定整个工程用 C17 编写网络层基于 Boost.AsioMQTT 客户端用的是 libmosquittoJSON 解析走的是 nlohmann/json 单头库。这套组合在嵌入式网关和边缘服务器上都非常常见依赖少、交叉编译友好。选择 C17 而不是 C11主要是为了std::string_view和std::optional。设备协议解析里大量涉及从缓冲区切字段string_view可以避免不必要的字符串拷贝字段缺省时的处理用optional表达比裸指针安全得多。Boost.Asio 负责异步读写配合io_context跑多线程事件循环解决一千个设备同时维持长连接时的线程开销问题。libmosquitto 在这里不是作为完整 broker 使用而是作为 MQTT 客户端库把过滤后的数据发布到 ThingsBoard 自带的 MQTT Transport。它提供了同步和异步两套 API这个项目里用的是同步发布因为转发层不需要高并发——真正的高并发在接入层已经被消化掉了。2.3 目录结构源码包一进来先看哪几个文件工程目录不算复杂但第一次看容易找不着北。我建议按下面的顺序去读而不是按字母序。rbi_thingsboard/ ├── CMakeLists.txt # 顶层构建脚本依赖查找都在这 ├── src/ │ ├── main.cpp # 入口加载配置、启动 io_context 线程池 │ ├── config/ # 配置加载与运行时参数 │ │ ├── rbi_config.h │ │ └── rbi_config.cpp │ ├── session/ # 设备会话管理 │ │ ├── device_session.h │ │ └── session_manager.h │ ├── rule/ # RBI 规则引擎 │ │ ├── rule_engine.h │ │ └── rule_engine.cpp │ └── forward/ # 转发层 │ ├── mqtt_forwarder.h │ └── mqtt_forwarder.cpp ├── proto/ # 内置协议定义MQTT、样例私有协议 ├── conf/ │ └── rbi_config.json # 运行配置模板 └── tests/ └── unit/ # 规则引擎单测src/config是整个工程最先被执行的部分配置一旦读错后面全部白搭。src/rule是 RBI 的核心二次开发时改得最多的就是这一层。src/forward最简单甚至可以暂时不看。2.4 数据链路一条报文从设备到平台要过几道手一条 MQTT 报文从设备发出到进入 ThingsBoard在工程内部要经历四道工序。第一道是「连接接入」设备通过 MQTT 协议连上接入层的监听端口session_manager为这个设备建立会话记录 clientId 和连接时间。第二道是「协议解析」报文 payload 被 JSON 解析器拆成结构化字段。第三道是「规则过滤」rule_engine按设备类型和消息主题匹配规则不满足规则的消息直接丢弃不再往下走。第四道是「转发」mqtt_forwarder把消息重新打包成 ThingsBoard 认识的 MQTT 主题和数据格式发布出去。每道工序之间通过std::shared_ptr传递消息对象避免大对象多次拷贝。设备接入边界上用的是异步读规则层是纯 CPU 计算转发层是阻塞式发布——这三者的速度天然不匹配所以接入层和规则层之间放了一个无界队列缓冲。队列在内存里不做持久化如果进程崩溃未转发的数据会丢这点在部署时要心里有数。3. 核心模块拆解一条消息从接收到转发的完整代码路径3.1 配置加载JSON 配置与默认值兜底配置加载用的是 nlohmann/json 的parse和value接口核心逻辑在rbi_config.cpp里。这个类设计上有一个比较实用的点每个配置项都有默认值配置文件里缺字段不会报错而是回退到默认值日志里打一条 warning。// src/config/rbi_config.cpp #include rbi_config.h #include nlohmann/json.hpp #include fstream #include iostream namespace rbi { bool RbiConfig::load(const std::string path) { std::ifstream ifs(path); if (!ifs.is_open()) { std::cerr config file not found: path std::endl; return false; } nlohmann::json j; ifs j; // 监听端口设备接入层的 TCP 端口默认 1885 listen_port j.value(listen_port, 1885); // 线程池大小io_context 的工作线程数默认等于 CPU 核数 worker_threads j.value(worker_threads, std::thread::hardware_concurrency()); // 转发目标ThingsBoard MQTT broker 的地址 tb_host j.value(tb_host, 127.0.0.1); tb_port j.value(tb_port, 1883); // 设备默认上报主题规则层会按这个主题前缀匹配 tb_topic j.value(tb_topic, v1/devices/me/telemetry); // 规则开关是否启用关键字过滤 rule_enabled j.value(rule_enabled, true); // 规则表数组形式每个元素是一个独立的过滤规则 if (j.contains(rules)) { for (const auto item : j[rules]) { RuleConfig rc; rc.name item.value(name, unnamed); rc.device_type item.value(device_type, ); rc.allow_keywords item.value(allow_keywords, std::vectorstd::string{}); rules.push_back(std::move(rc)); } } return true; } } // namespace rbi这里j.value(listen_port, 1885)是 nlohmann/json 提供的带默认值读取接口第一个参数是 JSON 字段名第二个是默认值。字段缺失或类型不匹配时不会抛异常而是返回默认值。规则表rules用contains做了存在性检查避免对不存在的数组做遍历。要注意的是worker_threads这里用了hardware_concurrency()作为默认值在容器环境里它返回的是宿主机的核数而不是容器分配到的核数。如果部署在 2 核的容器里但宿主机是 32 核线程池会创建 32 个线程反而增加上下文切换开销。我一般会在部署时显式在配置文件里写死这个值。3.2 会话管理高并发下哈希表与读写锁设备连接建立后会话信息要能在任意工作线程里被查到。session_manager用std::unordered_map存放会话key 是设备 clientIdvalue 是会话对象。因为会有多线程同时读、单线程增删改这里用读写锁保护而不是直接用std::mutex。// src/session/session_manager.h #pragma once #include device_session.h #include unordered_map #include shared_mutex #include string namespace rbi { class SessionManager { public: // 会话不存在时返回 nullptr调用方需要自行判断 std::shared_ptrDeviceSession find(const std::string client_id) { std::shared_lock lock(mutex_); auto it sessions_.find(client_id); if (it sessions_.end()) return nullptr; return it-second; } // 新设备接入时调用若已存在同名会话后者覆盖前者 void add(const std::string client_id, std::shared_ptrDeviceSession session) { std::unique_lock lock(mutex_); sessions_[client_id] std::move(session); } // 设备断开时移除会话避免内存泄漏 void remove(const std::string client_id) { std::unique_lock lock(mutex_); sessions_.erase(client_id); } size_t size() { std::shared_lock lock(mutex_); return sessions_.size(); } private: mutable std::shared_mutex mutex_; std::unordered_mapstd::string, std::shared_ptrDeviceSession sessions_; }; } // namespace rbi读写锁的思路是find是高频操作多个工作线程同时查同一个会话列表完全没有冲突用共享锁add和remove是低频操作才需要独占锁。这里std::shared_lock和std::unique_lock配合std::shared_mutex是 C17 的标准写法不用自己造读写锁。DeviceSession内部存了设备的最近一次心跳时间、会话创建时间、以及一个协议解析器指针。心跳时间的更新频率很高如果也放进锁保护的结构里会和会话表的锁打架。我的做法是在DeviceSession内部单独用一个原子变量存last_seen不参与哈希表的锁粒度这样读写锁的临界区尽量小。3.3 RBI 规则引擎关键字过滤与消息整形规则引擎是整个工程里最有含金量的部分。它做的事情是把一条 JSON 消息根据设备类型和内容决定「是否放行」以及「是否改写字段」。我摘录了核心的match和apply两个方法。// src/rule/rule_engine.cpp #include rule_engine.h #include algorithm namespace rbi { bool RuleEngine::match(const std::string device_type, const nlohmann::json msg) const { if (!enabled_) return true; for (const auto rule : rules_) { // 设备类型不匹配则跳过这条规则 if (!rule.device_type.empty() rule.device_type ! device_type) { continue; } // 规则里配置了 allow_keywords消息里必须至少包含其中一个 if (!rule.allow_keywords.empty()) { bool hit false; for (const auto kw : rule.allow_keywords) { // 把消息序列化成字符串做子串匹配简单粗暴但有效 if (msg.dump().find(kw) ! std::string::npos) { hit true; break; } } if (!hit) return false; } return true; // 规则命中放行 } return false; } nlohmann::json RuleEngine::apply(const nlohmann::json msg) const { nlohmann::json out msg; // 常见做法自动补充服务端时间戳避免设备时钟偏差导致时序错乱 out[ts] std::chrono::duration_caststd::chrono::milliseconds( std::chrono::system_clock::now().time_since_epoch()).count(); // 统一把原始 topic 字段压缩掉减少上行流量 out.erase(raw_topic); return out; } } // namespace rbimsg.dump().find(kw)这种实现方式有个明显的问题每次都把整个 JSON 序列化成字符串再匹配性能开销不小。如果设备上报频率极高这一步会成为新的瓶颈。但这个写法在规则数量少个位数、消息体小几百字节的场景下完全够用而且胜在规则配置灵活——不用规定 JSON 里哪个字段才能匹配任意位置出现关键词都算命中。apply方法做的事情是「整形」给消息加ts时间戳、删掉raw_topic字段。这里使用out msg先做一次深拷贝防止后面的修改影响上层持有的原始消息。如果你的设备消息体很大这个拷贝会有明显开销可以改成nlohmann::json的移动语义或者直接原地修改。3.4 转发器libmosquitto 发布到 ThingsBoard转发层用 libmosquitto 的同步 API代码很短但有一个值得注意的细节发布失败时的重连策略。// src/forward/mqtt_forwarder.cpp #include mqtt_forwarder.h #include mosquitto.h #include cstring #include iostream namespace rbi { bool MqttForwarder::publish(const std::string topic, const nlohmann::json payload) { if (!connected_) { // 尝试重连重连失败直接丢弃这条消息并打错误日志 if (mosquitto_reconnect(mosq_) ! MOSQ_ERR_SUCCESS) { std::cerr reconnect to ThingsBoard broker failed std::endl; return false; } connected_ true; } std::string body payload.dump(); int rc mosquitto_publish(mosq_, nullptr, // 不需要 mid topic.c_str(), // 目标主题 body.size(), body.c_str(), 1, // QoS 1 false); // 不保留消息 if (rc ! MOSQ_ERR_SUCCESS) { std::cerr publish failed, code rc std::endl; connected_ false; return false; } return true; } } // namespace rbimosquitto_publish的第三个参数topic.c_str()传的是 ThingsBoard 的遥测主题默认是v1/devices/me/telemetry。第四个参数是 payload 长度第五个是数据指针。这里 QoS 用 1保证消息至少送达一次如果 ThingsBoard broker 恰好在这时挂了消息会在本地堆积并重试。重连策略这里有个取舍mosquitto_reconnect是阻塞调用在 broker 不可达时会一直卡住导致转发线程被堵死。如果对实时性要求高应该把转发放到独立线程或者改用mosquitto_publish_async配合回调。这个工程选择同步重连图的是实现简单、行为可预测——转发线程本来就是低速链路阻塞一会儿不会影响接入层。4. 编译配置与运行从源码到连上 ThingsBoard4.1 编译环境CMake 与依赖库准备编译需要三样东西CMake 3.16 以上、支持 C17 的编译器GCC 8 或 Clang 7、系统里装好 libmosquitto-dev 和 libboost-dev。nlohmann/json 是单头库CMake 会用FetchContent自动拉取不需要手动安装。我一般在 Ubuntu 上用下面的命令准备环境其他发行版换对应的包管理器即可sudo apt update sudo apt install -y cmake g libmosquitto-dev libboost-system-dev libboost-thread-devlibmosquitto-dev提供头文件和动态库libboost-system-dev和libboost-thread-dev是 Boost.Asio 的运行时依赖。如果只装libboost-dev而没装 system 库链接阶段会报找不到boost::system的符号。Ninja 虽然比 Make 快但这个项目规模不大Make 足够。4.2 编译三步走编译流程是标准的 CMake 三步没有任何花活mkdir -p build cd build cmake .. -DCMAKE_BUILD_TYPERelease make -j$(nproc)-DCMAKE_BUILD_TYPERelease会开启-O2优化这对接入层的吞吐影响很大。Debug 模式跑出来的性能大约只有 Release 的一半做性能验证时务必用 Release。如果cmake ..时 FetchContent 下载 nlohmann/json 卡住八成是网络问题。解决办法是提前下载好头文件把NLOHMANN_JSON_INCLUDE_DIR通过-D参数指到本地路径或者在 CMakeLists 里换成find_package(nlohmann_json REQUIRED)优先找系统安装的版本。4.3 配置文件参数说明编译产物rbi_gateway启动后会读取conf/rbi_config.json核心参数如下参数类型默认值说明listen_portint1885设备接入层监听端口与 ThingsBoard 的 1883 区分开worker_threadsintCPU核数io_context 工作线程数容器环境务必显式指定tb_hoststring127.0.0.1ThingsBoard MQTT broker 地址tb_portint1883ThingsBoard MQTT broker 端口tb_topicstringv1/devices/me/telemetry遥测数据上报主题rule_enabledbooltrue总开关false 时规则层全部放行rulesarray空规则列表每项含 name/device_type/allow_keywordslisten_port和tb_port不能配成同一个否则设备流量会被直接转发给自己形成死循环。worker_threads在容器里必须写死这是我在线上踩过的坑后面避坑章节会详细说。4.4 启动与最小验证启动分三步先确保 ThingsBoard 的 MQTT broker 在监听再启动穿透网关本身最后用一个 MQTT 客户端模拟设备发一条消息。./rbi_gateway -c ../conf/rbi_config.json看到[INFO] Rbi gateway started, listening on 1885之类的日志说明接入层已经起来了。接着用 mosquitto_pub 模拟一台设备mosquitto_pub -h 127.0.0.1 -p 1885 -t dev/thermo/001 \ -m {temp: 36.5, raw_topic: dev/thermo/001}这条消息会经过规则层。默认配置里如果没有给dev/thermo/001配置规则规则层会放行然后转发到 ThingsBoard。验证方式是在 ThingsBoard 的设备遥测页面看有没有新数据进来。如果没见到数据优先检查转发层的日志看publish failed的报错码——最常见的错误码是 3表示 broker 连接还没建立。5. 避坑实录接入层落地最常见的四个问题5.1 现象设备连接正常但消息积压越来越严重CPU 却不高原因某条设备的消息触发了规则层的msg.dump()序列化而这个设备的 payload 特别大比如带图片 base64 数据。dump()在大字符串上进行频繁匹配效率极低工作线程全卡在字符串操作上CPU 使用率上不去是因为它们都在等内存带宽。解决给allow_keywords匹配加一个前置条件——先检查 payload 长度超过 4KB 的消息直接走旁路不做关键字过滤。另一个方案是改成字段级匹配只对msg[type]这样的特定字段做getstd::string()匹配而不是对整个 JSON 做序列化。5.2 现象高并发下偶发崩溃core dump 栈指向unordered_map的迭代器原因SessionManager早期版本用的是std::mutex同步所有操作但find之后返回的std::shared_ptrDeviceSession在函数外部被释放而另一个线程的remove同时删除了同一个会话导致双重释放。解决这是典型的生命周期问题。改法是find返回std::shared_ptr的拷贝让引用计数接管最后一个持有者的释放同时remove里只做erase不主动delete。要确认这一点可以在DeviceSession的析构函数里加日志验证没有重复释放。5.3 现象ThingsBoard 上看到的遥测时间戳是乱序的原因设备上报时自带本地时间戳而各设备时钟没有同步有的快 5 分钟有的慢 10 分钟。转发层直接透传了设备的ts字段平台按这个时间戳排序自然出现乱序。解决规则层的apply方法里覆盖写入服务器统一时间戳也就是之前代码里out[ts] now()那一行。设备自己的时间戳可以保留但改名成device_ts。这样平台侧永远按服务器时间排序而需要设备时间做分析的人仍然能从device_ts拿到原始值。5.4 现象编译链接都成功运行时却报mosquitto库符号找不到原因系统里有多个版本的 libmosquitto。CMake 找到的是/usr/lib下的老版本头文件但运行时链接到的动态库是另一个路径下的新版本ABI 对不上。解决用ldd rbi_gateway查看实际链接的库路径确认是/usr/lib/x86_64-linux-gnu/libmosquitto.so.1还是别处的库。如果版本不一致最省事的办法是改成静态链接在 CMakeLists 里把mosquitto替换成/usr/lib/x86_64-linux-gnu/libmosquitto.a同时加上-lpthread -ldl。静态链接虽然体积大一点但能彻底规避这类玄学问题。6. 二次开发实战接入一种私有 TCP 协议真正的定制需求往往不是调参数而是加协议。假设现场有一批老设备只支持私有 TCP 协议——报文是定长二进制帧头0xAA55中间 4 字节是温度值末字节是校验。要在 RBI_Thingsboard 里接入它只需三步。第一步在proto/private_tcp.h里定义一个解析器继承接入层的解析接口。接口只有一个方法parse(const char* data, size_t len)返回一个nlohmann::json。把这个解析器的实例挂到DeviceSession上接入层收到数据后会自动调用它。第二步在监听端口上做分发。接入层支持按端口分配协议1885 端口跑 MQTT1886 端口跑私有 TCP。在main.cpp里加一个listen_port到协议工厂的映射std::unordered_mapint, std::functionParserPtr()启动时注册进去即可。第三步验证。用一个 socket 脚本往 1886 端口发AA5501F40100A9然后看转发日志里有没有对应的 JSON 输出。验证前先把rule_enabled改成false排除规则层干扰——先证明链路通再开规则。我见过很多人二次开发时直接改parse方法把新协议写进 MQTT 解析器里结果 MQTT 那边的报文也被影响排查半天。从那以后我每次加新协议都强制走一遍「新端口 新解析器 独立验证」的流程先让专用解析器在自己端口上跑通再谈合并规则。这个习惯帮我挡住了不少低级故障希望帮到你。本文还有配套的精品资源点击获取