Browse Source

Merge branch 'master' of http://39.106.8.246:3003/persagy/ztk-datasecurity

wangwendongVMMacOS 3 years ago
parent
commit
0ebb7c9a2c

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

@@ -43,7 +43,7 @@ public class CloudKafkaProducer {
 
         String TOPIC = location+"_"+ msgObj.getString("projectid");
         //发送消息
-        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC, encryptInputMessageService.beforeBodyWrite(msgObj));
+        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC, encryptInputMessageService.beforeBodyWrite(msgObj.toString()));
         future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
             @Override
             public void onFailure(Throwable throwable) {

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

@@ -1,5 +1,6 @@
 package com.persagy.ztkedgeclouddatasecurity.netty.cloud;
 
+import cn.hutool.core.thread.ThreadUtil;
 import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.TypeReference;
 import com.persagy.ztkedgeclouddatasecurity.entity.NettyMessage;
@@ -13,7 +14,6 @@ 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;
@@ -34,17 +34,26 @@ public class NettyClient {
     private String Connectinfo;
 
     @PostConstruct
-    public void connect() {
-        EventLoopGroup workerGroup = new NioEventLoopGroup();
+    public void startNetty(){
         String[] connectInfoArr=Connectinfo.split(",");
         for (int i=0;i<connectInfoArr.length;i++){
+            int finalI = i;
+            ThreadUtil.execAsync(()->{connect(finalI,connectInfoArr);}, true);
+        }
+
+    }
+
+
+    public Runnable connect(int i, String[] connectInfoArr) {
+        EventLoopGroup workerGroup = new NioEventLoopGroup();
+        bootstrap.group(workerGroup);
+        bootstrap.channel(NioSocketChannel.class);
+        bootstrap.option(ChannelOption.AUTO_READ, true);
+
             String ip= connectInfoArr[i].split(":")[2];
             int port = Integer.valueOf(connectInfoArr[i].split(":")[3]);
             String projectId = connectInfoArr[i].split(":")[3]+"_"+connectInfoArr[i].split(":")[1];
         try {
-            bootstrap.group(workerGroup);
-            bootstrap.channel(NioSocketChannel.class);
-            bootstrap.option(ChannelOption.AUTO_READ, true);
             bootstrap.handler(new CenterChannelInitializer(cloudKafkaConsumer,this,projectId,cloudKafkaProducer));
             ChannelFuture f = bootstrap.connect(ip, port).sync();
             channel = f.channel();
@@ -54,7 +63,8 @@ public class NettyClient {
         } finally {
             //workerGroup.shutdownGracefully();
         }
-        }
+
+        return null;
     }