ThingsBoard 自定义万能查询节点 entity query

正文、配图与代码来自原稿。用自定义规则节点按 EntityDataQuery 万能查实体，结果放进 metadata，可大幅减少「为每种查询再写一个节点」。

1、概述

本篇分享如何自定义规则节点——万能查询节点（entity query）。它几乎可以覆盖很多业务侧「按条件查实体」的需求，是规则链里很实用的万金油节点。

核心能力：

1. 从消息 metadata 指定 key 读出一段 EntityDataQuery JSON
2. 调用 EntityService.findEntityDataByQuery 异步查询
3. 把结果 JSON 写回 metadata 的另一个 key，再 tellSuccess

2、实现

2.1、官方自定义规则节点工程

下载官方「自定义规则节点」示例工程，把文件夹改成自己的节点工程名：

用 IDEA 打开后，修改 POM 的 artifactId：

2.2、自定义节点

工程结构示意：

为什么要建 org.thingsboard.server.util？

ThingsBoard 启动类会扫描 org.thingsboard.server。自定义的 SpringContextUtil 需要被 Spring 管理，才能从容器里拿到 dao-api 里的 Service。若 TbContext 没直接暴露你需要的 Bean，就用 SpringContextUtil 动态取 Bean 再调方法。dao-api 中 Service 一览：

EntityQueryNode

package org.thingsboard.rule.engine.node.entityquery;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.lang.Nullable;
import org.thingsboard.rule.engine.api.*;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.rule.engine.util.EntitiesQueryAsyncLoader;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.data.query.EntityData;
import org.thingsboard.server.common.data.query.EntityDataQuery;
import org.thingsboard.server.common.msg.TbMsg;

import java.util.List;

@Slf4j
@RuleNode(
        type = ComponentType.ENRICHMENT,
        name = "entity query",
        relationTypes = {},
        configClazz = EntityQueryNodeConfig.class,
        nodeDescription = "query entity by condition",
        nodeDetails = "Custom Entity Query Node By Aldrich Eugene 2023-02-25",
        uiResources = {"static/rulenode/rulenode-core-config.js"}
)
public class EntityQueryNode implements TbNode {

    private String inputKey;
    private String outputKey;
    private EntityQueryNodeConfig config;
    private static final ObjectMapper mapper = new ObjectMapper();

    @Override
    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
        this.config = TbNodeUtils.convert(configuration, EntityQueryNodeConfig.class);
        this.inputKey = config.getInputKey();
        this.outputKey = config.getOutputKey();
    }

    @Override
    public void onMsg(TbContext ctx, TbMsg msg) {
        try {
            // 通过配置的 inputKey 从 metadata 取值（内容为 EntityDataQuery JSON）
            String queryJson = msg.getMetaData().getValue(inputKey);
            if (queryJson == null || "".equals(queryJson)) {
                ctx.tellFailure(msg, new Throwable("not find inputKey in metadata."));
                return;
            }
            EntityDataQuery query = new ObjectMapper().readValue(queryJson, EntityDataQuery.class);
            ListenableFuture<List<EntityData>> future = this.entityQueryAsync(ctx, query);
            Futures.addCallback(future, new FutureCallback<List<EntityData>>() {
                @Override
                public void onSuccess(@Nullable List<EntityData> entityData) {
                    try {
                        msg.getMetaData().getData().put(outputKey, mapper.writeValueAsString(entityData));
                        ctx.tellSuccess(msg);
                    } catch (Exception e) {
                        ctx.tellFailure(msg, new Throwable(e.getCause() + "," + e.getMessage()));
                    }
                }

                @Override
                public void onFailure(Throwable throwable) {
                    ctx.tellFailure(msg, throwable);
                }
            }, ctx.getDbCallbackExecutor());
        } catch (Exception e) {
            e.printStackTrace();
            ctx.tellFailure(msg, e);
        }
    }

    private ListenableFuture<List<EntityData>> entityQueryAsync(TbContext ctx, EntityDataQuery query) {
        return EntitiesQueryAsyncLoader.entityQueryAsync(ctx, query);
    }
}

