使用 TB 做计算新需求（小时聚合）

正文、配图与代码来自原稿。两类传感器按不同周期上报，做小时均值及后续矩阵运算；TB 侧用自定义节点批量查遥测。

1、需求

两类设备各约 20 台：

颗粒传感器——约每 5 分钟上报；每小时统计一次：

颗粒传感器需求

环境传感器——约每 20 分钟上报；每小时统计一次：

环境传感器需求

统计方式：同一小时内，将该类 20 台设备对应字段求和再除以上报次数，得到该小时平均值（每小时一条）。

再得到各字段小时均值结果：

小时均值结果示意

用每小时数据代入公式，每字段可得到多组结果（原稿附图 / 附件「计算过程」）：

公式计算示意

后续：用上午的 W 与得到的多组数据做矩阵相乘得 B，取 B 的最大值作为该小时最终结果。（细节见原稿附件计算过程。）

2、实现

2.1、创建新租户

系统层新建租户（略）。

2.2、创建两种设备配置

两种设备配置

2.3、创建设备（模拟）

每种配置先建设备用于模拟上报：

模拟设备

2.4、上报数据格式

颗粒设备（约 5 分钟一次）：

json

{
  "pm10": 13,
  "pm100": 59,
  "pm2_5": 15,
  "devicename": "BZ-PCMU-8",
  "位置": "舍内"
}

环境设备（约 20 分钟一次）：

json

{
  "co2": 555,
  "h2s": 0.2,
  "humidity": 98,
  "nh3": 98,
  "temperature": 24.5,
  "devicename": "BZ-PCMU-8",
  "位置": "舍内"
}

2.5、自定义查询遥测节点

按设备 ID 批量查指定 key 的时序。源码可单独工程打包后拷进容器 extensions：

bash

# 容器内若无 extensions，先 exec 创建目录
docker cp rule-engine-1.0.0-custom-nodes.jar mytb:/usr/share/thingsboard/extensions/

TbQueryAllDeviceTelemetryConfiguration

java

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

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

@Data
public class TbQueryAllDeviceTelemetryConfiguration
        implements NodeConfiguration<TbQueryAllDeviceTelemetryConfiguration> {

    private String deviceKeys;
    private String devices;
    private String startTimeKey;
    private String endTimeKey;

    @Override
    public TbQueryAllDeviceTelemetryConfiguration defaultConfiguration() {
        TbQueryAllDeviceTelemetryConfiguration configuration =
                new TbQueryAllDeviceTelemetryConfiguration();
        configuration.setDeviceKeys("keliDevcieKeys");
        configuration.setDevices("keliDevcies");
        configuration.setStartTimeKey("startTime");
        configuration.setEndTimeKey("endTime");
        return configuration;
    }
}

（默认 key 名与原稿一致，含拼写；与上游 script 写入的 metadata 键对齐即可。）

TbQueryAllDeviceTelemetryNode

节点名：get devices telemetry。从 metadata 读取遥测 key、设备列表、时间窗，批量 findAll 后写回 metadata。

java

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

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.*;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.kv.Aggregation;
import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@RuleNode(
        type = ComponentType.ENRICHMENT,
        name = "get devices telemetry",
        configClazz = TbQueryAllDeviceTelemetryConfiguration.class,
        nodeDescription = "查询指定设备的遥测",
        nodeDetails = "查询指定设备的遥测",
        uiResources = {"static/rulenode/custom-nodes-config.js"}
)
public class TbQueryAllDeviceTelemetryNode implements TbNode {

    private TbQueryAllDeviceTelemetryConfiguration config;
    private Aggregation aggregation;
    private List<TsKvEntry> resList;

