ThingsBoard消息优先级3层机制让高优先级消息跑在前面【免费下载链接】thingsboardOpen-source IoT Platform - Device management, data collection, processing and visualization.项目地址: https://gitcode.com/GitHub_Trending/th/thingsboard半夜两点温度传感器的告警消息卡在一堆常规上报数据后面大屏上的红色告警迟到了将近一分钟。你翻源码想找个priority字段去调结果发现 ThingsBoard 里根本没有 int 型的优先级字段——它给紧急消息直接开了一条独立的高优先级车道。整篇文章就拆两块先讲清楚优先级这东西在 ThingsBoard 里到底落在哪一层再带你用 3 步把这条高优先级车道实际跑起来。一条紧急消息是怎么被单独放行的打个比方医院不会给每个病人贴个紧急程度 0-10的牌子让护士在一支队伍里排序而是直接开一条急诊通道——排队短、响应快、不会被常规体检堵住。ThingsBoard 的思路一样优先级不是元数据而是你在哪个队列。优先级标记藏在哪消息体上的队列名先说结论优先级标识就是TbMsg上的queueName字段。它被序列化进 protobuf 消息体跟着消息走// common/proto/src/main/proto/queue.proto // 第 506/517/528 行各消息定义里都带着这个字段 string queueName 5;TbMsg构建时就能带上它比如规则引擎上下文里由设备档案决定目标队列// application/.../actors/ruleChain/DefaultTbContext.java defaultQueueName profile.getDefaultQueueName(); // 档案里配的目标队列 return TbMsg.newMsg() .queueName(defaultQueueName) .type(actionMsgType) .build();所以这条消息急不急在消息出生的那一刻就写死在路由信息里了后续每个服务只认这个字段。高优先级车道安装时自动开出来真正的车道是系统租户下的一个专用队列名字和主题都是常量// common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java public static final String HP_QUEUE_NAME HighPriority; public static final String HP_QUEUE_TOPIC tb_rule_engine.hp;安装初始化时DefaultSystemDataLoaderService会把它建出来注意几个关键参数这就是高优先级的物理含义——轮询更勤、每分区独立消费者、突发小批量提交// application/.../service/install/DefaultSystemDataLoaderService.java (节选) highPriorityQueue.setTopic(DataConstants.HP_QUEUE_TOPIC); highPriorityQueue.setPollInterval(25); // 25ms 轮询常规队列没这么勤 highPriorityQueue.setPartitions(10); highPriorityQueue.setConsumerPerPartition(true); // 每分区一个消费者 SubmitStrategy s new SubmitStrategy(); s.setType(SubmitStrategyType.BURST); // 突发小批量低延迟 s.setBatchSize(100);消费侧也有对应入口TbCoreQueueFactory接口上专门留了一个高优先级消费者工厂方法// common/queue/.../provider/TbCoreQueueFactory.java (Javadoc 原文) // Used to consume high priority messages by TB Core Service TbQueueConsumerTbProtoQueueMsgToCoreNotificationMsg createToCoreNotificationsMsgConsumer();再看边缘场景EdgeGrpcSession干脆维护了内存里的双队列普通事件走一个队列高优先级事件单独进ConcurrentLinkedQueue每次发送前先把高优队列掏空// application/.../edge/rpc/session/EdgeGrpcSession.java (节选) if (highPriorityQueue.isEmpty()) { /* 转去普通队列 */ } while ((event highPriorityQueue.poll()) ! null) { highPriorityEvents.add(event); // 高优事件先打包发走 }把整条链路串起来看从配置到生效3步跑通 HighPriority第 1 步确认高优先级队列存在全新安装会自动创建。存量环境可以用 REST API 建一个规则引擎服务类型的队列body 参考安装逻辑里的字段topic 用tb_rule_engine.hp、pollInterval25、consumerPerPartitiontrue、BURST 提交策略。建好后在系统租户的队列列表里能看到HighPriority这个名字。第 2 步把设备档案的默认队列指过去优先级是配在档案上的不是设备实例上。在设备/资产档案里设置defaultQueueName为HighPriorityUI 上即设备档案的默认队列下拉。之后该档案下所有设备产生的消息都会在DefaultTbClusterService里被改写路由// application/.../service/queue/DefaultTbClusterService.java (节选) String targetQueueName useQueueFromTbMsg ? tbMsg.getQueueName() : ruleEngineProfile.getDefaultQueueName(); if (isQueueTransform) { tbMsg tbMsg.transform(targetQueueName); // 重写消息目标队列 }第 3 步发一条遥测验证它走对了车道给该档案下的设备发属性/遥测然后确认消息落到了tb_rule_engine.hp这个 topic 上。仓库里现成的可运行参照是TenantControllerTest它创建了一个 HighPriority profile发属性更新后断言usedTpi.getTopic()等于HP_QUEUE_TOPIC。照着这个断言思路自查比肉眼看日志快。这些坑我替你踩过了⚠️defaultQueueName 配了却不生效。现象档案指向 HighPriority消息还是落在默认 topic。 原因transformMsg里useQueueFromTbMsg为真时消息自带的queueName优先档案配置直接被跳过。 解法先确认消息源头是谁塞的queueName规则节点、边缘处理器都可能别默认档案说了算。边缘高优队列悄悄丢弃最老消息。现象边缘网关高负载时个别高优先级事件凭空消失日志里没有报错。 原因EdgeGrpcSession高优队列有上限thingsboard.yml里默认max_high_priority_queue_size_per_session10000超了就直接poll()丢掉最老的。 解法按实际并发调EDGES_MAX_HIGH_PRIORITY_QUEUE_SIZE_PER_SESSION环境变量并对丢弃行为加监控告警别等对账才发现。把优先级理解成排序。现象以为往同一个 topic 里发消息、带上高优标记就能插队结果延迟毫无改善。 原因Kafka 单个 topic 内部没有优先级排序ThingsBoard 的优先级是物理分车道——走错 topic 不是慢一点是完全不进高优通道。 解法验证永远看 topic 归属第 3 步的断言法另外consumerPerPartitiontrue意味着 10 个分区要 10 个消费者才吃得满规则引擎实例不够时高优车道自己也会堵。延伸参考common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java各服务生产者/消费者工厂的总目录找高优消费者入口从这开始application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java消息按档案改写队列名的核心逻辑在这common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.javaHighPriority队列名与tb_rule_engine.hp主题的定义处application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java高优队列的完整初始化参数轮询、分区、提交策略application/src/test/java/org/thingsboard/server/controller/TenantControllerTest.java验证消息确实走对了 topic的可运行测试样例【免费下载链接】thingsboardOpen-source IoT Platform - Device management, data collection, processing and visualization.项目地址: https://gitcode.com/GitHub_Trending/th/thingsboard创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ThingsBoard消息优先级：3层机制让高优先级消息跑在前面
ThingsBoard消息优先级3层机制让高优先级消息跑在前面【免费下载链接】thingsboardOpen-source IoT Platform - Device management, data collection, processing and visualization.项目地址: https://gitcode.com/GitHub_Trending/th/thingsboard半夜两点温度传感器的告警消息卡在一堆常规上报数据后面大屏上的红色告警迟到了将近一分钟。你翻源码想找个priority字段去调结果发现 ThingsBoard 里根本没有 int 型的优先级字段——它给紧急消息直接开了一条独立的高优先级车道。整篇文章就拆两块先讲清楚优先级这东西在 ThingsBoard 里到底落在哪一层再带你用 3 步把这条高优先级车道实际跑起来。一条紧急消息是怎么被单独放行的打个比方医院不会给每个病人贴个紧急程度 0-10的牌子让护士在一支队伍里排序而是直接开一条急诊通道——排队短、响应快、不会被常规体检堵住。ThingsBoard 的思路一样优先级不是元数据而是你在哪个队列。优先级标记藏在哪消息体上的队列名先说结论优先级标识就是TbMsg上的queueName字段。它被序列化进 protobuf 消息体跟着消息走// common/proto/src/main/proto/queue.proto // 第 506/517/528 行各消息定义里都带着这个字段 string queueName 5;TbMsg构建时就能带上它比如规则引擎上下文里由设备档案决定目标队列// application/.../actors/ruleChain/DefaultTbContext.java defaultQueueName profile.getDefaultQueueName(); // 档案里配的目标队列 return TbMsg.newMsg() .queueName(defaultQueueName) .type(actionMsgType) .build();所以这条消息急不急在消息出生的那一刻就写死在路由信息里了后续每个服务只认这个字段。高优先级车道安装时自动开出来真正的车道是系统租户下的一个专用队列名字和主题都是常量// common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java public static final String HP_QUEUE_NAME HighPriority; public static final String HP_QUEUE_TOPIC tb_rule_engine.hp;安装初始化时DefaultSystemDataLoaderService会把它建出来注意几个关键参数这就是高优先级的物理含义——轮询更勤、每分区独立消费者、突发小批量提交// application/.../service/install/DefaultSystemDataLoaderService.java (节选) highPriorityQueue.setTopic(DataConstants.HP_QUEUE_TOPIC); highPriorityQueue.setPollInterval(25); // 25ms 轮询常规队列没这么勤 highPriorityQueue.setPartitions(10); highPriorityQueue.setConsumerPerPartition(true); // 每分区一个消费者 SubmitStrategy s new SubmitStrategy(); s.setType(SubmitStrategyType.BURST); // 突发小批量低延迟 s.setBatchSize(100);消费侧也有对应入口TbCoreQueueFactory接口上专门留了一个高优先级消费者工厂方法// common/queue/.../provider/TbCoreQueueFactory.java (Javadoc 原文) // Used to consume high priority messages by TB Core Service TbQueueConsumerTbProtoQueueMsgToCoreNotificationMsg createToCoreNotificationsMsgConsumer();再看边缘场景EdgeGrpcSession干脆维护了内存里的双队列普通事件走一个队列高优先级事件单独进ConcurrentLinkedQueue每次发送前先把高优队列掏空// application/.../edge/rpc/session/EdgeGrpcSession.java (节选) if (highPriorityQueue.isEmpty()) { /* 转去普通队列 */ } while ((event highPriorityQueue.poll()) ! null) { highPriorityEvents.add(event); // 高优事件先打包发走 }把整条链路串起来看从配置到生效3步跑通 HighPriority第 1 步确认高优先级队列存在全新安装会自动创建。存量环境可以用 REST API 建一个规则引擎服务类型的队列body 参考安装逻辑里的字段topic 用tb_rule_engine.hp、pollInterval25、consumerPerPartitiontrue、BURST 提交策略。建好后在系统租户的队列列表里能看到HighPriority这个名字。第 2 步把设备档案的默认队列指过去优先级是配在档案上的不是设备实例上。在设备/资产档案里设置defaultQueueName为HighPriorityUI 上即设备档案的默认队列下拉。之后该档案下所有设备产生的消息都会在DefaultTbClusterService里被改写路由// application/.../service/queue/DefaultTbClusterService.java (节选) String targetQueueName useQueueFromTbMsg ? tbMsg.getQueueName() : ruleEngineProfile.getDefaultQueueName(); if (isQueueTransform) { tbMsg tbMsg.transform(targetQueueName); // 重写消息目标队列 }第 3 步发一条遥测验证它走对了车道给该档案下的设备发属性/遥测然后确认消息落到了tb_rule_engine.hp这个 topic 上。仓库里现成的可运行参照是TenantControllerTest它创建了一个 HighPriority profile发属性更新后断言usedTpi.getTopic()等于HP_QUEUE_TOPIC。照着这个断言思路自查比肉眼看日志快。这些坑我替你踩过了⚠️defaultQueueName 配了却不生效。现象档案指向 HighPriority消息还是落在默认 topic。 原因transformMsg里useQueueFromTbMsg为真时消息自带的queueName优先档案配置直接被跳过。 解法先确认消息源头是谁塞的queueName规则节点、边缘处理器都可能别默认档案说了算。边缘高优队列悄悄丢弃最老消息。现象边缘网关高负载时个别高优先级事件凭空消失日志里没有报错。 原因EdgeGrpcSession高优队列有上限thingsboard.yml里默认max_high_priority_queue_size_per_session10000超了就直接poll()丢掉最老的。 解法按实际并发调EDGES_MAX_HIGH_PRIORITY_QUEUE_SIZE_PER_SESSION环境变量并对丢弃行为加监控告警别等对账才发现。把优先级理解成排序。现象以为往同一个 topic 里发消息、带上高优标记就能插队结果延迟毫无改善。 原因Kafka 单个 topic 内部没有优先级排序ThingsBoard 的优先级是物理分车道——走错 topic 不是慢一点是完全不进高优通道。 解法验证永远看 topic 归属第 3 步的断言法另外consumerPerPartitiontrue意味着 10 个分区要 10 个消费者才吃得满规则引擎实例不够时高优车道自己也会堵。延伸参考common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java各服务生产者/消费者工厂的总目录找高优消费者入口从这开始application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java消息按档案改写队列名的核心逻辑在这common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.javaHighPriority队列名与tb_rule_engine.hp主题的定义处application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java高优队列的完整初始化参数轮询、分区、提交策略application/src/test/java/org/thingsboard/server/controller/TenantControllerTest.java验证消息确实走对了 topic的可运行测试样例【免费下载链接】thingsboardOpen-source IoT Platform - Device management, data collection, processing and visualization.项目地址: https://gitcode.com/GitHub_Trending/th/thingsboard创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考