Browse Source

no message

wwd 3 years ago
parent
commit
d3cfeb0103

+ 2 - 1
src/main/java/com/persagy/ztkedgeclouddatasecurity/kafka/CloudKafkaConsumer.java

@@ -18,13 +18,14 @@ import java.util.Optional;
 
 @Component
 @Slf4j
-@ConditionalOnProperty(prefix = "spring", name="location", havingValue="Cloud")
+
 public class CloudKafkaConsumer {
 
     @Autowired
     public NettyClient nettyClient;
     //@KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP1)
     @KafkaListener(topicPattern = "Edge_.*")
+    @ConditionalOnProperty(prefix = "spring", name="location", havingValue="Cloud")
     public void topicList(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
       Optional message = Optional.ofNullable(JSONObject.parseObject(record.value().toString()));
         if (message.isPresent()) {

+ 14 - 4
src/main/java/com/persagy/ztkedgeclouddatasecurity/kafka/EdgeKakfaConsumer.java

@@ -1,8 +1,10 @@
 package com.persagy.ztkedgeclouddatasecurity.kafka;
 
 import com.alibaba.fastjson.JSONObject;
+import com.persagy.ztkedgeclouddatasecurity.datasafety.DecryptInputMessageService;
 import com.persagy.ztkedgeclouddatasecurity.entity.NettyMessage;
 import com.persagy.ztkedgeclouddatasecurity.netty.cloud.NettyClient;
+import com.persagy.ztkedgeclouddatasecurity.netty.edge.NettyServer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -23,22 +25,30 @@ public class EdgeKakfaConsumer {
     @Autowired
     public NettyClient nettyClient;
 
+    @Autowired
+    public NettyServer nettyServer;
+
+    @Autowired
+    private DecryptInputMessageService decryptInputMessageService;
+
     @KafkaListener(topicPattern = "Cloud_.*")
     public void topicList(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws InterruptedException {
 
-
         //System.out.println("............"+record.value());
         Optional message = Optional.ofNullable(JSONObject.parseObject(record.value().toString()));
         if (message.isPresent()) {
             Object msg = message.get();
             log.info("Cloud 消费了: Topic:" + topic + ",Message:" + msg);
             String[] TopicArr = topic.split("_");
-            if ("zkt-proj-alarm".equals(TopicArr[2])){
-                JSONObject.parseObject(JSONObject.toJSONString(msg)).getObject("msg", NettyMessage.class);
+            String channelID =JSONObject.parseObject((decryptInputMessageService.DecryptMsgInputMessage(msg.toString()))).getString("projectid");
+            NettyMessage mesobj =JSONObject.parseObject((decryptInputMessageService.DecryptMsgInputMessage(msg.toString()))).getObject("msg", NettyMessage.class);
+
+           // if ("zkt-proj-alarm".equals(TopicArr[2])){
+                NettyServer.sendMessage(channelID,mesobj);
                 //  nettyClient.sendMessage(JSONObject.parseObject(JSONObject.toJSONString(msg)).getObject("msg", NettyMessage.class));
                 if (ack != null) {
                     ack.acknowledge();
-                }
+             //   }
 
             }
 

+ 8 - 1
src/main/java/com/persagy/ztkedgeclouddatasecurity/netty/MsgHandler.java

@@ -21,6 +21,8 @@ import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.text.SimpleDateFormat;
 import java.util.Date;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -48,12 +50,14 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
 
     private EdgeKafkaProducer edgeKafkaProducer;
 
+    private  Map<String, Set<Channel>> socketChannelMap;
 
 
-    public MsgHandler(ChattingUser chattingUser, EdgeKafkaProducer edgeKafkaProducer) {
+    public MsgHandler(ChattingUser chattingUser, EdgeKafkaProducer edgeKafkaProducer, Map<String, Set<Channel>> socketChannelMap) {
         this.chattingUser =chattingUser;
         this.edgeKafkaProducer=edgeKafkaProducer;
         this.edgeKafkaProducer.producerMsg(messageQueue);
+        this.socketChannelMap= socketChannelMap;
 
     }
 
@@ -75,6 +79,7 @@ public class MsgHandler extends SimpleChannelInboundHandler<NettyMessage<JSONObj
         log.info("收到[{}]消息:{}", ctx.channel().remoteAddress(), msg);
         //super.channelRead(ctx, msg);
         // nConnection.decrementAndGet();
+
         InetSocketAddress inteSocket = (InetSocketAddress) ctx.channel().localAddress();
         String localip = inteSocket.getAddress().getHostAddress();
         String localport = String.valueOf(inteSocket.getPort());
@@ -99,10 +104,12 @@ 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);
 
         //   }
         // }
         System.out.println(">>>>"+obj.toString());
+        NettyServer.addChannel(localport+"_"+chattingUser.getUserId(), ctx);
         NettyMessageQueue.getNettyMessageQueue().produce(obj.toString());
        // messageQueue.offer(obj.toString(), 1000, TimeUnit.MICROSECONDS);
        // chattingUser.setMessageTcpSendQueue(messageQueue);

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

@@ -3,17 +3,23 @@ package com.persagy.ztkedgeclouddatasecurity.netty;
 
 import com.persagy.ztkedgeclouddatasecurity.entity.ChattingUser;
 import com.persagy.ztkedgeclouddatasecurity.kafka.EdgeKafkaProducer;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 
+import java.util.Map;
+import java.util.Set;
+
 public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
 
     private final EdgeKafkaProducer edgeKafkaProducer;
     private ChattingUser chattingUser;
+    private Map<String, Set<Channel>> socketChannelMap;
 
-    public MyChannelInitializer(ChattingUser chattingUser, EdgeKafkaProducer edgeKafkaProducer) {
+    public MyChannelInitializer(ChattingUser chattingUser, EdgeKafkaProducer edgeKafkaProducer, Map<String, Set<Channel>> socketChannelMap) {
         this.chattingUser = chattingUser;
         this.edgeKafkaProducer = edgeKafkaProducer;
+        this.socketChannelMap = socketChannelMap;
     }
 
     @Override
@@ -22,7 +28,7 @@ public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
         //对象传输处理[解码]
         channel.pipeline().addLast(new ObjDecoder());
         // 在管道中添加我们自己的接收数据实现方法
-        channel.pipeline().addLast(new MsgHandler(chattingUser,edgeKafkaProducer));
+        channel.pipeline().addLast(new MsgHandler(chattingUser,edgeKafkaProducer,socketChannelMap));
         //对象传输处理[编码]
         channel.pipeline().addLast(new ObjEncoder());
     }

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

@@ -80,18 +80,21 @@ public class CenterClientHandler extends SimpleChannelInboundHandler<NettyMessag
         log.info("Client received: {}", msg);
         System.out.println(">>>>>"+projectID);
         JSONObject cloudboj= new JSONObject();
+
         cloudboj.put("projectid",projectID);
         cloudboj.put("msg",msg);
         try {
             if ("Edge".equals(location)){
 
+
+
             }else {
                 cloudKafkaProducer.CloudproducerMsg(cloudboj);
             }
 
-            TimeInterval timer = DateUtil.timer();
-            handlerMsg(ctx, msg);
-            log.info("处理消息时间[{}]",timer.interval()  +">>>>>>"+ ctx.channel());
+           // TimeInterval timer = DateUtil.timer();
+           // handlerMsg(ctx, msg);
+           // log.info("处理消息时间[{}]",timer.interval()  +">>>>>>"+ ctx.channel());
         } catch (Exception e) {
             log.error("channelRead", e);
         }

+ 1 - 0
src/main/java/com/persagy/ztkedgeclouddatasecurity/netty/cloud/NettyClient.java

@@ -14,6 +14,7 @@ import io.netty.channel.socket.nio.NioSocketChannel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;

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

@@ -1,7 +1,9 @@
 package com.persagy.ztkedgeclouddatasecurity.netty.edge;
 
 import cn.hutool.core.collection.CollectionUtil;
+import com.alibaba.fastjson.JSONObject;
 import com.persagy.ztkedgeclouddatasecurity.entity.ChattingUser;
+import com.persagy.ztkedgeclouddatasecurity.entity.NettyMessage;
 import com.persagy.ztkedgeclouddatasecurity.kafka.EdgeKafkaProducer;
 import com.persagy.ztkedgeclouddatasecurity.netty.MyChannelInitializer;
 import io.netty.bootstrap.ServerBootstrap;
@@ -26,6 +28,8 @@ import java.util.concurrent.ConcurrentHashMap;
 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;
@@ -85,8 +89,8 @@ public class NettyServer {
                 chattingUser.setIp(ipPortArr[i - 1].split(":")[2]);
                 ip = ipPortArr[i - 1].split(":")[2];
             }
-
-            if ("zkt-proj-alarm".equals(ipPortArr[i - 1].split(":")[1])) {
+            //zkt-proj-alarm projectid=Pj4403050019
+            if ("Pj4403050019".equals(ipPortArr[i - 1].split(":")[1])) {
                 ServerBootstrap serverBootstrap = new ServerBootstrap();
                 EventLoopGroup bossGroup = new NioEventLoopGroup();
                 NioEventLoopGroup workerGroup = new NioEventLoopGroup();
@@ -96,7 +100,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));
+                serverBootstrap.childHandler(new MyChannelInitializer(chattingUser, edgeKafkaProducer,socketChannelMap));
                 ChannelFuture channelFuture = serverBootstrap.bind(port);
                 if (ChannelFutures == null) {
                     ChannelFutures = new ChannelFuture[8];
@@ -115,4 +119,42 @@ public class NettyServer {
         }
 
     }
+
+    public static synchronized void addChannel(String projectId, ChannelHandlerContext ctx) {
+        //200 建立和客户端的连接
+        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);
+    }
+
+    public static boolean sendMessage(String projectId, NettyMessage msg) {
+        if (socketChannelMap.containsKey(projectId)) {
+            Set<Channel> channels = socketChannelMap.get(projectId);
+            if (CollectionUtil.isNotEmpty(channels)) {
+                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);
+                        return true;
+                    } else {
+                        log.error("netty not writable now, message dropped");
+                    }
+                }
+            } else {
+                log.error("projectId[{}]channels获取失败", projectId);
+            }
+        } else {
+            log.info("...projectId[{}]未建立连接,消息不发送!", projectId);
+        }
+        return false;
+    }
+
+
+
+
+
+
+
+
 }

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

@@ -1,8 +1,8 @@
 server:
   port: 8090
 listener:
-  cloud1: 1:zkt-proj-alarm:192.168.0.26:9986,2:zkt-proj-alarm:192.168.0.26:30092
-  port: 1:iot-collect:192.168.2.128:30009,2:iot-control:192.168.2.128:30010,3:iot-project:192.168.2.128:30017,4:zkt-control:192.168.2.128:30087,5:zkt-alarm:192.168.2.128:30090,6:zkt-proj-alarm:192.168.2.128:30091,7:zkt-proj-alarm:192.168.2.128:30092
+  cloud1: 1:Pj4403050019:192.168.0.26:9986
+  port:  1:Pj4403050019:192.168.0.35:30091
 project:
   configs:
     - name: 9986_zkt-proj-alarm
@@ -120,7 +120,7 @@ alibaba:
   iot:
     proxy: true
     proxyhost: 192.168.64.16
-    proxyport: 3128
+    proxyport: 31280
     accessKey: "LTAI5tQUWCYstLgGDyq3iLAS"
     accessSecret: "C14Xgg4njYOSj7BP7UCuALLL1oU2g7"
     consumerGroupId: "DEFAULT_GROUP"

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

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