From 2b43275bd260eda468f305ebca4b4f4291b0605d Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Wed, 23 Oct 2024 14:50:33 +0800
Subject: [PATCH] 优化netty连接
---
ard-work/src/main/java/com/ard/tianyu/netty/handler/ClientHandler.java | 8 +-
/dev/null | 39 ---------
ard-work/src/main/java/com/ard/tianyu/netty/TyNettyTcpClient.java | 50 ++++++------
ard-work/src/main/java/com/ard/tianyu/netty/handler/MessageProtocolDecoder.java | 7 +
ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeCommand.java | 26 ++++++
ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java | 23 +++++
ard-work/src/main/java/com/ard/tianyu/controller/UavController.java | 7 +
ard-work/src/main/java/com/ard/tianyu/service/ITyService.java | 6 +
ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeReplyCommand.java | 26 ++++++
ard-work/src/main/java/com/ard/tianyu/netty/TyBoxSession.java | 2
10 files changed, 123 insertions(+), 71 deletions(-)
diff --git a/ard-work/src/main/java/com/ard/tianyu/controller/UavController.java b/ard-work/src/main/java/com/ard/tianyu/controller/UavController.java
index 23488fe..aa35cba 100644
--- a/ard-work/src/main/java/com/ard/tianyu/controller/UavController.java
+++ b/ard-work/src/main/java/com/ard/tianyu/controller/UavController.java
@@ -69,7 +69,12 @@
Boolean state = tyService.updateBoxSettings(boxSettingsDTO);
return AjaxResult.success(state);
}
-
+ @ApiOperation("无人机控制订阅")
+ @GetMapping("/uavControlSubscribe")
+ public AjaxResult uavControlSubscribe(String boxSn) {
+ tyService.uavControlSubscribe(boxSn);
+ return AjaxResult.success();
+ }
@ApiOperation("云台转动(基于角度)")
@GetMapping("/controlAngle")
public AjaxResult controlAngle(String boxSn) {
diff --git a/ard-work/src/main/java/com/ard/tianyu/netty/NettyConfig.java b/ard-work/src/main/java/com/ard/tianyu/netty/NettyConfig.java
deleted file mode 100644
index 304caa2..0000000
--- a/ard-work/src/main/java/com/ard/tianyu/netty/NettyConfig.java
+++ /dev/null
@@ -1,39 +0,0 @@
-package com.ard.tianyu.netty;
-
-import com.ard.tianyu.config.TyConfiguration;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-import javax.annotation.PostConstruct;
-import javax.annotation.Resource;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-@Component
-@Slf4j
-public class NettyConfig {
-
- @Resource
- private TyConfiguration tianyuConfiguration;
-
- @Resource
- private TyNettyTcpClient tyNettyTcpClient;
-
- @PostConstruct
- public void createTyConnect() {
- try {
- // 将Tcp访问的IP和地址加入到字典
- Map<String,Object> mapParam =new HashMap<String, Object>();
- mapParam.put("IP",tianyuConfiguration.getTcpserver());
- mapParam.put("PORT",tianyuConfiguration.getTcpport());
-
- //要么从数据库查询信息，要么从配置文件读取信息
- List<Map<String, Object>> list = new ArrayList<>();
- list.add(mapParam);
- tyNettyTcpClient.initAllChannel(list);
- }catch (Exception e) {
- e.printStackTrace();
- }
- }
-}
diff --git a/ard-work/src/main/java/com/ard/tianyu/netty/TyBoxSession.java b/ard-work/src/main/java/com/ard/tianyu/netty/TyBoxSession.java
index 9173c53..eb82f17 100644
--- a/ard-work/src/main/java/com/ard/tianyu/netty/TyBoxSession.java
+++ b/ard-work/src/main/java/com/ard/tianyu/netty/TyBoxSession.java
@@ -14,7 +14,7 @@
public String subBoxSn;
- public ChannelFuture future;
+ public ChannelFuture channelFuture;
public Boolean isReConnect;
diff --git a/ard-work/src/main/java/com/ard/tianyu/netty/TyNettyTcpClient.java b/ard-work/src/main/java/com/ard/tianyu/netty/TyNettyTcpClient.java
index 02f2fa7..08065a6 100644
--- a/ard-work/src/main/java/com/ard/tianyu/netty/TyNettyTcpClient.java
+++ b/ard-work/src/main/java/com/ard/tianyu/netty/TyNettyTcpClient.java
@@ -1,5 +1,6 @@
package com.ard.tianyu.netty;
+import com.ard.tianyu.config.TyConfiguration;
import com.ard.tianyu.netty.handler.ClientHandler;
import com.ard.tianyu.netty.handler.MessageProtocolDecoder;
import com.ard.tianyu.netty.handler.MessageProtocolEncoder;
@@ -12,10 +13,9 @@
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
+import javax.annotation.PostConstruct;
import javax.annotation.Resource;
-import java.net.InetSocketAddress;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -28,17 +28,16 @@
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
@Resource
+ private TyConfiguration tyConfiguration;
+ @Resource
ClientHandler clientHandler;
/**
* 初始化Bootstrap
*/
- public Bootstrap getBootstrap(EventLoopGroup group) {
- if (null == group) {
- group = eventLoopGroup;
- }
+ private Bootstrap getBootstrap() {
Bootstrap bootstrap = new Bootstrap();
- bootstrap.group(group)
+ bootstrap.group(eventLoopGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.SO_KEEPALIVE, true)
@@ -55,28 +54,21 @@
return bootstrap;
}
- // 获取所有连接
- public void initAllChannel(List<Map<String, Object>> portHosts) {
- for (Map<String, Object> map : portHosts) {
- String host = (String) map.get("IP");
- Integer port = (Integer) map.get("PORT");
- initOneConnection(host, port);
- }
- }
-
- public void initOneConnection(String host, Integer port) {
+ /**
+ * 初始化连接
+ */
+ public void initConnection(String host, Integer port) {
log.info("初始化连接天宇tcp服务器【{}:{}】", host, port);
- //判断该账号下，是否有盒子被控制
- //当前正在被控制的盒子
String currentBoxSn = "";
+ //判断是否已经存在连接，如果存在则关闭之前的连接
if (channels.containsKey(host + ":" + port)) {
currentBoxSn = channels.get(host + ":" + port).getSubBoxSn();
- channels.get(host + ":" + port).getFuture().channel().close();
+ channels.get(host + ":" + port).getChannelFuture().channel().close();
}
//异步连接tcp服务端
- Bootstrap bootstrap = getBootstrap(null);
+ Bootstrap bootstrap = getBootstrap();
ChannelFuture future = bootstrap.connect(host, port).addListener((ChannelFuture futureListener) -> {
if (futureListener.isSuccess()) {
log.info("连接天宇TCP服务器成功【{}:{}】", host, port);
@@ -85,16 +77,24 @@
eventLoop.schedule(() -> {
log.info("连接天宇TCP服务器失败【{}:{}】,正尝试重连...", host, port);
try {
- initOneConnection(host, port);
+ initConnection(host, port);
} catch (Exception e) {
e.printStackTrace();
}
}, 5, TimeUnit.SECONDS);
}
});
+ //将连接存入集合
+ channels.put(host + ":" + port, TyBoxSession.builder().subBoxSn(currentBoxSn).channelFuture(future).isReConnect(!currentBoxSn.equals("")).build());
+ }
- //判断是否有盒子被订阅
- channels.put(host + ":" + port, TyBoxSession.builder().subBoxSn(currentBoxSn).future(future)
- .isReConnect(!currentBoxSn.equals("")).build());
+ //创建天宇连接
+ @PostConstruct
+ public void createTyConnect() {
+ try {
+ initConnection(tyConfiguration.getTcpserver(), tyConfiguration.getTcpport());
+ } catch (Exception e) {
+ log.error("创建TCP连接异常:{}", e.getMessage());
+ }
}
}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeCommand.java b/ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeCommand.java
new file mode 100644
index 0000000..962e88a
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeCommand.java
@@ -0,0 +1,26 @@
+package com.ard.tianyu.netty.command;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Data;
+
+@Data
+public class SubscribeCommand extends BaseCommand {
+
+ //云盒SN号
+ private String BoxSn;
+
+ public SubscribeCommand(String BoxSn) {
+ setDataHeader((short) 0x7479); // 设置帧头
+ setDataLength((short) (6 + BoxSn.getBytes().length)); // 数据长度
+ setCmdCode((byte) 0x03); // 设置命令编号
+ this.BoxSn = BoxSn;
+ }
+
+ @Override
+ public void encode(ByteBuf out) {
+ out.writeShort(getDataHeader()); // 写入帧头
+ out.writeShort(getDataLength()); // 写入数据长度
+ out.writeByte(getCmdCode()); // 写入指令编号
+ out.writeBytes(BoxSn.getBytes()); // 写入云盒SN号
+ }
+}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeReplyCommand.java b/ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeReplyCommand.java
new file mode 100644
index 0000000..62cb5c7
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeReplyCommand.java
@@ -0,0 +1,26 @@
+package com.ard.tianyu.netty.command;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Data;
+
+@Data
+public class SubscribeReplyCommand extends BaseCommand {
+
+ //云盒SN号
+ private String BoxSn;
+
+ public SubscribeReplyCommand(String BoxSn) {
+ setDataHeader((short) 0x6A77); // 设置帧头
+ setDataLength((short) (6 + BoxSn.getBytes().length)); // 数据长度
+ setCmdCode((byte) 0x03); // 设置命令编号
+ this.BoxSn = BoxSn;
+ }
+
+ @Override
+ public void encode(ByteBuf out) {
+ out.writeShort(getDataHeader()); // 写入帧头
+ out.writeShort(getDataLength()); // 写入数据长度
+ out.writeByte(getCmdCode()); // 写入指令编号
+ out.writeBytes(BoxSn.getBytes()); // 写入云盒SN号
+ }
+}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/netty/handler/ClientHandler.java b/ard-work/src/main/java/com/ard/tianyu/netty/handler/ClientHandler.java
index 3fc7b5f..a84e551 100644
--- a/ard-work/src/main/java/com/ard/tianyu/netty/handler/ClientHandler.java
+++ b/ard-work/src/main/java/com/ard/tianyu/netty/handler/ClientHandler.java
@@ -56,9 +56,9 @@
if (event.state() == IdleState.READER_IDLE) {
log.info("客户端读超时");
} else if (event.state() == IdleState.WRITER_IDLE) {
- log.info("写空闲发送⼼跳!");
- HeartbeatCommand msg = new HeartbeatCommand(System.currentTimeMillis());
- ctx.writeAndFlush(msg);
+ // log.info("写空闲发送⼼跳!");
+ // HeartbeatCommand msg = new HeartbeatCommand(System.currentTimeMillis());
+ //ctx.writeAndFlush(msg);
} else if (event.state() == IdleState.ALL_IDLE) {
log.info("客户端所有操作超时");
}
@@ -75,7 +75,7 @@
ctx.channel().eventLoop().schedule(() -> {
log.error("天宇服务端链接不上，开始重连操作...");
try {
- tyNettyTcpClient.initOneConnection(host, port);
+ tyNettyTcpClient.initConnection(host, port);
} catch (Exception e) {
log.error("天宇服务端连接异常: {}", e.getMessage());
}
diff --git a/ard-work/src/main/java/com/ard/tianyu/netty/handler/MessageProtocolDecoder.java b/ard-work/src/main/java/com/ard/tianyu/netty/handler/MessageProtocolDecoder.java
index 4dcb9e0..2867a9b 100644
--- a/ard-work/src/main/java/com/ard/tianyu/netty/handler/MessageProtocolDecoder.java
+++ b/ard-work/src/main/java/com/ard/tianyu/netty/handler/MessageProtocolDecoder.java
@@ -42,6 +42,13 @@
baseCommand = new HeartbeatReplyCommand(timestamp);
log.debug("心跳回复: 时间戳 {}", timestamp);
break;
+ case 0x03:
+ case 0x04:
+ byte[] boxSn = new byte[dataLength - 1];
+ in.readBytes(boxSn);
+ SubscribeCommand payload = new SubscribeCommand(new String(boxSn));
+ log.debug("订阅回复: {}", payload.getBoxSn());
+ break;
case (byte) 0xA8:
byte[] uavStatus = new byte[dataLength - 1];
in.readBytes(uavStatus); // 读取 ProtoBuf 数据
diff --git a/ard-work/src/main/java/com/ard/tianyu/service/ITyService.java b/ard-work/src/main/java/com/ard/tianyu/service/ITyService.java
index 85f64d5..55d919d 100644
--- a/ard-work/src/main/java/com/ard/tianyu/service/ITyService.java
+++ b/ard-work/src/main/java/com/ard/tianyu/service/ITyService.java
@@ -58,6 +58,12 @@
* 注册
*/
void registerMessage();
+
+ /**
+ * 无人机控制订阅
+ */
+ void uavControlSubscribe(String boxSn);
+
/**
* 云台转动(基于角度)
*/
diff --git a/ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java b/ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java
index b18e6d8..9ec76ae 100644
--- a/ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java
+++ b/ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java
@@ -119,11 +119,19 @@
}
/**
+ * 无人机控制订阅
+ */
+ public void uavControlSubscribe(String boxSn) {
+ SubscribeCommand subscribeCommand = new SubscribeCommand(boxSn);
+ this.sendMessage(tyConfiguration.getIpPort(), subscribeCommand);
+ }
+
+ /**
* 发送消息
*/
public void sendMessage(String key, BaseCommand msg) {
try {
- ChannelFuture future = TyNettyTcpClient.channels.get(key).getFuture();
+ ChannelFuture future = TyNettyTcpClient.channels.get(key).getChannelFuture();
Channel channel = future.channel();
if (channel.isActive()) {
InetSocketAddress ipSocket = (InetSocketAddress) channel.remoteAddress();
@@ -142,6 +150,9 @@
}
}
+ /**
+ * 云台控制基于角度
+ */
@Override
public void controlAngle(String boxSn) {
ControlAngleCommand controlAngleCommand = new ControlAngleCommand(boxSn.getBytes(), (byte) 0x00, (byte) 0x00, (byte) 0x05);
@@ -167,6 +178,16 @@
HeartbeatReplyCommand heartbeatReplyCommand = (HeartbeatReplyCommand) tyCommand;
log.info("收到心跳消息-->处理心跳消息{}", heartbeatReplyCommand.getTimestamp());
break;
+ case 0x03:
+ // 处理订阅消息
+ SubscribeCommand subscribeCommand = (SubscribeCommand) tyCommand;
+ log.info("收到订阅消息-->处理订阅消息{}", subscribeCommand.getBoxSn());
+ break;
+ case 0x04:
+ // 处理取消订阅消息
+ subscribeCommand = (SubscribeCommand) tyCommand;
+ log.info("收到取消订阅消息-->处理取消订阅消息{}", subscribeCommand.getBoxSn());
+ break;
case (byte) 0xA8:
// 处理无人机状态消息
log.info("收到无人机状态消息-->无人机状态消息");
--
Gitblit v1.9.3