优化netty连接

liusuyi

2024-10-23 2b43275bd260eda468f305ebca4b4f4291b0605d

优化netty连接

 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) {

 ard-work/src/main/java/com/ard/tianyu/netty/NettyConfig.java

File was deleted

 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;


 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());
        }
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeCommand.java

New file
@@ -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号
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/command/SubscribeReplyCommand.java

New file
@@ -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号
    }
}

 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());
            }

 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 数据

 ard-work/src/main/java/com/ard/tianyu/service/ITyService.java

@@ -58,6 +58,12 @@
     * 注册
     */
    void registerMessage();

    /**
     * 无人机控制订阅
     */
    void uavControlSubscribe(String boxSn);

    /**
     * 云台转动(基于角度)
     */

 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("收到无人机状态消息-->无人机状态消息");