Springboot+Netty实现基于天翼物联网平台CTWing(AIOT)终端TCP协议(透传模式)-设备终端（南向设备）
使用Springboot+Netty方式模拟天翼物联网平台CTWing(AIOT)的终端设备，模拟的是AIOT平台的TCP协议(透传模式) 原来的AEP平台
·
电信的天翼物联网平台CTWing(AIOT)原先是我们俗称的aep，主要用于接入nb-iot设备，当然也可以接入其他的设备，在熟悉AIOT平台后，做后端的我有时候急需终端样品(智能门禁，支付识别终端，CPE终端，考勤机等)做北向订阅的应用开发，可南向设备的开发需要一段时间，因此可以使用其他办法，工具啥的模拟终端设备进行数据交互的开发。
对于实时性要求高的设备，比如智能门禁机，当触发需要开门的请求后，需要立即给设备发送开门指令(信号)，那么常用的tcp协议成为最好最简单快捷的一种选择方式。
对于天翼物联网平台的基础使用(注册，登录，创建产品)就不记录了，直接创建产品
要求：设备直连 TCP协议 明文 特征字符串 一型一密 透传 分类可选择智慧社区的配件
在这里我们首先获取到产品ID和Master-APIkey，在点击详情里面
产品ID：15506850
Master-APIkey：d894a1c38274440986dd4f4cc3a7799a
特征串：IicRLZ58eW_4LYp5EUIKdJcqyL5DU7XuepoQaV4U-SY
一般设备最长使用imei作为唯一的标识来进行通讯的，这里我模拟一个imei号码注册，拿到该设备的认证信息
869401041201815
这里有一个点，设备ID是产品ID+设备编号(imei)拼接而成
deviceId: 15506850869401041201815这些参数在终端程序中是需要使用到的，因此先行拿出来放着
在看AIOT平台对于TCP的协议，透传模式和非透传模式
tcp数据协议的地址（官网可找）接口介绍-中国电信天翼物联网CTWing门户网站
上图是协议的关键，至于协议的业务交互流程，AIOT有文档里已经给出了，还有示例，因此直接上使用Springboot+Netty模拟此协议的代码
新建Springboot的maven项目，pom.xml文件导入依赖包（用到了swagger来测试终端上报数据）
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.5.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>boot.ctwing.tcp.terminal</groupId>
<artifactId>boot-example-ctwing-tcp-terminal-2.0.5</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>boot-example-ctwing-tcp-terminal-2.0.5</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.29.Final</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>com.github.xiaoymin</groupId>
<artifactId>swagger-bootstrap-ui</artifactId>
<version>1.9.2</version>
</dependency>
</dependencies>
<build>
<plugins>
<!-- 打包成一个可执行jar -->
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
Springboot启动类，Netty启动
package boot.ctwing.tcp.terminal;
import boot.ctwing.tcp.terminal.config.CtWingConstant;
import boot.ctwing.tcp.terminal.netty.TcpClient;
import boot.ctwing.tcp.terminal.utils.CtWingUtils;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;
/**
* 蚂蚁舞
*/
@SpringBootApplication
@EnableScheduling
public class BootCtWingTcpTerminal implements CommandLineRunner {
public static void main( String[] args ) throws Exception {
SpringApplication.run(BootCtWingTcpTerminal.class, args);
System.out.println( "Hello World!" );
}
@Override
public void run(String... args) throws Exception {
byte[] bytes = CtWingUtils.tcp_01_auth();
new TcpClient().startup(CtWingConstant.port, CtWingConstant.address, bytes);
// int port = 8996;
// new IotTcpClient().connect(port, "127.0.0.1");
}
}
SwaggerConfig配置
package boot.ctwing.tcp.terminal.config;
import com.google.common.base.Predicates;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.builders.PathSelectors;
import springfox.documentation.builders.RequestHandlerSelectors;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;
/**
* 蚂蚁舞
*/
@Configuration
@EnableSwagger2
public class SwaggerConfig {
@Bean
public Docket createRestApi(){
return new Docket(DocumentationType.SWAGGER_2).apiInfo(apiInfo()).select()
.apis(RequestHandlerSelectors.any()).paths(PathSelectors.any())
.paths(Predicates.not(PathSelectors.regex("/error.*")))
.paths(PathSelectors.regex("/.*"))
.build().apiInfo(apiInfo());
}
private ApiInfo apiInfo(){
return new ApiInfoBuilder()
.title("天翼物联网CtWing终端模拟mock")
.description("终端模拟需要的测试接口")
.version("0.01")
.build();
}
/**
* http://localhost:8177/doc.html 地址和端口根据实际项目查看
*/
}
CtWingConstant静态类，里面包括产品Id，特征字符串和设备imei
package boot.ctwing.tcp.terminal.config;
/**
* 蚂蚁舞
*/
public class CtWingConstant {
// 产品ID
public static final String productId = "15506850";
// 设备imei
public static final String imei = "869401041201815";
// 特征字符串
public static final String password = "IicRLZ58eW_4LYp5EUIKdJcqyL5DU7XuepoQaV4U-SY";
public static final String version = "1.0";
public static final String address = "tcp.ctwing.cn";
public static final int port = 8996;
// 登录认证
public static final String tcp_hex_01 = "01";
// 上行数据报文
public static final String tcp_hex_02 = "02";
// 下行数据报文
public static final String tcp_hex_03 = "03";
// 上行心跳
public static final String tcp_hex_04 = "04";
// 登录响应
public static final String tcp_hex_05 = "05";
// 心跳响应
public static final String tcp_hex_06 = "06";
}
TcpClient
package boot.ctwing.tcp.terminal.netty;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 蚂蚁舞
*/
public class TcpClient {
private final Logger log = LoggerFactory.getLogger(this.getClass());
public static SocketChannel socketChannel = null;
private static final EventLoopGroup group = new NioEventLoopGroup();
public void startup(int port, String host, byte[] bytes) throws Exception{
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group);
bootstrap.channel(NioSocketChannel.class);
bootstrap.option(ChannelOption.TCP_NODELAY, true);
bootstrap.handler(new TcpChannelInitializer<SocketChannel>());
ChannelFuture f = bootstrap.connect(host, port).sync();
if (f.isSuccess()) {
socketChannel = (SocketChannel) f.channel();
log.info("connect server success");
f.channel().writeAndFlush(Unpooled.buffer().writeBytes(bytes));
log.info("send success");
f.channel().closeFuture().sync();
}
} catch (Exception e) {
System.out.println(e.toString());
} finally {
group.shutdownGracefully().sync();
}
}
}
TcpChannelInitializer<SocketChannel>
package boot.ctwing.tcp.terminal.netty;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
/**
* 蚂蚁舞
* @param <SocketChannel>
*/
public class TcpChannelInitializer<SocketChannel> extends ChannelInitializer<Channel>{
@Override
protected void initChannel(Channel ch) throws Exception {
// 二者选择一个就可以
// 使用netty自带的
// ch.pipeline().addLast("decoder", new ByteArrayDecoder());
// ch.pipeline().addLast("encoder", new ByteArrayEncoder());
// 使用自定义的
ch.pipeline().addLast(new TcpMessageCodec());
ch.pipeline().addLast(new TcpChannelInboundHandlerAdapter());
}
}
TcpChannelInboundHandlerAdapter
package boot.ctwing.tcp.terminal.netty;
import java.io.IOException;
import java.net.InetSocketAddress;
import boot.ctwing.tcp.terminal.config.CtWingConstant;
import boot.ctwing.tcp.terminal.utils.CtWingUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 蚂蚁舞
*/
public class TcpChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{
private final Logger log = LoggerFactory.getLogger(this.getClass());
/**
* 从服务端收到新的数据时，这个方法会在收到消息时被调用
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException {
byte[] req = (byte[]) msg;
if(req.length > 0){
String hex = CtWingUtils.bytesToHexStr(req);
log.info("data--"+hex);
String soh = hex.substring(0,2);
switch (soh) {
case CtWingConstant.tcp_hex_06:
// 平台回复终端心跳的响应 保活用
break;
case CtWingConstant.tcp_hex_05:
// 认证消息返回 05 00 00 登录结果： 0 成功 1 未知错误 2 设备未注册 3 设备认证失败 4 设备已登录
// to do
break;
case CtWingConstant.tcp_hex_03:
// 0x03 +数据长度（2字节） +业务数据 下行数据，处理下行逻辑
String dataHex = hex.substring(6);
log.info("hexStr--"+dataHex);
// 如果是字符串 16进制字符串转字符串
log.info("str--"+CtWingUtils.hexStrToStr(dataHex));
default:
break;
}
}
}
/**
* 从服务端收到新的数据、读取完成时调用
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
System.out.println("channelReadComplete");
}
/**
* 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException {
System.out.println("exceptionCaught");
cause.printStackTrace();
ctx.close();//抛出异常，断开与客户端的连接
}
/**
* 客户端与服务端第一次建立连接时 执行
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException {
System.out.println("channelActive");
super.channelActive(ctx);
}
/**
* 客户端与服务端 断连时 执行
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelInactive(ctx);
InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = inSocket.getAddress().getHostAddress();
ctx.close(); //断开连接时，必须关闭，否则造成资源浪费
System.out.println("channelInactive:"+clientIp);
}
}
TcpMessageCodec
package boot.ctwing.tcp.terminal.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import java.util.List;
/**
* 蚂蚁舞
*/
@ChannelHandler.Sharable
public class TcpMessageCodec extends MessageToMessageCodec<ByteBuf, ByteBuf> {
@Override
protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
byte[] array = new byte[msg.readableBytes()];
msg.getBytes(0, array);
out.add(Unpooled.wrappedBuffer(array));
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
byte[] array = new byte[msg.readableBytes()];
msg.getBytes(0, array);
out.add(array);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
System.out.println("OutIn异常！"+cause);
}
}
TcpHeartTimer
package boot.ctwing.tcp.terminal.netty;
import boot.ctwing.tcp.terminal.config.CtWingConstant;
import boot.ctwing.tcp.terminal.utils.CtWingUtils;
import io.netty.buffer.Unpooled;
import io.netty.channel.socket.SocketChannel;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
/**
* 蚂蚁舞
*/
@Service
public class TcpHeartTimer {
// 使用定时器发送心跳
@Scheduled(cron = "0 0/3 * * * ?")
public void tcp_ct_wing_heart_timer() {
String back = CtWingConstant.tcp_hex_04;
byte[] data = CtWingUtils.hexStrToBytes(back);
SocketChannel socketChannel = TcpClient.socketChannel;
if( socketChannel != null && socketChannel.isOpen()) {
socketChannel.writeAndFlush(Unpooled.buffer().writeBytes(data));
}
}
}
CtWingUtils
package boot.ctwing.tcp.terminal.utils;
import boot.ctwing.tcp.terminal.config.CtWingConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
/**
* 蚂蚁舞
*/
public class CtWingUtils {
private static final Logger log = LoggerFactory.getLogger(CtWingUtils.class);
private static final char[] HEXES = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'};
public static String bytesToHexStr(byte[] bytes) {
if (bytes == null || bytes.length == 0) {
return null;
}
StringBuilder hex = new StringBuilder(bytes.length * 2);
for (byte b : bytes) {
hex.append(HEXES[(b >> 4) & 0x0F]);
hex.append(HEXES[b & 0x0F]);
}
return hex.toString().toUpperCase();
}
public static byte[] hexStrToBytes(String hex) {
if (hex == null || hex.length() == 0) {
return null;
}
char[] hexChars = hex.toCharArray();
byte[] bytes = new byte[hexChars.length / 2]; // 如果 hex 中的字符不是偶数个, 则忽略最后一个
for (int i = 0; i < bytes.length; i++) {
bytes[i] = (byte) Integer.parseInt("" + hexChars[i * 2] + hexChars[i * 2 + 1], 16);
}
return bytes;
}
public static String strToHexStr(String str) {
StringBuilder sb = new StringBuilder();
byte[] bs = str.getBytes();
int bit;
for (int i = 0; i < bs.length; i++) {
bit = (bs[i] & 0x0f0) >> 4;
sb.append(HEXES[bit]);
bit = bs[i] & 0x0f;
sb.append(HEXES[bit]);
}
return sb.toString().trim();
}
public static String hexStrToStr(String hexStr) {
//能被16整除,肯定可以被2整除
byte[] array = new byte[hexStr.length() / 2];
try {
for (int i = 0; i < array.length; i++) {
array[i] = (byte) (0xff & Integer.parseInt(hexStr.substring(i * 2, i * 2 + 2), 16));
}
hexStr = new String(array, StandardCharsets.UTF_8);
} catch (Exception e) {
e.printStackTrace();
return "";
}
return hexStr;
}
public static String hexLen4Calc(int fixed, int len){
StringBuilder x = new StringBuilder(Integer.toHexString(len));
int xC = fixed - x.length();
for(int i = 0; i< xC; i++){
x.insert(0, "0");
}
return x.toString();
}
public static byte[] tcp_01_auth(){
String deviceId = CtWingConstant.productId+CtWingConstant.imei;
String deviceIdLenHex = hexLen4Calc(4, deviceId.length());
String deviceIdHex = strToHexStr(deviceId);
String passwordLenHex = hexLen4Calc(4, CtWingConstant.password.length());
String passwordHex = strToHexStr(CtWingConstant.password);
String versionLenHex = hexLen4Calc(4, CtWingConstant.version.length());
String versionHex = strToHexStr(CtWingConstant.version);
String cmd = CtWingConstant.tcp_hex_01+deviceIdLenHex+deviceIdHex+passwordLenHex+passwordHex+versionLenHex+versionHex;
log.info(cmd);
return hexStrToBytes(cmd);
}
public static byte[] tcp_02_up_msg(String str){
String upHexStr = strToHexStr(str);
String upHexStrLenHex = hexLen4Calc(4, upHexStr.length()/2);
String cmd = CtWingConstant.tcp_hex_02+upHexStrLenHex+upHexStr;
log.info(cmd);
return hexStrToBytes(cmd);
}
}
TcpTerminalController
package boot.ctwing.tcp.terminal.controller;
import boot.ctwing.tcp.terminal.netty.TcpClient;
import boot.ctwing.tcp.terminal.utils.CtWingUtils;
import io.netty.buffer.Unpooled;
import io.netty.channel.socket.SocketChannel;
import org.springframework.web.bind.annotation.*;
@RestController
public class TcpTerminalController {
@GetMapping(value = {"", "/"})
public String index() {
return "天翼物联网CtWing终端模拟mock";
}
@GetMapping("/reportData")
public String reportData(@RequestParam(name="content",defaultValue="hello-myw-terminal") String content) {
byte[] data = CtWingUtils.tcp_02_up_msg(content);
SocketChannel socketChannel = TcpClient.socketChannel;
if( socketChannel != null && socketChannel.isOpen()) {
socketChannel.writeAndFlush(Unpooled.buffer().writeBytes(data));
}
return "success";
}
}
代码目录结构
boot-example-ctwing-tcp-terminal-2.0.5
│ pom.xml
│
└─src
├─main
│ ├─java
│ │ └─boot
│ │ └─ctwing
│ │ └─tcp
│ │ └─terminal
│ │ │ BootCtWingTcpTerminal.java
│ │ │
│ │ ├─config
│ │ │ CtWingConstant.java
│ │ │ SwaggerConfig.java
│ │ │
│ │ ├─controller
│ │ │ TcpTerminalController.java
│ │ │
│ │ ├─netty
│ │ │ TcpChannelInboundHandlerAdapter.java
│ │ │ TcpChannelInitializer.java
│ │ │ TcpClient.java
│ │ │ TcpHeartTimer.java
│ │ │ TcpMessageCodec.java
│ │ │
│ │ └─utils
│ │ CtWingUtils.java
│ │
│ └─resources
│ application.properties
│ logback-spring.xml
│
└─test
└─java
└─boot
└─ctwing
└─tcp
└─terminal
BootCtWingTcpTerminalTest.java
启动Springboot项目后可以看到最后的日志信息是这样的
20:31:55.987 spring-boot-logging [main] INFO o.a.tomcat.util.net.NioSelectorPool - Using a shared selector for servlet write/read
20:31:56.014 spring-boot-logging [main] INFO o.s.b.w.e.tomcat.TomcatWebServer - Tomcat started on port(s): 8177 (http) with context path ''
20:31:56.022 spring-boot-logging [main] INFO b.c.t.terminal.BootCtWingTcpTerminal - Started BootCtWingTcpTerminal in 7.813 seconds (JVM running for 8.702)
20:31:56.027 spring-boot-logging [main] INFO b.c.tcp.terminal.utils.CtWingUtils - 0100173135353036383530383639343031303431323031383135002b496963524c5a353865575f344c5970354555494b644a6371794c35445537587565706f51615634552d53590003312e30
channelActive
20:31:58.379 spring-boot-logging [main] INFO b.c.tcp.terminal.netty.TcpClient - connect server success
20:31:58.413 spring-boot-logging [main] INFO b.c.tcp.terminal.netty.TcpClient - send success
20:31:58.542 spring-boot-logging [nioEventLoopGroup-2-1] INFO b.c.t.t.n.TcpChannelInboundHandlerAdapter - data--050000
channelReadComplete
启动就开始认证
0100173135353036383530383639343031303431323031383135002b496963524c5a353865575f344c5970354555494b644a6371794c35445537587565706f51615634552d53590003312e30得到服务端的响应信息
050000
心跳返回是（定时器3分钟发一次（04），平台说的是5分钟内）
06此时我们再看AIOT平台的设备管理详情里面
显示已经激活了
认证成功 心跳也正常 那么开始发送数据，一般发送的数据在设备端是不含中文的，我这里把中文也带上，测试下发送中文是否可用可行，我的端口是8177 因此本地访问
http://localhost:8177/doc.html
我发送了好几条数据（字符串形式），回到AIOT平台的数据里查看
如此模拟设备的数据成功上传到天翼物联网平台AIOT
6JqC6JqB6Iie
6JqC6JqB5Lmf5Lya6Lez6Iie
bXl3
bXl5aHR3MTIzNDU0NjU0Njc0ZXF3amRjcW93ZWljcW93aXhjbmRjeA==四条数据是经过base64的，因此需要解开，在转成我发送的字符串就可以
package boot.ctwing.tcp.app;
import boot.ctwing.tcp.app.utils.CtWingUtils;
import java.util.Base64;
public class Test {
public static void main(String[] args) {
byte[] decoded1 = Base64.getDecoder().decode("6JqC6JqB6Iie");
String hex1 = CtWingUtils.bytesToHexStr(decoded1);
System.out.println(CtWingUtils.hexStrToStr(hex1));
byte[] decoded2 = Base64.getDecoder().decode("bXl3");
String hex2 = CtWingUtils.bytesToHexStr(decoded2);
System.out.println(CtWingUtils.hexStrToStr(hex2));
byte[] decoded3 = Base64.getDecoder().decode("6JqC6JqB5Lmf5Lya6Lez6Iie");
String hex3 = CtWingUtils.bytesToHexStr(decoded3);
System.out.println(CtWingUtils.hexStrToStr(hex3));
byte[] decoded4 = Base64.getDecoder().decode("bXl5aHR3MTIzNDU0NjU0Njc0ZXF3amRjcW93ZWljcW93aXhjbmRjeA==");
String hex4 = CtWingUtils.bytesToHexStr(decoded4);
System.out.println(CtWingUtils.hexStrToStr(hex4));
}
}
最终得到的字符串数据
蚂蚁舞
myw
蚂蚁也会跳舞
myyhtw123454654674eqwjdcqoweicqowixcndcx如此使用springboot+netty模拟天翼物联网CtWing的终端设备算是完成了。
更多推荐
0
0- 0
已为社区贡献1条内容
所有评论(0)