Browse Source

no message

wwd 3 years ago
parent
commit
25e98b2b85

+ 3 - 3
src/main/java/com/persagy/ztkedgeclouddatasecurity/kafka/EdgeKafkaProducer.java

@@ -147,7 +147,7 @@ public class EdgeKafkaProducer {
 //                    executorService.shutdown();
                     JSONObject object = JSONObject.parseObject(messageObject.toString());
                     String obj2String = JSONObject.toJSONString(object);
-                    logger.info("准备发送消息为:{}", obj2String);
+                    logger.info(location+">>>kafka准备发送消息为:{}", obj2String);
                     String TOPIC = location+"_"+ object.getString("port") + "_" + object.getString("userid");
                     //发送消息
                     ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC, encryptInputMessageService.beforeBodyWrite(obj2String));
@@ -169,12 +169,12 @@ public class EdgeKafkaProducer {
                 }
 
             } catch (InterruptedException e) {
-                logger.error("executorService_error>>>" + e);
+                logger.error(location+">>>executorService_error>>>" + e);
                // ProducerConfigbyself();
             } catch (Exception e) {
                 e.printStackTrace();
                // ProducerConfigbyself();
-                logger.error("executorService_error>>>" + e);
+                logger.error(location+">>>executorService_error>>>" + e);
             }
             // messageQueue.poll();
 

+ 13 - 10
src/main/java/com/persagy/ztkedgeclouddatasecurity/netty/MsgHandler.java

@@ -52,12 +52,15 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
 
     private  Map<String, Set<Channel>> socketChannelMap;
 
+    private  String location;
 
-    public MsgHandler(ChattingUser chattingUser, EdgeKafkaProducer edgeKafkaProducer, Map<String, Set<Channel>> socketChannelMap) {
+
+    public MsgHandler(ChattingUser chattingUser, EdgeKafkaProducer edgeKafkaProducer, Map<String, Set<Channel>> socketChannelMap,String location) {
         this.chattingUser =chattingUser;
         this.edgeKafkaProducer=edgeKafkaProducer;
         this.edgeKafkaProducer.producerMsg(messageQueue);
         this.socketChannelMap= socketChannelMap;
+        this.location =location;
 
     }
 
@@ -76,7 +79,7 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, NettyMessage<JSONObject> msg) throws Exception {
-        log.info("收到[{}]消息:{}", ctx.channel().remoteAddress(), msg);
+        log.info(location+">>>收到[{}]消息:{}", ctx.channel().remoteAddress(), msg);
         //super.channelRead(ctx, msg);
         // nConnection.decrementAndGet();
 
@@ -104,7 +107,7 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
         //    if (Integer.valueOf(localport) == list.get(j).getPort() ){
         obj.put("userid",chattingUser.getUserId());
         obj.put("targetAddress",chattingUser.getIp());
-        obj.put("channelID",chattingUser.getUserId()+"_"+localport);
+        obj.put("channelID",localport+"_"+chattingUser.getUserId());
 
         //   }
         // }
@@ -188,7 +191,7 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
         SocketAddress socketAddress = ctx.channel().remoteAddress();
         String remoteAddress = socketAddress.toString();
-        log.warn("--某个客户端绑定地址:[{}]--", remoteAddress);
+        log.warn(location+">>>--某个客户端绑定地址:[{}]--", remoteAddress);
     }
 
 
@@ -201,7 +204,7 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
      */
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        log.error("netty 连接异常" + ctx.channel().remoteAddress(), cause);
+        log.error(location+">>>netty 连接异常" + ctx.channel().remoteAddress(), cause);
         ctx.close();
     }
 
@@ -212,10 +215,10 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
         //NettyServer.socketChannelMap.put(ctx.channel().remoteAddress().toString(), ctx.channel());
         channelGroup.add(ctx.channel());
-        log.info("当前连接数:[{}],新建立连接为[{}]...", channelGroup.size(), ctx.channel().remoteAddress().toString());
+        log.info(location+">>>当前连接数:[{}],新建立连接为[{}]...", channelGroup.size(), ctx.channel().remoteAddress().toString());
         if (CollectionUtil.isNotEmpty(channelGroup)) {
             for (Channel channel : channelGroup) {
-                log.info("已有连接地址为[{}]", channel.remoteAddress().toString());
+                log.info(location+">>>已有连接地址为[{}]", channel.remoteAddress().toString());
             }
         }
         super.handlerAdded(ctx);
@@ -231,7 +234,7 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
         //下面这行代码Netty会自动调用
         channelGroup.remove(ctx.channel());
         super.handlerRemoved(ctx);
-        log.warn("----客户端[{}] ----离开", ctx.channel().remoteAddress().toString());
+        log.warn(location+">>>----客户端[{}] ----离开", ctx.channel().remoteAddress().toString());
     }
 
     /**
@@ -239,7 +242,7 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
      */
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
-        log.warn("[{}]:channelActive", ctx.channel().remoteAddress());
+        log.warn(location+">>>[{}]:channelActive", ctx.channel().remoteAddress());
         super.channelActive(ctx);
     }
 
