是别的分区节点处理的 ＞ 转发到正确分区

org.thingsboard.server.actors.ruleChain.RuleNodeActorMessageProcessor#putToNodePartition/*** 将消息路由到正确服务实例上的节点分区单例节点跨服务转发** 仅在 singletonModetrue 且当前服务不负责该节点分区时触发。* 典型场景集群部署中某个规则节点设为全局单例只在特定服务实例运行* 其他实例收到消息时需将消息转发过去。* 流程* 1. 拷贝消息并将 ruleNodeId 设为本节点 ID确保目标服务路由到正确节点* 2. 解析节点 ID 对应的 TPI目标分区该分区属于负责运行此单例节点的服务实例* 3. 序列化并推送到 Rule Engine 队列目标实例消费后再次进入 onRuleChainToRuleNodeMsg* 4. ack(source)释放当前消息的 callback 计数原消息在本实例已完成由目标实例重新追踪* // 消息处理完成后会重新路由回来参见 RuleChainActorMessageProcessor.pushToTarget/private void putToNodePartition(TbMsg source) {// 拷贝消息将 ruleNodeId 记录为本节点目标实例据此找到对应 RuleNodeActorTbMsg tbMsg TbMsg.newMsg(source, source.getQueueName(), source.getRuleChainId(), entityId);// 以节点 ID 为 originator 解析分区得到负责运行该单例节点的服务实例的分区TopicPartitionInfo tpi systemContext.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, ruleNode.getId());TransportProtos.ToRuleEngineMsg toQueueMsg TransportProtos.ToRuleEngineMsg.newBuilder().setTenantIdMSB(tenantId.getId().getMostSignificantBits()).setTenantIdLSB(tenantId.getId().getLeastSignificantBits()).setTbMsgProto(TbMsg.toProto(tbMsg)).build();// 推送到队列由目标分区的服务实例消费处理systemContext.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), toQueueMsg, null);// 原消息在本实例已完成流转转发出去调用 ack 释放 packCtx 计数defaultCtx.ack(source);}对于路由过来的节点/*处理节点向自身发送的延迟消息ctx.tellSelf(msg, delayMs)常用于需要定时重试或延迟执行的场景如等待设备响应后重新评估处理逻辑与 onRuleChainToRuleNodeMsg 相同但不需要分区检查必然在本节点*/public void onRuleToSelfMsg(RuleNodeToSelfMsg msg) throws Exception {checkComponentStateActive(msg.getMsg());TbMsg tbMsg msg.getMsg();int ruleNodeCount tbMsg.getAndIncrementRuleNodeCounter();var tenantProfileConfiguration getTenantProfileConfiguration();int maxRuleNodeExecutionsPerMessage tenantProfileConfiguration.getMaxRuleNodeExecsPerMessage();if (maxRuleNodeExecutionsPerMessage 0 || ruleNodeCount maxRuleNodeExecutionsPerMessage) {apiUsageClient.report(tenantId, tbMsg.getCustomerId(), ApiUsageRecordKey.RE_EXEC_COUNT);persistDebugInputIfAllowed(msg.getMsg(), “Self”);try {tbNode.onMsg(defaultCtx, msg.getMsg());} catch (Exception e) {defaultCtx.tellFailure(msg.getMsg(), e);}} else {tbMsg.getCallback().onFailure(new RuleNodeException(“Message is processed by more than maxRuleNodeExecutionsPerMessage rule nodes!”, ruleChainName, ruleNode));}}对于 Debugr 模式的 保存调试数据org.thingsboard.server.actors.ruleChain.DefaultTbContext#tellNext(org.thingsboard.server.common.msg.TbMsg, java.util.Setjava.lang.String)/*** 规则节点核心输出方法将消息按指定关系类型路由到下游节点* 由规则节点实现TbNode在完成业务逻辑后调用** 流程* 1. persistDebugOutputDebug 模式下持久化输出事件供前端调试界面展示* 2. callback.onProcessingEnd(ruleNodeId)通知 TbMsgPackProcessingContext 本节点处理完毕* 更新最后访问节点信息用于超时日志注意此时消息并未计入完成仍在 pendingMap 中* 3. tell RuleNodeToRuleChainTellNextMsg 到 RuleChainActor* 由 RuleChainActorMessageProcessor.onTellNext() 根据 relationTypes 查找下游节点并继续路由* 若没有下游节点 → 触发 callback.onSuccess()消息从 pendingMap 移除packCtx 计数减一*/Overridepublic void tellNext(TbMsg msg, Set relationTypes) {RuleNode ruleNode nodeCtx.getSelf();persistDebugOutput(msg, relationTypes);// 标记当前节点处理完毕更新诊断信息不触发 packCtx 计数msg.getCallback().onProcessingEnd(ruleNode.getId());// 将路由决策交还给规则链 Actor由其负责查找并分发到下游节点nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(ruleNode.getRuleChainId(), ruleNode.getId(), relationTypes, msg, null));}内存和Kafka队列关于 QueueFactory 选择由Spring 启动时注入Factory 条件InMemoryMonolithQueueFactory queue.typein-memory service.typemonolithKafkaMonolithQueueFactory queue.typekafka service.typemonolithKafkaTbRuleEngineQueueFactory queue.typekafka service.typetb-rule-engine自定义节点https://thingsboard.io/docs/user-guide/contribution/rule-node-development/核心概念在 ThingsBoard 中每条数据 → 被包装成 TbMsg每个节点 → 处理一条消息输出 → 发送给下一个节点带 relationThingsBoard 类比Rule Chain 工作流 DAGRule Node 函数节点TbMsg 消息对象relationType 分支条件节点生命周期init() → 初始化解析配置onMsg() → 处理消息核心写逻辑的地方destroy() → 销毁核心逻辑onMsg真正写业务逻辑的地方Overridepublic void onMsg(TbContext ctx, TbMsg msg) {// 处理 msg}ThingsBoard 不会自动继续流程必须手动ctx.tellSuccess(msg);或者ctx.tellNext(msg, “True”);另外, 还有两种流转ctx.tellSelf // 只用于延迟一段时间后叫醒自己1个/节点(单实例只有一个)ctx.enqueueForTellNext //它走的是持久化的消息队列Kafka/内存队列通常用于触发下游节点N个/触发批次(与设备数相关但异步消费)节点定义RuleNode 注解这是“声明元数据”类似前端组件注册/**Copyright © 2016-2026 The Thingsboard AuthorsLicensed under the Apache License, Version 2.0 (the “License”);you may not use this file except in compliance with the License.You may obtain a copy of the License athttp://www.apache.org/licenses/LICENSE-2.0Unless required by applicable law or agreed to in writing, softwaredistributed under the License is distributed on an “AS IS” BASIS,WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.See the License for the specific language governing permissions andlimitations under the License.*/package org.thingsboard.rule.engine.api;import org.thingsboard.server.common.data.msg.TbNodeConnectionType;import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;import org.thingsboard.server.common.data.plugin.ComponentScope;import org.thingsboard.server.common.data.plugin.ComponentType;import org.thingsboard.server.common.data.rule.RuleChainType;import java.lang.annotation.ElementType;import java.lang.annotation.Retention;import java.lang.annotation.RetentionPolicy;import java.lang.annotation.Target;Retention(RetentionPolicy.RUNTIME)Target(ElementType.TYPE)public interface RuleNode {ComponentType type(); String name(); String nodeDescription(); String nodeDetails(); Class? extends NodeConfiguration configClazz(); ComponentClusteringMode clusteringMode() default ComponentClusteringMode.ENABLED; boolean hasQueueName() default false; boolean inEnabled() default true; boolean outEnabled() default true; ComponentScope scope() default ComponentScope.TENANT; String[] relationTypes() default {TbNodeConnectionType.SUCCESS, TbNodeConnectionType.FAILURE}; String[] uiResources() default {}; String configDirective() default ; String icon() default ; String iconUrl() default ; String docUrl() default ; boolean customRelations() default false;//自定义类型分支, 否则只有 Success 或者 Failual boolean ruleChainNode() default false; RuleChainType[] ruleChainTypes() default {RuleChainType.CORE, RuleChainType.EDGE}; int version() default 0;}字段 本质type 节点分类影响UI分组name 节点名字relationTypes 输出分支非常重要configClazz 配置结构configDirective UI配置可选relationTypes “你这个节点会产生几种出口”例如True / FalseSuccess / Failure核心能力获取输入内容能力msg.getData() // JSON 数据msg.getMetadata() // 元数据msg.getOriginator() // 来源设备使用平台能力ctx.getDeviceService() //设备服务ctx.getTelemetryService()//遥测数据服务ctx.getAlarmService() //告警服务参考 org.thingsboard.rule.engine.api.TbContextRuleNodeId getSelfId(); RuleNode getSelf(); String getRuleChainName(); String getQueueName(); TenantId getTenantId(); AttributesService getAttributesService(); CustomerService getCustomerService(); TenantService getTenantService(); UserService getUserService(); AssetService getAssetService(); DeviceService getDeviceService(); DeviceProfileService getDeviceProfileService(); AssetProfileService getAssetProfileService(); DeviceCredentialsService getDeviceCredentialsService(); DeviceStateManager getDeviceStateManager(); String getDeviceStateNodeRateLimitConfig(); TbClusterService getClusterService(); DashboardService getDashboardService(); RuleEngineAlarmService getAlarmService(); AlarmCommentService getAlarmCommentService(); RuleChainService getRuleChainService(); RuleEngineRpcService getRpcService(); RuleEngineTelemetryService getTelemetryService(); TimeseriesService getTimeseriesService(); RelationService getRelationService(); EntityViewService getEntityViewService(); ResourceService getResourceService(); TbResourceDataCache getTbResourceDataCache(); OtaPackageService getOtaPackageService(); RuleEngineDeviceProfileCache getDeviceProfileCache(); RuleEngineAssetProfileCache getAssetProfileCache(); EdgeService getEdgeService(); EdgeEventService getEdgeEventService(); QueueService getQueueService(); QueueStatsService getQueueStatsService(); ListeningExecutor getMailExecutor(); ListeningExecutor getSmsExecutor(); ListeningExecutor getDbCallbackExecutor(); ListeningExecutor getExternalCallExecutor(); ListeningExecutor getNotificationExecutor(); ExecutorProvider getPubSubRuleNodeExecutorProvider(); MailService getMailService(boolean isSystem); SmsService getSmsService(); SmsSenderFactory getSmsSenderFactory(); NotificationCenter getNotificationCenter(); NotificationTargetService getNotificationTargetService(); NotificationTemplateService getNotificationTemplateService(); NotificationRequestService getNotificationRequestService(); NotificationRuleService getNotificationRuleService(); OAuth2ClientService getOAuth2ClientService(); DomainService getDomainService(); MobileAppService getMobileAppService(); MobileAppBundleService getMobileAppBundleService(); SlackService getSlackService(); CalculatedFieldService getCalculatedFieldService(); RuleEngineCalculatedFieldQueueService getCalculatedFieldQueueService(); JobService getJobService(); JobManager getJobManager(); ApiKeyService getApiKeyService();调用通常是异步的, 所以不要阻塞, 使用回调同一个设备的数据 → 永远发到同一个节点实例, 所以可以放心做缓存, 状态管理按设备输出消息生成新消息TbMsg newMsg TbMsg.transformMsg(msg, newData);ctx.tellNext(newMsg, “Success”);a sample projectClone the repository and navigate to the repo folder:git clone -b release-4.3 https://github.com/thingsboard/rule-node-examplescd rule-node-examples自定义实现设备定时器功能TB规则链居然没有类似定时触发器这样的功能, 比如设定一个时间段定时的开关, 这种很常见的场景都办法实现想了非常多, 要实时性, 但不能扫数据库; 干脆使用类似 xxl-job 调度系统在外部触发, 但是不契合tb的设计逻辑, 与TB内部的实体 数据联动, 扩展性非常的差…类似的有一个 generator 节点, 周期性的产生一条消息到下游, 可以参考一下它的实现两个核心的问题:在哪里保存/记录定时, 类似中断不要消耗CPU?怎么保证集群更改/服务重启时不会丢失?TbMsgGeneratorNode 参考通常规则引擎是“事件驱动”的设备发一条消息规则链处理一条。它是一个“无中生有”的消息触发器按照你设定的固定时间间隔自动生成消息并驱动规则链的后续逻辑它的配置包含两个部分Period (ms) / 触发间隔设定每隔多少毫秒生成一次消息例如 60000 代表每分钟一次。Generator Script / 生成脚本一段 JavaScript 代码用于定义每次生成的消息内容。脚本必须返回一个包含 msg消息体、metadata元数据和 msgType消息类型的对象。服务重启或者集群新增节点重新分区时, 规则链怎么重建的?TbMsgGeneratorNode 的解法——依赖 onPartitionChangeMsg 重新初始化RuleNodeActorMessageProcessor.onPartitionChangeMsg():├── 如果该节点不属于本分区 → stop() → tbNode.destroy()└── 如果该节点属于本分区├── 已运行中 → tbNode.onPartitionChangeMsg(ctx, msg)└── 已停止 → start() → tbNode.init(ctx, config) ← 重启后在这里重建org.thingsboard.server.actors.ruleChain.RuleNodeActorMessageProcessor#onPartitionChangeMsgOverridepublic void onPartitionChangeMsg(PartitionChangeMsg msg) throws Exception {log.debug(“[{}][{}] onPartitionChangeMsg: [{}]”, tenantId, entityId, msg);if (tbNode ! null) {if (!isMyNodePartition()) {//如果该节点不属于本分区 ctx.getSelfId()Rule Node 自身的 ID作为 originator → 规则节点 ID 被分配到固定分区 → 集群中只有一个节点负责该规则节点不会重复触发stop(null);tbNode null;} else {tbNode.onPartitionChangeMsg(defaultCtx, msg);}} else if (isMyNodePartition()) {//如果该节点属于本分区start(null);}}在这两个入口里 TbMsgGeneratorNode 重新 scheduleNextTick()就是 tellSelf 给自己再发一个队列消息…org.thingsboard.rule.engine.debug.TbMsgGeneratorNode#scheduleTickMsgprivate void scheduleTickMsg(TbContext ctx, TbMsg msg) {long curTs System.currentTimeMillis();if (lastScheduledTs 0L) {lastScheduledTs curTs;}lastScheduledTs lastScheduledTs delay;long curDelay Math.max(0L, (lastScheduledTs - curTs));TbMsg tickMsg ctx.newMsg(queueName, TbMsgType.GENERATOR_NODE_SELF_MSG, ctx.getSelfId(),getCustomerIdFromMsg(msg), TbMsgMetaData.EMPTY, TbMsg.EMPTY_STRING);nextTickId tickMsg.getId();ctx.tellSelf(tickMsg, curDelay);//注意这个给自己发一个延迟消息! 延迟放到队列中, 但会在服务重启时会丢失log.trace(“[{}] Scheduled tick msg with delay {}, msg: {}, config: {}”, originatorId, curDelay, tickMsg, config);}so, 回答上述两个问题使用 ctx.tellSelf(tickMsg, curDelay) 把定时记录逻辑放到队列中在 onPartitionChangeMsg 节点回调时再重建规则消息完整实现代码 TbDeviceSchedulerNode关键问题解决了, 照搬就行, 注意一点在重建规则消息时, 要补偿一条消息 经验之谈/**设备定时调度触发器节点per-device 精确定时版本。启动时一次性扫描归属本规则链的设备为每台设备的每条 trigger 注册独立的延迟消息tellSelf在 cron 命中时刻精确触发触发后自动重新注册下一次。运行期间无全局周期扫描。scheduleConfig 属性示例存于设备 Server Attribute{code{enabled: true,triggers: [{ cron: 0 8 * * 1-5, type: open, param: { level: 100, mode: auto } },{ cron: 0 22 * * *, type: close, param: { level: 0 } }]}}使用规则链将 ATTRIBUTES_UPDATEDServer scopescheduleConfig key路由到本节点节点处理后 ack不产生输出。其他属性变更会被 tellSuccess 透传不影响其他节点。节点取消旧 tick 并注册新 tick无需重启。取消机制tellSelf 无 cancel API取消靠 UUID 失效——新 tick 覆盖 activeTickIds 中的值旧 tick 到来时 UUID 不匹配被静默丢弃。*/Slf4jRuleNode(type ComponentType.ACTION,name “device schedule trigger”,configClazz TbDeviceSchedulerNodeConfig.class,nodeDescription “按 cron 表达式为每台设备独立注册定时触发在精确时刻发出自定义类型消息”,nodeDetails “启动时一次性扫描属于本规则链 Profile 的设备注册 per-device 定时 tick。” “调度配置存于设备 Server Attributekey 可配置默认 scheduleConfig。” “格式{“enabled”:true,“triggers”:[{“cron”:“0 8 * * 1-5”,“type”:“open”},…]}。” “cron 为 5 字段分 时 日 月 周type 即输出连接名完全自定义。” “ATTRIBUTES_UPDATED 消息路由到本节点可实时更新调度配置无需重启。” “服务重启后自动补偿当天最后一次已过触发isReconciletrue 时下游节点可选择跳过。”,inEnabled true,icon “alarm”)public class TbDeviceSchedulerNode implements TbNode {private static final DateTimeFormatter FMT DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm); // tick 消息 metadata 内部字段 key private static final String META_TICK_KEY __tickKey; private static final String META_DEVICE_ID __tickDeviceId; private static final String META_CUSTOMER_ID __tickCustomerId; private static final String META_CRON __tickCron; private static final String META_TRIGGER_TYPE __tickTriggerType; private static final String META_SCHED_TIME __tickScheduledTime; private static final String META_PARAM __tickParam; private TbDeviceSchedulerNodeConfig config; private final AtomicBoolean initialized new AtomicBoolean(false); /** key deviceId:triggerIndexvalue 当前有效 tick UUIDUUID 不匹配时旧 tick 被静默丢弃 */ private final ConcurrentHashMapString, UUID activeTickIds new ConcurrentHashMap(); /** 归属本规则链的 DeviceProfile ID启动时一次性收集 */ private volatile SetDeviceProfileId targetProfileIds null; Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { config TbNodeUtils.convert(configuration, TbDeviceSchedulerNodeConfig.class); startIfMine(ctx); } Override public void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) { if (ctx.isLocalEntity(ctx.getSelfId())) { if (initialized.compareAndSet(false, true)) { startAsync(ctx); } } else { initialized.set(false); activeTickIds.clear(); targetProfileIds null; } } Override public void onMsg(TbContext ctx, TbMsg msg) { if (!initialized.get()) { ctx.ack(msg); return; } String type msg.getType(); if (TbMsgType.GENERATOR_NODE_SELF_MSG.name().equals(type)) { handleTick(ctx, msg); } else if (TbMsgType.ATTRIBUTES_UPDATED.name().equals(type)) { handleAttributeUpdate(ctx, msg); } else { ctx.tellSuccess(msg);