定义节点消费 TBMQ 或 EMQX 的数据

正文、配图与代码来自原稿。官方「MQTT」节点偏发布；本节点在 init 时订阅 Broker，把外部消息转成 POST_TELEMETRY_REQUEST。

1、目录结构

2、TbMqttSubscribeNodeConfiguration

package org.thingsboard.rule.engine.mqtt.subscription;

import lombok.Data;
import org.thingsboard.rule.engine.api.NodeConfiguration;
import org.thingsboard.rule.engine.credentials.AnonymousCredentials;
import org.thingsboard.rule.engine.credentials.ClientCredentials;

@Data
public class TbMqttSubscribeNodeConfiguration
        implements NodeConfiguration<TbMqttSubscribeNodeConfiguration> {

    private String topicPattern;
    private String host;
    private int port;
    private int connectTimeoutSec;
    private String clientId;
    private boolean appendClientIdSuffix;
    private boolean retainedMessage;
    private boolean cleanSession;
    private boolean ssl;
    private ClientCredentials credentials;

    @Override
    public TbMqttSubscribeNodeConfiguration defaultConfiguration() {
        TbMqttSubscribeNodeConfiguration configuration = new TbMqttSubscribeNodeConfiguration();
        configuration.setTopicPattern("my-topic");
        configuration.setPort(1883);
        configuration.setConnectTimeoutSec(10);
        configuration.setCleanSession(true);
        configuration.setSsl(false);
        configuration.setRetainedMessage(false);
        configuration.setCredentials(new AnonymousCredentials());
        return configuration;
    }
}

3、TbMqttSubscribeNode

核心：init 建客户端并 once/subscribe Topic；onMsg 透传 Success；destroy 断开。

package org.thingsboard.rule.engine.mqtt.subscription;

import io.netty.handler.ssl.SslContext;
import io.netty.util.concurrent.Future;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.mqtt.MqttClient;
import org.thingsboard.mqtt.MqttClientConfig;
import org.thingsboard.mqtt.MqttConnectResult;
import org.thingsboard.rule.engine.api.*;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.rule.engine.credentials.BasicCredentials;
import org.thingsboard.rule.engine.credentials.ClientCredentials;
import org.thingsboard.rule.engine.credentials.CredentialsType;
import org.thingsboard.rule.engine.external.TbAbstractExternalNode;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;

import javax.net.ssl.SSLException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
@RuleNode(
        type = ComponentType.EXTERNAL,
        name = "mqtt subscribe",
        configClazz = TbMqttSubscribeNodeConfiguration.class,
        clusteringMode = ComponentClusteringMode.USER_PREFERENCE,
        nodeDescription = "Subscribe messages from the MQTT broker",
        nodeDetails = "Will subscribe message payload to the MQTT broker with QoS <b>AT_LEAST_ONCE</b>.",
        uiResources = {"static/rulenode/rulenode-core-config.js"},
        configDirective = "tbExternalNodeMqttConfig",
        icon = "call_split"
)
public class TbMqttSubscribeNode extends TbAbstractExternalNode {

    private static final Charset UTF8 = StandardCharsets.UTF_8;
    private static final String ERROR = "error";
    protected TbMqttSubscribeNodeConfiguration mqttNodeConfiguration;
    protected MqttClient mqttClient;
    private ExecutorService executor;
    private TbMqttSubscribeNode instance;

