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(); // 关闭通道
}
}
2.5、通过TCP链接工具
链接到服务器：localhost，端口:8092上
发送区发送数据：
{time:”2024年5月5日”}
2.6、查看遥测数据
