wwd 3 years ago
parent
commit
b35076bad1

+ 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) {

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

@@ -24,7 +24,7 @@ public class EdgeKakfaConsumer {
     public NettyClient nettyClient;
 
     @KafkaListener(topicPattern = "Cloud_.*")
-    public void topicList(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
+    public void topicList(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws InterruptedException {
 
 
         //System.out.println("............"+record.value());
@@ -35,13 +35,9 @@ public class EdgeKakfaConsumer {
             String[] TopicArr = topic.split("_");
             if ("zkt-proj-alarm".equals(TopicArr[2])){
                 JSONObject.parseObject(JSONObject.toJSONString(msg)).getObject("msg", NettyMessage.class);
-                try {
-                  //  nettyClient.sendMessage(JSONObject.parseObject(JSONObject.toJSONString(msg)).getObject("msg", NettyMessage.class));
-                    if (ack != null) {
-                        ack.acknowledge();
-                    }
-                } catch (InterruptedException e) {
-                    log.error(e.getMessage(),e);
+                //  nettyClient.sendMessage(JSONObject.parseObject(JSONObject.toJSONString(msg)).getObject("msg", NettyMessage.class));
+                if (ack != null) {
+                    ack.acknowledge();
                 }
 
             }

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

@@ -36,15 +36,18 @@ public class NettyClient {
     @PostConstruct
     public void connect() {
         EventLoopGroup workerGroup = new NioEventLoopGroup();
+        bootstrap.group(workerGroup);
+        bootstrap.channel(NioSocketChannel.class);
+        bootstrap.option(ChannelOption.AUTO_READ, true);
+
+
         String[] connectInfoArr=Connectinfo.split(",");
         for (int i=0;i<connectInfoArr.length;i++){
+
             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();