From fe21f972ae17b9fc37cc03287bff2e3c89e26b0e Mon Sep 17 00:00:00 2001
From: linrf <linrf@qq.com>
Date: Tue, 8 Aug 2023 17:34:32 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A4=84=E7=90=86netty=20=E4=B8=8B=E5=8F=91?=
 =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E7=B2=98=E5=8C=85=E5=88=86=E5=8C=85=E9=97=AE?=
 =?UTF-8?q?=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../server/handler/NettyBusinessHandler.java  | 15 +----------
 .../server/handler/NettyHeartbeatHandler.java | 14 +---------
 .../netty/client/handler/ClientHandler.java   | 27 ++++++++-----------
 .../client/handler/HeartbeatHandler.java      |  2 --
 .../server/handler/NettyBusinessHandler.java  | 25 +++--------------
 5 files changed, 17 insertions(+), 66 deletions(-)

diff --git a/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/handler/NettyBusinessHandler.java b/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/handler/NettyBusinessHandler.java
index 5da0ee8..6a3c84e 100644
--- a/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/handler/NettyBusinessHandler.java
+++ b/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/handler/NettyBusinessHandler.java
@@ -36,20 +36,7 @@ public class NettyBusinessHandler extends SimpleChannelInboundHandler<CommMsg<Se
 
     @Override
     protected void channelRead0(ChannelHandlerContext channelHandlerContext, CommMsg commMsg) throws Exception {
-//        ByteBuf buf = (ByteBuf) msg;
-//        //创建目标大小的数组
-//        byte[] barray = new byte[buf.readableBytes()];
-//        //把数据从bytebuf转移到byte[]
-//        buf.getBytes(0, barray);
-//        //将byte[]转成字符串用于打印
-//        String message = new String(barray);
-//
-//        //空消息不处理
-//        if (!StringUtils.hasText(message)) {
-//            return;
-//        }
-//
-//        CommMsg commMsg = JSON.fromJSON(message, CommMsg.class);
+
 
         String messageType = commMsg.getMessageType();
         String messageTime = commMsg.getMessageTime();
diff --git a/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/handler/NettyHeartbeatHandler.java b/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/handler/NettyHeartbeatHandler.java
index a70302b..867afa4 100644
--- a/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/handler/NettyHeartbeatHandler.java
+++ b/collector-scheduling-management/src/main/java/com/docus/server/common/netty/server/handler/NettyHeartbeatHandler.java
@@ -24,19 +24,7 @@ public class NettyHeartbeatHandler extends ChannelInboundHandlerAdapter {
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-//        ByteBuf buf = (ByteBuf) msg;
-//        //创建目标大小的数组
-//        byte[] barray = new byte[buf.readableBytes()];
-//        //把数据从bytebuf转移到byte[]
-//        buf.getBytes(0, barray);
-//        //将byte[]转成字符串用于打印
-//        String message = new String(barray);
-//        //空消息不处理
-//        if (!StringUtils.hasText(message)) {
-//            return;
-//        }
-
-//        CommMsg commMsg = JSON.fromJSON((String) message, CommMsg.class);
+
 
         CommMsg commMsg = (CommMsg) msg;
         String messageType = commMsg.getMessageType();
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/ClientHandler.java b/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/ClientHandler.java
index e6ebb0f..9cd858f 100644
--- a/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/ClientHandler.java
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/ClientHandler.java
@@ -31,20 +31,6 @@ public class ClientHandler extends SimpleChannelInboundHandler<CommMsg<Serializa
 
     @Override
     protected void channelRead0(ChannelHandlerContext context, CommMsg commMsg) throws Exception {
-//        ByteBuf buf = (ByteBuf) msg;
-//        //创建目标大小的数组
-//        byte[] barray = new byte[buf.readableBytes()];
-//        //把数据从bytebuf转移到byte[]
-//        buf.getBytes(0, barray);
-//        //将byte[]转成字符串用于打印
-//        String message = new String(barray);
-//
-//        //空消息不处理
-//        if (!StringUtils.hasText(message)) {
-//            return;
-//        }
-
-//        CommMsg commMsg = JSON.fromJSON(message, CommMsg.class);
 
         String messageType = commMsg.getMessageType();
         String messageTime = commMsg.getMessageTime();
@@ -98,9 +84,18 @@ public class ClientHandler extends SimpleChannelInboundHandler<CommMsg<Serializa
                     if (isExistProcess) {
                         collectorChannelCacheMap.writeAndFlush(collectorId, commMsg);
                     } else {
-                        //不存在进程，则启动指定采集器进程
+                        //不存在进程，则启动指定采集器进程,等待启动完成下发任务
                         StartUpExeUtils.startUpExeOnly(saveCollectorPackagePath + collectorId + "\\collector\\" + processName);
-                        collectorChannelCacheMap.writeAndFlush(collectorId, commMsg);
+
+                        while (true) {
+                            Channel channel = collectorChannelCacheMap.get(collectorId);
+                            log.info("等待采集器={}启动中", collectorId);
+                            if (null != channel && channel.isOpen()) {
+                                log.info("等待采集器={}启动成功并注册到终端", collectorId);
+                                collectorChannelCacheMap.writeAndFlush(collectorId, commMsg);
+                                break;
+                            }
+                        }
                     }
                 }
             }
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/HeartbeatHandler.java b/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/HeartbeatHandler.java
index 9805364..899997f 100644
--- a/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/HeartbeatHandler.java
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/netty/client/handler/HeartbeatHandler.java
@@ -43,7 +43,6 @@ public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
             .content(appName + " 我来了")
             .build();
 
-//        ctx.writeAndFlush(Unpooled.copiedBuffer(JSON.toJSON(onlineRegister), CharsetUtil.UTF_8));
         ctx.writeAndFlush(onlineRegister);
         super.channelActive(ctx);
     }
