ThingsBoard 自定义分发节点 duplicate to related

正文、配图与代码来自原稿。参考 PE 的 duplicate to related，在开源侧实现「按关系把消息复制到多个关联实体」——批量 RPC 下发几乎必备。

1、概述

ThingsBoard PE 有 duplicate to related 节点：按关系查出关联实体，为每个实体复制一条消息（发起者换成关联实体）。CE 默认没有同名能力时，可按本篇自定义。

典型用途：资产 / 网关收到一条指令后，复制到其下所有设备，再接 RPC Call Request 做批量下发。

2、实现

2.1、项目结构

仍基于官方「自定义规则节点」示例工程改造：

2.2、代码实现

TbAbstractDuplicateMsgNode

抽象基类：异步 duplicate 后，一条则直接 Success，多条则先 ack 原消息再逐条 tellNext。

package org.thingsboard.rule.engine.duplicate;

import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.common.util.DonAsynchron;
import org.thingsboard.rule.engine.api.*;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.msg.TbMsg;

import java.util.List;

public abstract class TbAbstractDuplicateMsgNode implements TbNode {

    private TbDuplicateMsgNodeConfiguration config;

    @Override
    public void init(TbContext context, TbNodeConfiguration configuration) throws TbNodeException {
        this.config = TbNodeUtils.convert(configuration, TbDuplicateMsgNodeConfiguration.class);
    }

    @Override
    public void onMsg(TbContext ctx, TbMsg msg) {
        DonAsynchron.withCallback(this.duplicate(ctx, msg), (messages) -> {
            if (messages != null && !messages.isEmpty()) {
                if (messages.size() == 1) {
                    ctx.tellNext(messages.get(0), TbRelationTypes.SUCCESS);
                } else {
                    ctx.ack(msg);
                    messages.forEach(message ->
                            ctx.tellNext(message, TbRelationTypes.SUCCESS));
                }
            } else {
                ctx.tellNext(msg, TbRelationTypes.FAILURE);
            }
        }, (t) -> ctx.tellFailure(msg, t));
    }

    protected abstract ListenableFuture<List<TbMsg>> duplicate(TbContext context, TbMsg tbMsg);

    public void setConfig(TbDuplicateMsgNodeConfiguration config) {
        this.config = config;
    }
}

TbAbstractDuplicateMsgToOriginatorsNode

按「新发起者列表」复制消息：

package org.thingsboard.rule.engine.duplicate;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.msg.TbMsg;

import java.util.ArrayList;
import java.util.List;

public abstract class TbAbstractDuplicateMsgToOriginatorsNode extends TbAbstractDuplicateMsgNode {

    @Override
    protected ListenableFuture<List<TbMsg>> duplicate(TbContext ctx, TbMsg msg) {
        ListenableFuture<List<EntityId>> newOriginators =
                this.getNewOriginators(ctx, msg.getOriginator());
        return Futures.transform(newOriginators, (entityIds) -> {
            if (entityIds != null && !entityIds.isEmpty()) {
                List<TbMsg> messages = new ArrayList<>();
                if (entityIds.size() == 1) {
                    messages.add(ctx.transformMsg(
                            msg, msg.getType(), entityIds.get(0),
                            msg.getMetaData(), msg.getData()));
                } else {
                    entityIds.forEach(entityId -> messages.add(
                            ctx.newMsg(msg.getQueueName(), msg.getType(), entityId,
                                    msg.getCustomerId(), msg.getMetaData(), msg.getData())));
                }
                return messages;
            }
            return null;
        }, ctx.getDbCallbackExecutor());
    }

    protected abstract ListenableFuture<List<EntityId>> getNewOriginators(
            TbContext context, EntityId entityId);
}

TbDuplicateMsgNodeConfiguration

package org.thingsboard.rule.engine.duplicate;

public class TbDuplicateMsgNodeConfiguration {
}

TbDuplicateMsgToRelatedNode

package org.thingsboard.rule.engine.duplicate;

import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.thingsboard.rule.engine.api.*;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.rule.engine.util.EntitiesRelatedEntityIdAsyncLoader;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.plugin.ComponentType;

import java.util.List;

@RuleNode(
        type = ComponentType.TRANSFORMATION,
        name = "duplicate to related",
        configClazz = TbDuplicateMsgToRelatedNodeConfiguration.class,
        nodeDescription = "Duplicates message to related entities fetched by relation query",
        nodeDetails = "Related Entities found using configured relation direction and Relation Type. "
                + "For each found related entity new message is created with related entity as originator "
                + "and message parameters copied from original message.",
        uiResources = {"static/rulenode/rulenode-core-config.js"},
        icon = "call_split"
)
public class TbDuplicateMsgToRelatedNode extends TbAbstractDuplicateMsgToOriginatorsNode {

    private static final Logger log = LoggerFactory.getLogger(TbDuplicateMsgToRelatedNode.class);
    private TbDuplicateMsgToRelatedNodeConfiguration config;

