ThingsBoard如何接受设备通过TCP发送的报文

1、概述

2、案例

2.1、阐述

2.2、导入依赖

2.3、构建Netty服务链接，接受的端口为8092

2.4、对数据进行相应的处理发送到ThingsBoard客户端

2.5、通过TCP链接工具

2.6、查看遥测数据

1、概述

TCP（Transmission Control Protocol，传输控制协议）是互联网中的一种面向连接的、可靠的、基于字节流的传输层通信协议。它提供了点对点的通信，并且可靠地传递数据。TCP是Internet 协议族中最为重要、应用最为广泛的一种协议。

2、案例

ThinsBoard原生代码中，只支持 Coap、Https、Lwm2m、Mqtt、Snmp五种协议的传输格式，并不支持TCP的报文协议，但是很多场景下设备的发送协议就是TCP协议或者是UDP协议的情况，如何把这类的设备接收到ThingsBoard平台中，目前共有三种实现方式：

第一种 ：通过外部服务将报文转换成ThingsBoard原生代码支持的协议进行发送

第二种：通过ThingsBoard GateWay进行进行数据的转换

第三种：修改ThingsBoatd源码，创建Transport来支持TCP协议的数据

三种转换的方式各有利弊，再次并不推荐说那种好或者那种不好，本次文章只对第一种情况进行讲解，在不修改源代码的情况下，将TCP接收到的数据，发送到ThingsBoard平台

2.1、阐述

设备通过TCP发送消息，其实就是需要建立对应的Socket链接，我们使用Netty来暴露对应的端口，让设备往对应的端口上发送数据，服务接受到数据后，对数据进行解析，解析完成后，推送到ThingsBoard平台

2.2、导入依赖

        <dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.63.Final</version>
            </dependency>
            <dependency>
                <groupId>org.eclipse.paho</groupId>
                <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
                <version>1.2.5</version>
            </dependency>

2.3、构建Netty服务链接，接受的端口为8092

package com.example.tpson_tcp;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

/**
 * Netty服务端类，用于启动TCP服务器。
 */
@Service
public class NettyServer {

    /**
     * 在对象初始化后自动调用的方法，用于启动服务器。
     * 无参数和返回值。
     */
    @PostConstruct
    public void main() {
        // 创建EventLoopGroup用于处理网络事件
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 配置服务器启动参数
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class) // 指定使用的NIO通道类
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 设置通道初始化处理器
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // 在通道中添加处理链，用于解码、编码和处理设备数据
                            ch.pipeline().addLast(new StringDecoder());
                            ch.pipeline().addLast(new StringEncoder());
                            ch.pipeline().addLast(new DeviceDataHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128) // 设置连接队列大小
                    .childOption(ChannelOption.SO_KEEPALIVE, true); // 启用TCP KeepAlive

            // 绑定端口并启动服务器
            ChannelFuture channelFuture = serverBootstrap.bind(8092).sync();
            // 等待服务器关闭
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            // 异常处理，抛出运行时异常
            throw new RuntimeException(e);
        } finally {
            // 关闭EventLoopGroup，释放资源
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

2.4、对数据进行相应的处理发送到ThingsBoard客户端

mqtt的username是设备的ID，根据你们自己的设备ID进行修改

/**
 * 该类继承自SimpleChannelInboundHandler，用于处理设备数据。
 * 当设备数据到达时，会将数据发送到MQTT服务器。
 */
package com.example.tpson_tcp;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;

public class DeviceDataHandler extends SimpleChannelInboundHandler<String> {

    private IMqttClient mqttClient; // MQTT客户端接口

    /**
     * 构造函数：初始化MQTT客户端并连接到服务器。
     */
    public DeviceDataHandler() {
        try {
            mqttClient = new MqttClient("tcp://127.0.0.1:1883", MqttClient.generateClientId());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true); // 设置会话清除标志
            options.setUserName("001001001"); // 设置用户名
            mqttClient.connect(options); // 连接到MQTT服务器
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 当通道中有可读数据时调用该方法。
     *
     * @param ctx 通道上下文
     * @param msg 接收到的消息
     * @throws Exception 抛出的异常
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println("Received device data: " + msg);
        // 将接收到的设备数据发送到MQTT的主题
        MqttMessage message = new MqttMessage(msg.getBytes());
        System.out.println(message);
        mqttClient.publish("v1/devices/me/telemetry", message);
    }

    /**
     * 当捕获到异常时调用该方法。
     *
     * @param ctx 通道上下文
     * @param cause 异常原因
     * @throws Exception 抛出的异常
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close(); // 关闭通道
    }
}