Преглед на файлове

完善启动后的的握手(发送项目ID)

lgy преди 4 години
родител
ревизия
ea6d86c349

+ 8 - 6
src/main/java/com/persagy/client/GroupNettyClient.java

@@ -14,6 +14,7 @@ import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import io.netty.handler.codec.LengthFieldPrepender;
 import io.netty.handler.codec.string.StringDecoder;
 import io.netty.handler.codec.string.StringEncoder;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
@@ -28,13 +29,14 @@ import java.util.concurrent.ConcurrentHashMap;
  * @date 2020年8月31日 上午10:34:45
  */
 @Service
+@Slf4j
 public class GroupNettyClient {
     @Autowired
     AlarmDefineService alarmDefineService;
     @Autowired
     AlarmRecordRepository alarmRecordRepository;
     static Bootstrap groupBootstrap = new Bootstrap();
-    static Channel channelGroup;
+    public static Channel channelGroup;
     @Value("${group.alarm.host:127.0.0.1}")
     public String host;
     @Value("${group.alarm.port}")
@@ -102,14 +104,14 @@ public class GroupNettyClient {
 
     public void connect(GroupNettyClient groupNettyClient, Channel channel) {
         // 加入断线后自动重连监听器
-        System.out.println("try connect!");
+        log.info("try connect!");
         for (Map.Entry<String, Channel> entry : channelsMap.entrySet()) {
             if (entry.getValue() == channel && !channel.isActive()) {
                 channelGroup = groupBootstrap.connect(groupNettyClient.host, groupNettyClient.port).addListener(new ChannelFutureListener() {
                     @Override
                     public void operationComplete(ChannelFuture future) throws Exception {
                         if (future.cause() != null) {
-                            System.out.println("Failed to connect: " + future.cause());
+                            log.info("Failed to connect: " + future.cause());
                         }
                     }
                 }).channel();
@@ -126,11 +128,11 @@ public class GroupNettyClient {
 /*    public void sendMessage(Object msg,String type) {
         for (Map.Entry<String, Channel> ch:channelsMap.entrySet() ) {
             if(type.equals(ch.getKey())) {
-                System.out.println(msg);
-                System.out.println(ch.getKey());
+                log.info(msg);
+                log.info(ch.getKey());
                 ch.getValue().writeAndFlush(msg);
             }else {
-                System.out.println("----to group --");
+                log.info("----to group --");
                 ch.getValue().writeAndFlush(msg);
             }
         }

+ 6 - 5
src/main/java/com/persagy/client/GroupNettyClientHandler.java

@@ -55,7 +55,7 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
         }
         IdleStateEvent e = (IdleStateEvent) evt;
         if (e.state() == IdleState.READER_IDLE) {
-            System.out.println("no inbound traffic");
+            log.info("no inbound traffic");
             // The connection was OK but there was no traffic for last period.
             // 长时间不操作的时候自动关闭连接; ctx.close();
         }
@@ -69,7 +69,8 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
      */
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
-        System.out.println("Connected to: " + ctx.channel().remoteAddress());
+        log.info("Connected to: " + ctx.channel().remoteAddress());
+        ctx.channel().writeAndFlush(new NettyMessage<>(200,groupNettyClient.projectId).toString());
         //启动的时候发送消息,获取全部报警定义
         //{"groupCode":"wd", "projectId":"Pj123"}
         NettyMessage nettyMessage = new NettyMessage();
@@ -184,17 +185,17 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         //客户端自己不正常情况下自己在重连一次
-        System.out.println("Disconnected from: " + ctx.channel().remoteAddress());
+        log.info("Disconnected from: " + ctx.channel().remoteAddress());
 
     }
 
     @Override
     public void channelUnregistered(final ChannelHandlerContext ctx) throws Exception {
-        System.out.println("Sleeping for: " + RECONNECT_DELAY + "s,Reconnecting to: " + groupNettyClient.host + ':' + groupNettyClient.port);
+        log.info("Sleeping for: " + RECONNECT_DELAY + "s,Reconnecting to: " + groupNettyClient.host + ':' + groupNettyClient.port);
         ctx.channel().eventLoop().schedule(new Runnable() {
             @Override
             public void run() {
-                System.out.println("Reconnecting to: " + groupNettyClient.host + ':' + groupNettyClient.port);
+                log.info("Reconnecting to: " + groupNettyClient.host + ':' + groupNettyClient.port);
                 groupNettyClient.connect(groupNettyClient, ctx.channel());
             }
         }, RECONNECT_DELAY, TimeUnit.SECONDS);

+ 11 - 2
src/main/java/com/persagy/entity/NettyMessage.java

@@ -2,14 +2,14 @@ package com.persagy.entity;
 
 import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.annotation.JSONField;
+import lombok.Data;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 
 import java.util.List;
 
-@Getter
-@Setter
+@Data
 @NoArgsConstructor
 public class NettyMessage<T> {
     /*
@@ -62,6 +62,15 @@ public class NettyMessage<T> {
         this.content = content;
     }
 
+    public NettyMessage(int opCode, String source) {
+        this.opCode = opCode;
+        this.source = source;
+    }
+
+    public NettyMessage(String source) {
+        this.source = source;
+    }
+
     @Override
     public String toString() {
         return JSONObject.toJSONString(this);

+ 23 - 6
src/main/java/com/persagy/job/SpringSchedule.java

@@ -2,11 +2,14 @@ package com.persagy.job;
 
 import com.persagy.client.GroupNettyClient;
 import com.persagy.entity.NettyMessage;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
+import java.util.Objects;
+
 /**
  * @author LuoGuangyi
  * @title: SpringSchedule
@@ -16,15 +19,29 @@ import org.springframework.stereotype.Service;
  */
 
 @Service
+@Slf4j
 public class SpringSchedule {
     @Autowired
     GroupNettyClient groupNettyClient;
 
-    //@Scheduled(cron="${alarm.get.all.alarmdefine.cron}")
-    @Scheduled(initialDelay=1000, fixedRate=60000)
-    public void demoCron() {
-//        NettyMessage message = new NettyMessage();
-//        message.setOpCode(4);
-//        groupNettyClient.sendMessage(message.toString());
+    @Scheduled(cron="${alarm.get.all.alarmdefine.cron}")
+    public void allResetCron() {
+        NettyMessage message = new NettyMessage();
+        message.setOpCode(4);
+        groupNettyClient.sendMessage(message.toString());
+    }
+
+
+    @Scheduled(initialDelay=1000, fixedRate=600000)
+    public void connectAnalizeCron() {
+        if(Objects.isNull(GroupNettyClient.channelGroup)) {
+            log.warn("NettyClient is not init");
+            return;
+        }
+        log.info("NettyClient State:isActive[{}],isOpen[{}],isRegistered[{}],isWritable[{}]",GroupNettyClient.channelGroup.isActive(),GroupNettyClient.channelGroup.isOpen(),GroupNettyClient.channelGroup.isRegistered(),GroupNettyClient.channelGroup.isWritable());
+
     }
+
+
+
 }

+ 18 - 3
src/main/resources/application.yml

@@ -39,7 +39,7 @@ spring:
 group:
   code: WD   #标识哪个接团 比如万达使用WD
   alarm:
-    host: 39.102.43.179 #39.102.43.179     #netty IP
+    host: 39.102.43.179     #netty IP
     port: 9986          #netty 端口9986
 terminal:  #边缘端IOT采集程序地址
   alarm:  # 拼接后的地址为ws://host:port/suffix
@@ -69,7 +69,7 @@ autolog:
   time-consume-threshold: -1
   # 是否打印日志到本地
   local:
-    enable: true
+    enable: false
     # 本地打印耗时阈值,只有当方法耗时大于这个值时打印到本地,单位毫秒,默认不限制
     time-consume-threshold: -1
   # 选择开启哪些切面
@@ -86,4 +86,19 @@ autolog:
     args-full: true
     args-max-length: 512
     result-full: false
-    result-max-length: 512
+    result-max-length: 512
+  kafka:
+    # 若开启 kafka 上报,则以下四个选项为必填项
+    enable: false
+    bootstrap-server:
+    client-id:
+    topic:
+    # 以下 kafka 细节配置,如无特殊情况,请保持默认
+    acks: all
+    batch-size: 16384
+    buffer-memory: 33554432
+    max-block-ms: 0
+    metadata-max-age-ms: 300000
+    request-timeout-ms: 30000
+    key-serializer: org.apache.kafka.common.serialization.StringSerializer
+    value-serializer: org.apache.kafka.common.serialization.StringSerializer