    @Override
    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
        this.config = TbNodeUtils.convert(configuration, TbQueryAllDeviceTelemetryConfiguration.class);
        this.aggregation = parseAggregationConfig();
        this.resList = new ArrayList<>();
    }

    @Override
    public void onMsg(TbContext ctx, TbMsg msg)
            throws ExecutionException, InterruptedException, TbNodeException {
        try {
            resList.clear();
            Map<String, String> metadata = msg.getMetaData().getData();
            String deviceKeys = metadata.get(config.getDeviceKeys());
            List<String> keys = Arrays.asList(deviceKeys.split(","));
            String devices = metadata.get(config.getDevices());
            JSONArray deviceArray = JSONArray.parseArray(devices);
            long startTime = Long.valueOf(metadata.get(config.getStartTimeKey()));
            long endTime = Long.valueOf(metadata.get(config.getEndTimeKey()));
            Interval interval = getInterval(startTime, endTime);

            for (Object obj : deviceArray) {
                JSONObject device = (JSONObject) obj;
                String id = device.getString("id");
                DeviceId deviceId = DeviceId.fromString(id);
                ListenableFuture<List<TsKvEntry>> list = ctx.getTimeseriesService()
                        .findAll(ctx.getTenantId(), deviceId, buildQueries(interval, keys));
                resList.addAll(list.get());
            }
            process(resList, msg, keys);
            ctx.tellSuccess(ctx.transformMsg(
                    msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), msg.getData()));
        } catch (Exception e) {
            ctx.tellFailure(msg, e.getCause());
        }
    }

    private void process(List<TsKvEntry> entries, TbMsg msg, List<String> keys) {
        ObjectNode resultNode = JacksonUtil.newObjectNode(JacksonUtil.ALLOW_UNQUOTED_FIELD_NAMES_MAPPER);
        entries.forEach(entry -> processArray(resultNode, entry));
        for (String key : keys) {
            if (resultNode.has(key)) {
                msg.getMetaData().putValue(key, resultNode.get(key).toString());
            }
        }
    }

    private void processArray(ObjectNode node, TsKvEntry entry) {
        if (node.has(entry.getKey())) {
            ((ArrayNode) node.get(entry.getKey())).add(buildNode(entry));
        } else {
            ArrayNode arrayNode = JacksonUtil.ALLOW_UNQUOTED_FIELD_NAMES_MAPPER.createArrayNode();
            arrayNode.add(buildNode(entry));
            node.set(entry.getKey(), arrayNode);
        }
    }

    private ObjectNode buildNode(TsKvEntry entry) {
        ObjectNode obj = JacksonUtil.newObjectNode(JacksonUtil.ALLOW_UNQUOTED_FIELD_NAMES_MAPPER);
        obj.put("ts", entry.getTs());
        JacksonUtil.addKvEntry(obj, entry, "value", JacksonUtil.ALLOW_UNQUOTED_FIELD_NAMES_MAPPER);
        return obj;
    }

    private Interval getInterval(long startTime, long endTime) {
        Interval interval = new Interval();
        interval.setStartTs(startTime);
        interval.setEndTs(endTime);
        return interval;
    }

    private List<ReadTsKvQuery> buildQueries(Interval interval, List<String> keys) {
        final long aggIntervalStep = Aggregation.NONE.equals(aggregation) ? 1
                : interval.getEndTs() - interval.getStartTs();
        return keys.stream()
                .map(key -> new BaseReadTsKvQuery(
                        key, interval.getStartTs(), interval.getEndTs(),
                        aggIntervalStep, 1000, aggregation, "DESC"))
                .collect(Collectors.toList());
    }

    Aggregation parseAggregationConfig() {
        return Aggregation.NONE;
    }

    @Data
    @NoArgsConstructor
    private static class Interval {
        private Long startTs;
        private Long endTs;
    }
}

定时触发可用 Generator 调度每小时跑一次规则链，再接 script 做均值与矩阵。

补充（辅）

小时窗：startTime/endTime 建议脚本按整点对齐。
均值：查回原始点后在 script 中 sum/count；或改 Aggregation.AVG（需与查询 API 行为一致）。
矩阵运算可放规则链 script，或结果出站到外部计算服务。

使用 TB 做计算新需求（小时聚合） ​

1、需求 ​

2、实现 ​

2.1、创建新租户 ​

2.2、创建两种设备配置 ​

2.3、创建设备（模拟） ​

2.4、上报数据格式 ​

2.5、自定义查询遥测节点 ​

TbQueryAllDeviceTelemetryConfiguration ​

TbQueryAllDeviceTelemetryNode ​

补充（辅） ​

相关文章 ​