@@ -249,7 +252,7 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         ctx.channel().close();
-        log.warn("[{}]:channelInactive", ctx.channel().remoteAddress());
+        log.warn(location+">>>[{}]:channelInactive", ctx.channel().remoteAddress());
         super.channelInactive(ctx);
     }
 }

+ 4 - 2
src/main/java/com/persagy/ztkedgeclouddatasecurity/netty/MyChannelInitializer.java

@@ -15,11 +15,13 @@ public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
     private final EdgeKafkaProducer edgeKafkaProducer;
     private ChattingUser chattingUser;
     private Map<String, Set<Channel>> socketChannelMap;
+    private String location;
 
-    public MyChannelInitializer(ChattingUser chattingUser, EdgeKafkaProducer edgeKafkaProducer, Map<String, Set<Channel>> socketChannelMap) {
+    public MyChannelInitializer(ChattingUser chattingUser, EdgeKafkaProducer edgeKafkaProducer, Map<String, Set<Channel>> socketChannelMap,String location) {
         this.chattingUser = chattingUser;
         this.edgeKafkaProducer = edgeKafkaProducer;
         this.socketChannelMap = socketChannelMap;
+        this.location = location;
     }
 
     @Override
@@ -28,7 +30,7 @@ public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
         //对象传输处理[解码]
         channel.pipeline().addLast(new ObjDecoder());
         // 在管道中添加我们自己的接收数据实现方法
-        channel.pipeline().addLast(new MsgHandler(chattingUser,edgeKafkaProducer,socketChannelMap));
+        channel.pipeline().addLast(new MsgHandler(chattingUser,edgeKafkaProducer,socketChannelMap,location));
         //对象传输处理[编码]
         channel.pipeline().addLast(new ObjEncoder());
     }

+ 6 - 6
src/main/java/com/persagy/ztkedgeclouddatasecurity/netty/cloud/CenterClientHandler.java

@@ -55,7 +55,7 @@ public class CenterClientHandler extends SimpleChannelInboundHandler<NettyMessag
      */
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
-        log.info("Connected to: {}", ctx.channel().remoteAddress());
+        log.info(location+">>>Connected to: {}", ctx.channel().remoteAddress());
         //建立连接
         nettyClient.sendMessage(new NettyMessage("",200,"Pj4403070003","ZS"));
         //启动的时候发送消息,获取全部报警定义
@@ -77,7 +77,7 @@ public class CenterClientHandler extends SimpleChannelInboundHandler<NettyMessag
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, NettyMessage msg) throws Exception {
         //接收msg消息{与上一章节相比,此处已经不需要自己进行解码}
-        log.info("Client received: {}", msg);
+        log.info(location+">>>Client received: {}", msg);
         System.out.println(">>>>>"+projectID);
         JSONObject cloudboj= new JSONObject();
 
@@ -96,7 +96,7 @@ public class CenterClientHandler extends SimpleChannelInboundHandler<NettyMessag
            // handlerMsg(ctx, msg);
            // log.info("处理消息时间[{}]",timer.interval()  +">>>>>>"+ ctx.channel());
         } catch (Exception e) {
-            log.error("channelRead", e);
+            log.error(location+">>>channelRead", e);
         }
     }
 
@@ -114,7 +114,7 @@ public class CenterClientHandler extends SimpleChannelInboundHandler<NettyMessag
      */
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        log.info("exceptionCaught", cause);
+        log.info(location+">>>exceptionCaught", cause);
         ctx.close();
     }
 