@@ -62,7 +61,6 @@ public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
                     .build();
 
                 //发送心跳消息，并在发送失败时关闭该接连
-//                ctx.writeAndFlush(Unpooled.copiedBuffer(JSON.toJSON(heartbeat), CharsetUtil.UTF_8)).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                 ctx.writeAndFlush(heartbeat).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
             }
         } else {
diff --git a/collector-terminal-management/src/main/java/com/docus/server/common/netty/server/handler/NettyBusinessHandler.java b/collector-terminal-management/src/main/java/com/docus/server/common/netty/server/handler/NettyBusinessHandler.java
index 4a959af..9395903 100644
--- a/collector-terminal-management/src/main/java/com/docus/server/common/netty/server/handler/NettyBusinessHandler.java
+++ b/collector-terminal-management/src/main/java/com/docus/server/common/netty/server/handler/NettyBusinessHandler.java
@@ -9,7 +9,6 @@ import com.docus.server.common.MsgConstants;
 import com.docus.server.common.netty.server.CollectorChannelCacheMap;
 import com.docus.server.common.netty.state.DeviceStateContext;
 import com.docus.server.dto.scheduling.management.schcollectrecord.EditSchCollectRecordDTO;
-import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -17,7 +16,6 @@ import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.handler.timeout.IdleStateEvent;
-import io.netty.util.CharsetUtil;
 import io.netty.util.concurrent.GlobalEventExecutor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -45,20 +43,6 @@ public class NettyBusinessHandler extends SimpleChannelInboundHandler<CommMsg<Se
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, CommMsg commMsg) throws Exception {
         DeviceStateContext deviceStateContext = collectorChannelCacheMap.getAttributeValue(ctx);
-//        ByteBuf buf = (ByteBuf) msg;
-//        //创建目标大小的数组
-//        byte[] barray = new byte[buf.readableBytes()];
-//        //把数据从bytebuf转移到byte[]
-//        buf.getBytes(0, barray);
-//        //将byte[]转成字符串用于打印
-//        String message = new String(barray);
-//
-//        //空消息不处理
-//        if (!StringUtils.hasText(message)) {
-//            return;
-//        }
-
-//        CommMsg commMsg = JSON.fromJSON(message, CommMsg.class);
 
         String messageType = commMsg.getMessageType();
         String messageTime = commMsg.getMessageTime();
@@ -68,12 +52,12 @@ public class NettyBusinessHandler extends SimpleChannelInboundHandler<CommMsg<Se
             //===========login ok,切换到已登录状态===============messageContent=collectorId=========
             deviceStateContext.onLoginSucc(messageContent, System.currentTimeMillis(), 10, "采集器认证通过");
 
-            CommMsg<Serializable> login_ok = CommMsg.builder()
+            CommMsg<Serializable> authMsg = CommMsg.builder()
                 .messageTime(DateUtil.formatDateTime(new Date()))
-                .content("login ok")
+                .content("采集器认证通过")
                 .messageType("response")
                 .build();
-            ctx.writeAndFlush(Unpooled.copiedBuffer(JSON.toJSON(login_ok), CharsetUtil.UTF_8));
+            ctx.writeAndFlush(authMsg);
 
 
             Channel channel = collectorChannelCacheMap.get(messageContent);
@@ -97,7 +81,7 @@ public class NettyBusinessHandler extends SimpleChannelInboundHandler<CommMsg<Se
                 .content("HEARTBEAT_REQUEST ok")
                 .messageType("response HEARTBEAT_REQUEST")
                 .build();
-            ctx.writeAndFlush(Unpooled.copiedBuffer(JSON.toJSON(resp), CharsetUtil.UTF_8));
+            ctx.writeAndFlush(resp);
 
 
             log.info("接受到【终端服务端-接收采集器消息】客户端的心跳消息：消息类型={}，消息时间={}，消息内容={}", messageType, messageTime, messageContent);
@@ -130,7 +114,6 @@ public class NettyBusinessHandler extends SimpleChannelInboundHandler<CommMsg<Se
             log.info("【终端服务端-接收采集器消息】收到采集器上报的任务状态，内容{}", messageContent);
             EditSchCollectRecordDTO editSchCollectRecordDTO = JSON.fromJSON(JSON.toJSON(messageContent), EditSchCollectRecordDTO.class);
             schCollectRecordApi.edit(editSchCollectRecordDTO);
-            System.out.println(1);
         }