    @Override
    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
        this.config = TbNodeUtils.convert(configuration, TbDuplicateMsgToRelatedNodeConfiguration.class);
        this.validateConfig(this.config);
        this.setConfig(this.config);
    }

    @Override
    protected ListenableFuture<List<EntityId>> getNewOriginators(TbContext ctx, EntityId original) {
        return EntitiesRelatedEntityIdAsyncLoader.findEntitiesAsync(
                ctx, original, this.config.getRelationsQuery());
    }

    private void validateConfig(TbDuplicateMsgToRelatedNodeConfiguration conf) {
        if (conf.getRelationsQuery() == null) {
            log.error("TbDuplicateMsgToRelatedNode configuration should have relations query");
            throw new IllegalArgumentException(
                    "Wrong configuration for TbDuplicateMsgToRelatedNode: relation query is missing.");
        }
    }

    @Override
    public void destroy() {
    }
}

TbDuplicateMsgToRelatedNodeConfiguration

package org.thingsboard.rule.engine.duplicate;

import lombok.Data;
import org.thingsboard.rule.engine.api.NodeConfiguration;
import org.thingsboard.rule.engine.data.RelationsQuery;
import org.thingsboard.server.common.data.relation.EntitySearchDirection;
import org.thingsboard.server.common.data.relation.RelationEntityTypeFilter;

import java.util.Collections;

@Data
public class TbDuplicateMsgToRelatedNodeConfiguration
        extends TbDuplicateMsgNodeConfiguration
        implements NodeConfiguration<TbDuplicateMsgToRelatedNodeConfiguration> {

    private RelationsQuery relationsQuery;

    @Override
    public TbDuplicateMsgToRelatedNodeConfiguration defaultConfiguration() {
        TbDuplicateMsgToRelatedNodeConfiguration configuration =
                new TbDuplicateMsgToRelatedNodeConfiguration();
        RelationsQuery relationsQuery = new RelationsQuery();
        relationsQuery.setDirection(EntitySearchDirection.FROM);
        relationsQuery.setMaxLevel(1);
        RelationEntityTypeFilter entityTypeFilter =
                new RelationEntityTypeFilter("Contains", Collections.emptyList());
        relationsQuery.setFilters(Collections.singletonList(entityTypeFilter));
        configuration.setRelationsQuery(relationsQuery);
        return configuration;
    }
}

EntitiesRelatedEntityIdAsyncLoader（原稿未贴全码）

原稿写「请私信获取」。较新的 ThingsBoard CE 源码里已有同名工具类（rule-engine-components → org.thingsboard.rule.engine.util.EntitiesRelatedEntityIdAsyncLoader），并提供 findEntitiesAsync，可直接复用。

若你分支没有 findEntitiesAsync，可按官方逻辑自备一版（思路如下）：

package org.thingsboard.rule.engine.util;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.data.RelationsQuery;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.EntityRelationsQuery;
import org.thingsboard.server.common.data.relation.EntitySearchDirection;
import org.thingsboard.server.common.data.relation.RelationsSearchParameters;

import java.util.Collections;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;

public final class EntitiesRelatedEntityIdAsyncLoader {

    private EntitiesRelatedEntityIdAsyncLoader() {
    }

    public static ListenableFuture<List<EntityId>> findEntitiesAsync(
            TbContext ctx, EntityId originator, RelationsQuery relationsQuery) {

        EntityRelationsQuery query = new EntityRelationsQuery();
        query.setParameters(new RelationsSearchParameters(
                originator,
                relationsQuery.getDirection(),
                relationsQuery.getMaxLevel(),
                relationsQuery.isFetchLastLevelOnly()
        ));
        query.setFilters(relationsQuery.getFilters());

        ListenableFuture<List<EntityRelation>> asyncRelation =
                ctx.getRelationService().findByQuery(ctx.getTenantId(), query);

        Function<EntityRelation, EntityId> mapFunction;
        if (relationsQuery.getDirection() == EntitySearchDirection.FROM) {
            mapFunction = EntityRelation::getTo;
        } else if (relationsQuery.getDirection() == EntitySearchDirection.TO) {
            mapFunction = EntityRelation::getFrom;
        } else {
            return Futures.immediateFailedFuture(
                    new IllegalStateException("Unknown direction"));
        }

        return Futures.transform(asyncRelation, relations -> {
            if (relations == null || relations.isEmpty()) {
                return Collections.emptyList();
            }
            return relations.stream().map(mapFunction).collect(Collectors.toList());
        }, ctx.getDbCallbackExecutor());
    }
}

API（如 isFetchLastLevelOnly、构造参数）以你检出的 TB 版本为准；与主工程已有类冲突时，优先用官方已有实现。

打包与放入 extensions/ 的方式同 entity query 节点。

3、节点预览效果

规则链中的节点与配置界面：

---

补充（辅）

• 默认关系：FROM + Contains + maxLevel=1，按现场改方向与关联类型名。
• 批量 RPC：duplicate to related →（可选 script）→ rpc call request；每个副本的 originator 已是目标设备。
• 多消息时原稿对原消息 ack 再分发，避免队列重复投递，勿随意删掉。

相关文章

• 自定义万能查询节点 entity query
• 实现定时任务或调度器（CE 定时 + 单设备 RPC）
• 规则链：check relation
• 专栏导读