@@ -125,14 +125,14 @@ public class CenterClientHandler extends SimpleChannelInboundHandler<NettyMessag
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         //客户端自己不正常情况下自己在重连一次
-        log.info("Disconnected from: " + ctx.channel().remoteAddress());
+        log.info(location+">>>Disconnected from: " + ctx.channel().remoteAddress());
 
     }
 
 
     @Override
     public void channelUnregistered(final ChannelHandlerContext ctx) throws Exception {
-        log.info("channelUnregistered and reconnecting to: {}:{} ", RECONNECT_DELAY, CommonConst.inetHost, CommonConst.inetPort);
+        log.info(location+">>>channelUnregistered and reconnecting to: {}:{} ", RECONNECT_DELAY, CommonConst.inetHost, CommonConst.inetPort);
         ctx.channel().eventLoop().schedule(new Runnable() {
             @Override
             public void run() {

+ 7 - 4
src/main/java/com/persagy/ztkedgeclouddatasecurity/netty/cloud/NettyClient.java

@@ -34,6 +34,9 @@ public class NettyClient {
     @Value("${listener.cloud1}")
     private String Connectinfo;
 
+    @Value("${spring.location}")
+    private String location;
+
     @PostConstruct
     public void startNetty(){
         String[] connectInfoArr=Connectinfo.split(",");
@@ -60,7 +63,7 @@ public class NettyClient {
             channel = f.channel();
             // f.channel().closeFuture().sync();
         } catch (Exception e) {
-            log.info("连接异常", e);
+            log.info(location+">>>连接异常", e);
         } finally {
             //workerGroup.shutdownGracefully();
         }
@@ -92,19 +95,19 @@ public class NettyClient {
 
 
     public void sendMessage(NettyMessage msg) throws InterruptedException {
-        log.info("给云端发送数据:[{}]", msg);
+        log.info(location+">>>给云端发送数据:[{}]", msg);
         if (channel.isWritable()) {
             try {
                 //!!!!!!注意,NettyMessage<T>是有泛型的,如果把发送一个数据,接收方必须有一个和<T>的全限定类名完全一样的实体类,
                 //发送方不应该限制接收方,所有使用JSONObject.parseObject(msg.toString(), NettyMessage.class)操作让泛型擦除掉
                 channel.writeAndFlush(JSONObject.parseObject(msg.toString(), new TypeReference<NettyMessage<JSONObject>>(){}));
             } catch (Exception e) {
-                log.error("发送数据异常,放入缓冲队列中", e);
+                log.error(location+">>>发送数据异常,放入缓冲队列中", e);
                 NettyMessageQueue.getNettyMessageQueue().cloudproduce(msg);
                 channel.close();
             }
         } else {
-            log.warn("云端netty不可写,放入缓冲队列中[{}]", msg);
+            log.warn(location+">>>云端netty不可写,放入缓冲队列中[{}]", msg);
             NettyMessageQueue.getNettyMessageQueue().cloudproduce(msg);
         }
     }

+ 7 - 7
src/main/java/com/persagy/ztkedgeclouddatasecurity/netty/edge/NettyServer.java

@@ -29,7 +29,6 @@ public class NettyServer {
 
     public static Map<String, Set<Channel>> socketChannelMap = new ConcurrentHashMap<>();
 
-
     private Logger logger = LoggerFactory.getLogger(this.getClass());
     @Value("${listener.port}")
     private String listenerPort;
@@ -100,7 +99,7 @@ public class NettyServer {
                 int port = Integer.valueOf(ipPortArr[i - 1].split(":")[3]);
                 chattingUser.setPort(port);
                 list.add(chattingUser);
-                serverBootstrap.childHandler(new MyChannelInitializer(chattingUser, edgeKafkaProducer,socketChannelMap));
+                serverBootstrap.childHandler(new MyChannelInitializer(chattingUser, edgeKafkaProducer,socketChannelMap,location));
                 ChannelFuture channelFuture = serverBootstrap.bind(port);
                 if (ChannelFutures == null) {
                     ChannelFutures = new ChannelFuture[8];
@@ -121,8 +120,9 @@ public class NettyServer {
     }
 
     public static synchronized void addChannel(String projectId, ChannelHandlerContext ctx) {
+
         //200 建立和客户端的连接
-        log.warn("添加channel[{}],项目id[{}]", ctx.channel().remoteAddress().toString(), projectId);
+        log.warn(">>>添加channel[{}],项目id[{}]", ctx.channel().remoteAddress().toString(), projectId);
         Set<Channel> channels = CollectionUtil.isEmpty(socketChannelMap.get(projectId)) ? new HashSet<>() : socketChannelMap.get(projectId);
         channels.add(ctx.channel());
         socketChannelMap.put(projectId, channels);
@@ -135,17 +135,17 @@ public class NettyServer {
                 for (Channel channel : channels) {
                     if (channel.isActive() && channel.isWritable()) {
                         channel.writeAndFlush(JSONObject.parseObject(msg.toString(), NettyMessage.class));
-                        log.info("netty 向边缘端[{}][{}]发送消息:{}", channel.remoteAddress().toString(), projectId, msg);
+                        log.info(">>>netty 向边缘端[{}][{}]发送消息:{}", channel.remoteAddress().toString(), projectId, msg);
                         return true;
                     } else {
-                        log.error("netty not writable now, message dropped");
+                        log.error(">>>netty not writable now, message dropped");
                     }
                 }
             } else {
-                log.error("projectId[{}]channels获取失败", projectId);
+                log.error(">>>projectId[{}]channels获取失败", projectId);
             }
         } else {
-            log.info("...projectId[{}]未建立连接,消息不发送!", projectId);
+            log.info(">>>...projectId[{}]未建立连接,消息不发送!", projectId);
         }
         return false;
     }

+ 4 - 1
src/main/resources/application-dev.yml

@@ -1,5 +1,5 @@
 server:
-  port: 8090
+  port: 8091
 listener:
   cloud1: 1:Pj4403050019:192.168.0.26:9986
   port:  1:Pj4403050019:192.168.0.35:30091
@@ -30,6 +30,9 @@ spring:
       batch-size: 16384
       buffer-memory: 33554432
       client-id: _clent_
+      compression:
+        type: gzip
+
       # 键的序列化方式
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       # 值的序列化方式

+ 1 - 1
src/main/resources/application.yml

@@ -1,6 +1,6 @@
 spring:
   profiles:
     active: dev
-  location: Edge  # Edge 边缘测 or Cloud 云端
+  location: Cloud  # Edge 边缘测 or Cloud 云端