原稿里曾把 inputKey 配置字段直接赋成 metadata 取值；上面示例用局部变量 queryJson，避免冲掉配置里的 key 名。

EntityQueryNodeConfig

package org.thingsboard.rule.engine.node.entityquery;

import lombok.Data;
import org.thingsboard.rule.engine.api.NodeConfiguration;

@Data
public class EntityQueryNodeConfig implements NodeConfiguration<EntityQueryNodeConfig> {

    private String inputKey;
    private String outputKey;

    @Override
    public EntityQueryNodeConfig defaultConfiguration() {
        EntityQueryNodeConfig configuration = new EntityQueryNodeConfig();
        configuration.setInputKey("inputKey");
        configuration.setOutputKey("outputKey");
        return configuration;
    }
}

EntitiesQueryAsyncLoader

package org.thingsboard.rule.engine.util;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.query.EntityData;
import org.thingsboard.server.common.data.query.EntityDataQuery;
import org.thingsboard.server.dao.entity.EntityService;
import org.thingsboard.server.util.SpringContextUtil;

import java.util.List;

@Slf4j
public class EntitiesQueryAsyncLoader {

    public static ListenableFuture<List<EntityData>> entityQueryAsync(
            TbContext ctx, EntityDataQuery entityDataQuery) {
        List<EntityData> entityData = null;
        try {
            EntityService entityService = SpringContextUtil.getInstance()
                    .getApplicationContext()
                    .getBean(EntityService.class);
            PageData<EntityData> entityDataPageData = entityService.findEntityDataByQuery(
                    ctx.getTenantId(),
                    new CustomerId(CustomerId.NULL_UUID),
                    entityDataQuery
            );
            entityData = entityDataPageData.getData();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return Futures.immediateFuture(entityData);
    }
}

SpringContextUtil

package org.thingsboard.server.util;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

@Component
public class SpringContextUtil implements ApplicationContextAware {

    private static ApplicationContext applicationContext;
    private static SpringContextUtil springContextUtil;

    private SpringContextUtil() {
    }

    public static SpringContextUtil getInstance() {
        if (springContextUtil == null) {
            springContextUtil = new SpringContextUtil();
        }
        return springContextUtil;
    }

    public ApplicationContext getApplicationContext() {
        return this.applicationContext;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        springContextUtil = this;
    }

    public static Object getBean(String name) throws BeansException {
        return SpringContextUtil.getInstance().getApplicationContext().getBean(name);
    }

    public static <T> T getBean(Class<T> clazz) throws BeansException {
        return SpringContextUtil.getInstance().getApplicationContext().getBean(clazz);
    }
}

（setApplicationContext 时把单例指回 Spring 创建的实例，避免 getInstance() 新建空对象拿不到 Context。）

2.3、install 生成包

执行 mvn clean install 时会校验文件头 license，需加上与工程其它类一致的 Apache 头（见 本地编译常见问题）。

2.4、导入工程

部署方式	放置位置
本地跑 TB	自定义节点 jar 放到 application 工程下的 lib
Docker 线上	拷进容器 /usr/share/thingsboard/extensions/

cp rule-engine-1.0.0-custom-nodes.jar /usr/share/thingsboard/extensions/
chown thingsboard:thingsboard /usr/share/thingsboard/extensions/*

重启 ThingsBoard 后，规则链 Enrichment 面板应能看到 entity query。

3、注意：也可直接并进主工程

也可以不单独起官方自定义节点工程，把代码拷进 ThingsBoard 源码对应包下。例如在红色框所示位置新建 entityquery 包，把相关类放进去即可：

---

补充（辅）

• 上游节点需先把完整的 EntityDataQuery JSON 写入 metadata（可用 script 组装）。
• CustomerId.NULL_UUID 表示租户级查询；若要按客户隔离，需改成实际 CustomerId。
• 新版本若 TbContext 已能拿到 EntityService，可少一层 SpringContextUtil。

相关文章

• 自定义分发节点 duplicate to related
• 源码分析规则节点
• 自定义华为云 SMS 规则节点
• 专栏导读