ThingsBoard 如何自定义 UDP Transport

正文与配图来自原稿。可参考 Netty UDP 实现，并对照 自定义 TCP Transport 的模块接入方式。

0、参考

按 Netty 实现 UDP 服务的思路，在 ThingsBoard 里增加 transport.udp。可参考：Netty Simple UDP MessageDecoder。联调工具可用卓岚 TCP&UDP 调试助手。

1、在 common/transport 下创建 udp 模块

仿 MQTT 建子模块：

pom 可拷贝 mqtt，将 artifactId 改为 udp，依赖保留 transport-api、netty 等（与 MQTT 模块同级）。

2、核心类

包结构示意：

UdpTransportService

绑定地址端口，启动 NioDatagramChannel：

package org.thingsboard.server.transport.udp;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.util.ResourceLeakDetector;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.TbTransportService;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Service("UdpTransportService")
@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.udp.enabled}'=='true')")
@Slf4j
public class UdpTransportService implements TbTransportService {

    @Value("${transport.udp.bind_address}")
    private String host;
    @Value("${transport.udp.bind_port}")
    private Integer port;
    @Value("${transport.udp.netty.leak_detector_level}")
    private String leakDetectorLevel;
    @Value("${transport.udp.netty.boss_group_thread_count}")
    private Integer bossGroupThreadCount;

    @Autowired
    private UdpTransportContext context;

    private Channel serverChannel;
    private EventLoopGroup bossGroup;

    @PostConstruct
    public void init() throws Exception {
        log.info("Setting resource leak detector level to {}", leakDetectorLevel);
        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.valueOf(leakDetectorLevel.toUpperCase()));
        log.info("Starting UDP transport...");
        bossGroup = new NioEventLoopGroup(bossGroupThreadCount);
        Bootstrap b = new Bootstrap();
        b.group(bossGroup)
                .channel(NioDatagramChannel.class)
                .handler(new UdpTransportServerInitializer(context))
                .option(ChannelOption.SO_BROADCAST, true);
        serverChannel = b.bind(host, port).sync().channel();
        log.info("Udp transport started!");
    }

    @PreDestroy
    public void shutdown() throws InterruptedException {
        log.info("Stopping UDP transport!");
        try {
            serverChannel.close().sync();
        } finally {
            bossGroup.shutdownGracefully();
        }
        log.info("UDP transport stopped!");
    }

    @Override
    public String getName() {
        return DataConstants.UDP_TRANSPORT_NAME;
    }
}

辅

需在 DataConstants（或等价处）增加 UDP_TRANSPORT_NAME 常量，与 MQTT/HTTP 命名方式一致。

UdpTransportServerInitializer / Handler / Context

Pipeline：MessageDecoder → UdpTransportHandler。Handler 示例中打印报文并用 DatagramPacket 回写 "1"；后续可在此解析设备凭证并调用 TransportService 上报（对齐 MQTT Session）。

UdpTransportContext 继承 TransportContext，条件注解建议改为依赖 transport.udp.enabled（原稿曾误写成 mqtt 开关，联调时注意）。

Message / MessageDecoder：把 DatagramPacket 转成带 sender 的业务对象，便于回包。

3、thingsboard.yml

udp:
  enabled: "${UDP_ENABLED:true}"
  bind_address: "${UDP_BIND_ADDRESS:0.0.0.0}"
  bind_port: "${UDP_BIND_PORT:6789}"
  netty:
    leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}"
    boss_group_thread_count: "${NETTY_BOSS_GROUP_THREADS:1}"
    worker_group_thread_count: "${NETTY_WORKER_GROUP_THREADS:12}"

挂在 transport: 下，与 mqtt/http 并列。

4、独立 udp 启动模块（可选）

微服务风格可再建 transport/udp 启动工程：

POM 引入 org.thingsboard.common.transport:udp，mainClass 指向 ThingsboardUdpTransportApplication，并准备 conf/logback.xml、thingsboard-udp-transport.yml 等（可参考 mqtt-transport 启动模块拷贝后改名）。

5、接入 application / 根 POM

application 增加：

<dependency>
  <groupId>org.thingsboard.common.transport</groupId>
  <artifactId>udp</artifactId>
</dependency>

主工程按需增加 common udp 与 msa docker-info（与官方 mqtt 一致）。

6、启动与联调

启动后日志应出现 UDP transport started：

用 UDP 客户端工具连 bind_port（如 6789）：

发送测试数据（如数字 1，按需关闭十六进制发送），应收到服务端响应：

7、小结

1. common/transport/udp + Netty Datagram。
2. yml 打开 transport.udp.enabled。
3. 单体依赖 application，或独立 transport 进程。
4. Handler 内补齐鉴权与 TransportService 上报后，才与官方设备会话体系打通。