    @Override
    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
        super.init(ctx);
        this.mqttNodeConfiguration = TbNodeUtils.convert(configuration, TbMqttSubscribeNodeConfiguration.class);
        this.executor = Executors.newFixedThreadPool(1);
        this.instance = new TbMqttSubscribeNode();
        try {
            this.mqttClient = initClient(ctx);
            executor.execute(() ->
                    mqttClient.once(mqttNodeConfiguration.getTopicPattern(),
                            new TbMqttSubscribeHandler(ctx, instance)));
        } catch (Exception e) {
            throw new TbNodeException(e);
        }
    }

    @Override
    public void onMsg(TbContext ctx, TbMsg msg) {
        tellSuccess(ctx, msg);
    }

    private TbMsg processException(TbMsg origMsg, Throwable e) {
        TbMsgMetaData metaData = origMsg.getMetaData().copy();
        metaData.putValue(ERROR, e.getClass() + ": " + e.getMessage());
        return TbMsg.transformMsgMetadata(origMsg, metaData);
    }

    @Override
    public void destroy() {
        if (this.mqttClient != null) {
            this.mqttClient.disconnect();
        }
    }

    String getOwnerId(TbContext ctx) {
        return "Tenant[" + ctx.getTenantId().getId() + "]RuleNode[" + ctx.getSelf().getId().getId() + "]";
    }

    protected MqttClient initClient(TbContext ctx) throws Exception {
        MqttClientConfig config = new MqttClientConfig(getSslContext());
        config.setOwnerId(getOwnerId(ctx));
        if (!StringUtils.isEmpty(this.mqttNodeConfiguration.getClientId())) {
            config.setClientId(this.mqttNodeConfiguration.isAppendClientIdSuffix()
                    ? this.mqttNodeConfiguration.getClientId() + "_" + ctx.getServiceId()
                    : this.mqttNodeConfiguration.getClientId());
        }
        config.setCleanSession(this.mqttNodeConfiguration.isCleanSession());
        prepareMqttClientConfig(config);
        MqttClient client = MqttClient.create(config, null, ctx.getExternalCallExecutor());
        client.setEventLoop(ctx.getSharedEventLoop());
        Future<MqttConnectResult> connectFuture =
                client.connect(this.mqttNodeConfiguration.getHost(), this.mqttNodeConfiguration.getPort());
        MqttConnectResult result;
        try {
            result = connectFuture.get(this.mqttNodeConfiguration.getConnectTimeoutSec(), TimeUnit.SECONDS);
        } catch (TimeoutException ex) {
            connectFuture.cancel(true);
            client.disconnect();
            String hostPort = this.mqttNodeConfiguration.getHost() + ":" + this.mqttNodeConfiguration.getPort();
            throw new RuntimeException(String.format("Failed to connect to MQTT broker at %s.", hostPort));
        }
        if (!result.isSuccess()) {
            connectFuture.cancel(true);
            client.disconnect();
            String hostPort = this.mqttNodeConfiguration.getHost() + ":" + this.mqttNodeConfiguration.getPort();
            throw new RuntimeException(String.format(
                    "Failed to connect to MQTT broker at %s. Result code is: %s",
                    hostPort, result.getReturnCode()));
        }
        return client;
    }

    protected void prepareMqttClientConfig(MqttClientConfig config) throws SSLException {
        ClientCredentials credentials = this.mqttNodeConfiguration.getCredentials();
        if (credentials.getType() == CredentialsType.BASIC) {
            BasicCredentials basicCredentials = (BasicCredentials) credentials;
            config.setUsername(basicCredentials.getUsername());
            config.setPassword(basicCredentials.getPassword());
        }
    }

    private SslContext getSslContext() throws SSLException {
        return this.mqttNodeConfiguration.isSsl()
                ? this.mqttNodeConfiguration.getCredentials().initSslContext()
                : null;
    }
}

4、TbMqttSubscribeHandler

按 Topic 约定解析设备名，查库后构造遥测消息。示例 Topic：

tb/mqtt-integration-tutorial/sensors/SN-001/temperature → 第 4 段为设备名。

package org.thingsboard.rule.engine.mqtt.subscription;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import io.netty.buffer.ByteBuf;
import org.springframework.util.ObjectUtils;
import org.thingsboard.mqtt.MqttHandler;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgDataType;
import org.thingsboard.server.common.msg.TbMsgMetaData;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class TbMqttSubscribeHandler implements MqttHandler {

    private TbContext ctx;
    private TbMqttSubscribeNode instance;

    public TbMqttSubscribeHandler(TbContext ctx, TbMqttSubscribeNode instance) {
        this.ctx = ctx;
        this.instance = instance;
    }

    @Override
    public ListenableFuture<Void> onMessage(String topic, ByteBuf payload) {
        try {
            String deviceName = topic.split("/")[3];
            Device device = ctx.getDeviceService()
                    .findDeviceByTenantIdAndName(ctx.getTenantId(), deviceName);
            if (ObjectUtils.isEmpty(device)) {
                throw new RuntimeException("Not find device [" + deviceName + "]");
            }
            byte[] bytes = new byte[payload.readableBytes()];
            payload.readBytes(bytes);
            String msg = new String(bytes, StandardCharsets.UTF_8);
            Map<String, String> metadata = new HashMap<>();
            metadata.put("deviceName", deviceName);
            metadata.put("deviceType", device.getType());
            metadata.put("ts", String.valueOf(System.currentTimeMillis()));
            TbMsg tbMsg = TbMsg.newMsg(
                    TbMsgType.POST_TELEMETRY_REQUEST,
                    device.getId(),
                    new TbMsgMetaData(metadata),
                    TbMsgDataType.JSON,
                    msg,
                    null,
                    null);
            instance.onMsg(ctx, tbMsg);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return Futures.immediateFuture(null);
    }
}

辅

原稿 handler 中曾漏掉 payload.readBytes(bytes)，上例已补上。集群下注意 clusteringMode 与 ClientId，避免重复消费；长期订阅宜用 on/subscribe 而非一次性 once，按你使用的 MQTT 客户端 API 调整。

5、小结

TBMQ / EMQX 只当 Broker：节点连上后订阅，再按 Topic 映射到 TB 设备，后续可接 Save Timeseries 等节点。