ThingsBoard 数据分析：自定义节点订阅 Kafka Stream

正文、配图与代码来自原稿。官方教程第三步依赖 PE Kafka Integration；CE 需自定义消费者节点把 Stream 结果拉回规则链。

1、概述

规则引擎擅长阈值等基础分析；机器学习、预测等高级分析常放到外部。官方思路：

1. 规则链 kafka 节点把遥测推到 Kafka Topic（CE/PE 均可）
2. Kafka Streams 应用做聚合分析
3. PE Kafka Integration 把结果写回 TB 持久化与展示

CE 没有 Integration，拿不到 Stream 输出 Topic 的数据，因此自定义规则节点订阅 Kafka，再继续规则链逻辑。

2、流程架构

设备数据 → 规则节点 kafka → Kafka IN TOPIC → Kafka Streams 计算 → OUT TOPIC →（PE：Integration / CE：自定义 consumer 节点）→ 规则链后续处理。

3、案例

3.1、需求

参考官方太阳能板模块退化分析案例：

先决条件：ThingsBoard CE、Kafka 服务。

3.2、创建设备配置 solar-module

3.3、创建三个设备

3.4、Generator 模拟上报

规则链中用三个 generator 模拟 Module 1/2/3。

Module 1：

var msg = { power: 5, name: "Module 1", panel: "Panel 1" };
var metadata = {};
var msgType = "POST_TELEMETRY_REQUEST";
return { msg: msg, metadata: metadata, msgType: msgType };

Module 2： power: 5, name: "Module 2"

Module 3： power: 3.5, name: "Module 3"

3.5、推入 Kafka

kafka 节点示例：

• Topic：solar-module-raw
• 地址：如 172.16.5.248:19092（改成你的 Broker）

最终规则链：

调试无报错、均为 Success 即表示已写入 Kafka：

4、启动 Kafka Streams 应用

按官方下载示例应用，用 IDEA 打开，修改 SolarConsumer.java 中的 Kafka 地址：

启动后看控制台：

应用消费 solar-module-raw，把分析结果推到 solar-module-anomalies，再由自定义节点订阅：

5、自定义 kafka consumer 节点

节点预览：

KafkaConsumerConfiguration

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

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.rule.engine.api.NodeConfiguration;

@Slf4j
@Data
public class KafkaConsumerConfiguration implements NodeConfiguration<KafkaConsumerConfiguration> {

    private String server;
    private String groupId;
    private String topic;
    private String keyDeserializer;
    private String valueDeserializer;
    private String entityType;
    private String entityId;
    private String entityName;

    @Override
    public KafkaConsumerConfiguration defaultConfiguration() {
        KafkaConsumerConfiguration configuration = new KafkaConsumerConfiguration();
        configuration.setServer("localhost:9092");
        configuration.setGroupId("group1");
        configuration.setTopic("topic1");
        configuration.setKeyDeserializer(
                "org.apache.kafka.common.serialization.StringDeserializer");
        configuration.setValueDeserializer(
                "org.apache.kafka.common.serialization.StringDeserializer");
        configuration.setEntityType("ASSET");
        configuration.setEntityId("<ENTITY_UUID>");
        configuration.setEntityName("deviceA");
        return configuration;
    }
}

KafkaConsumerNode

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

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.thingsboard.rule.engine.api.*;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.session.SessionMsgType;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

@Slf4j
@RuleNode(
        type = ComponentType.EXTERNAL,
        name = "kafka consumer",
        relationTypes = {"True", "False"},
        configClazz = KafkaConsumerConfiguration.class,
        nodeDescription = "Listen to Kafka's topic message",
        nodeDetails = "If the selected key exists - send Message via True chain, otherwise False chain is used.",
        uiResources = {"static/rulenode/custom-nodes-config.js"}
)
public class KafkaConsumerNode implements TbNode {

    private KafkaConsumerConfiguration config;
    private KafkaConsumer<String, String> consumer;

    @Override
    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
        this.config = TbNodeUtils.convert(configuration, KafkaConsumerConfiguration.class);
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getServer());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, config.getGroupId());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, config.getKeyDeserializer());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, config.getValueDeserializer());

        TbMsgMetaData metaData = new TbMsgMetaData();
        metaData.putValue("entityName", config.getEntityName());

        try {
            this.consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Collections.singletonList(config.getTopic()));
            ctx.getExternalCallExecutor().executeAsync(() -> {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    records.forEach(record -> {
                        log.info("Received message: " + record.value());
                        this.onMsg(ctx, TbMsg.newMsg(
                                SessionMsgType.POST_TELEMETRY_REQUEST.name(),
                                getEntityId(config.getEntityType(), config.getEntityId()),
                                metaData,
                                record.value()));
                    });
                }
            });
        } catch (Exception e) {
            throw new TbNodeException(e);
        }
    }

    @Override
    public void onMsg(TbContext ctx, TbMsg msg) {
        try {
            ctx.tellNext(msg, "True");
        } catch (Exception e) {
            ctx.tellFailure(msg, e);
        }
    }

    @Override
    public void destroy() {
        if (this.consumer != null) {
            try {
                this.consumer.close();
            } catch (Exception e) {
                log.error("Failed to close consumer during destroy()", e);
            }
        }
    }

    private EntityId getEntityId(String entityType, String entityId) {
        if (EntityType.ASSET.name().equals(entityType)) {
            return AssetId.fromString(entityId);
        } else if (EntityType.DEVICE.name().equals(entityType)) {
            return DeviceId.fromString(entityId);
        }
        throw new RuntimeException("Incorrect entity type");
    }
}

打包放入 extensions/ 后重启，规则链挂上该节点订阅 solar-module-anomalies。

6、运行测试效果

7、部署 Kafka（Docker Compose 示例）

version: '3.0'
services:
  zookeeper:
    restart: always
    image: "zookeeper:3.8.0"
    ports:
      - "2181:2181"
    environment:
      ZOO_MY_ID: 1
      ZOO_SERVERS: server.1=zookeeper:2888:3888;zookeeper:2181
      ZOO_ADMINSERVER_ENABLED: "false"
  kafka:
    restart: always
    image: "bitnami/kafka:3.2.0"
    ports:
      - "9092:9092"
    environment:
      ALLOW_PLAINTEXT_LISTENER: yes
      KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092
      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://<你的宿主机IP>:9092
    depends_on:
      - zookeeper

docker compose up -d

---

补充（辅）

• while(true) 消费循环需在 destroy 中可中断；生产可加重试、提交 offset、健康检查。
• 官方分析教程见 ThingsBoard Docs「Analytics」相关 Kafka Streams 示例。

相关文章

• 自定义万能查询节点
• 实现定时任务或调度器
• 专栏导读