Browse Source

更新招商配置

luoguangyi 3 years ago
parent
commit
3d675be84b

+ 2 - 2
src/main/java/com/persagy/iottransfer/kafka/KafkaConsumerCloud2Edge.java

@@ -25,7 +25,7 @@ import java.util.Optional;
 @ConditionalOnProperty(prefix = "iot.type", name = "edge", havingValue = "true", matchIfMissing = false)
 public class KafkaConsumerCloud2Edge {
 
-    @KafkaListener(topics = KafkaProducer.TOPIC_COLLECT2EDGE, groupId = KafkaProducer.TOPIC_EDGE)
+    @KafkaListener(topics = KafkaProducer.TOPIC_COLLECT2EDGE, groupId = KafkaProducer.GROUP_IOT)
     public void topicCollect2Edge(ConsumerRecord<String, String> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
         Optional<String> message = Optional.ofNullable(record.value());
         if (message.isPresent()) {
@@ -38,7 +38,7 @@ public class KafkaConsumerCloud2Edge {
         }
     }
 
-    @KafkaListener(topics = KafkaProducer.TOPIC_CONTROL2EDGE, groupId = KafkaProducer.TOPIC_EDGE)
+    @KafkaListener(topics = KafkaProducer.TOPIC_CONTROL2EDGE, groupId = KafkaProducer.GROUP_IOT)
     public void topicControl2Edge(ConsumerRecord<String, String> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
         Optional<String> message = Optional.ofNullable(record.value());
         if (message.isPresent()) {

+ 2 - 2
src/main/java/com/persagy/iottransfer/kafka/KafkaConsumerEdge2Cloud.java

@@ -26,7 +26,7 @@ import java.util.Optional;
 public class KafkaConsumerEdge2Cloud {
 
 
-    @KafkaListener(topics = KafkaProducer.TOPIC_EDGE2COLLECT, groupId = KafkaProducer.TOPIC_CLOUD)
+    @KafkaListener(topics = KafkaProducer.TOPIC_EDGE2COLLECT, groupId = KafkaProducer.GROUP_IOT)
     public void topicEdge2Collect(ConsumerRecord<String, String> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
         Optional<String> message = Optional.ofNullable(record.value());
         try {
@@ -43,7 +43,7 @@ public class KafkaConsumerEdge2Cloud {
         ack.acknowledge();
     }
 
-    @KafkaListener(topics = KafkaProducer.TOPIC_EDGE2CONTROL, groupId = KafkaProducer.TOPIC_CLOUD)
+    @KafkaListener(topics = KafkaProducer.TOPIC_EDGE2CONTROL, groupId = KafkaProducer.GROUP_IOT)
     public void topicEdge2Control(ConsumerRecord<String, String> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
         Optional<String> message = Optional.ofNullable(record.value());
         try {

+ 7 - 7
src/main/java/com/persagy/iottransfer/kafka/KafkaProducer.java

@@ -21,13 +21,13 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
 public class KafkaProducer {
 
     //自定义topic
-    public static final String TOPIC_EDGE2COLLECT = "iot.edge2collect";
-    public static final String TOPIC_EDGE2CONTROL = "iot.edge2control";
-    public static final String TOPIC_COLLECT2EDGE = "iot.collect2edge";
-    public static final String TOPIC_CONTROL2EDGE = "iot.control2edge";
+    public static final String TOPIC_EDGE2COLLECT = "persagy.iot.edge2collect";
+    public static final String TOPIC_EDGE2CONTROL = "persagy.iot.edge2control";
+    public static final String TOPIC_COLLECT2EDGE = "persagy.iot.collect2edge";
+    public static final String TOPIC_CONTROL2EDGE = "persagy.iot.control2edge";
     //group
-    public static final String TOPIC_EDGE = "topic.edge";
-    public static final String TOPIC_CLOUD = "topic.cloud";
+    public static final String GROUP_IOT = "persagy.group.iot";
+
     @Autowired
     private KafkaTemplate<String, Object> kafkaTemplate;
 
@@ -35,7 +35,7 @@ public class KafkaProducer {
         String obj2String = JSONObject.toJSONString(obj);
         log.info("准备发送消息为:{}", obj2String);
         if (StrUtil.isEmpty(obj.getProjectId())) {
-            //log.info("项目id为空,丢去:{}", obj2String);
+            log.info("项目id为空,丢去:{}", obj2String);
             return;
         }
         //发送消息

+ 19 - 15
src/main/resources/application.yml

@@ -8,15 +8,17 @@ spring:
   kafka:
     enable: true
     #kafka-servers,需要更改
-    bootstrap-servers: 192.168.64.16:9092 #120.24.88.152:9093,120.24.91.199:9093,120.24.81.140:9093 #192.168.64.16:9092
+    bootstrap-servers: 113.108.52.72:4439,113.108.52.72:4440 #kafka.docker.ssl:9094  #192.168.64.16:9092
     #ssl认证,根据情况配置,store-location相关的认证文件对应添加,需要更改
     ssl:
-      key-password: test
-      key-store-location: /usr/persagy/iot-transfer/config/server.keystore.jks
-      key-store-password: test
+      protocol: SSL
       key-store-type: JKS
-      trust-store-location: /usr/persagy/iot-transfer/config/client.truststore.jks
-      trust-store-password: test
+      key-password: cmsk-ioc-1234 #kafkadockerclient
+      key-store-location: file://D:\temp\ioc-ssl\client.keystore.jks # file://e:\certs\docker.kafka.client.keystore.jks
+      key-store-password: cmsk-ioc-1234 #kafkadockerclient
+      trust-store-location: file://D:\temp\ioc-ssl\client.truststore.jks #file://e:\certs\docker.kafka.client.truststore.jks
+      trust-store-password: cmsk-ioc-1234 #kafkadockerclient
+      trust-store-type: JKS
     producer:
       # 发生错误后,消息重发的次数。
       retries: 2
@@ -32,9 +34,9 @@ spring:
       # acks=1 : 只要集群的首领节点收到消息,生产者就会收到一个来自服务器成功响应。
       # acks=all :只有当所有参与复制的节点全部收到消息时,生产者才会收到一个来自服务器的成功响应。
       acks: 1
-#      properties:
-#        max.request.size: 50000000
-      #   生产着启用SSL,需要启用的环境放开,不需要启用的环境注释掉下面一行 ,需要更改
+      properties:
+        max.request.size: 50000000
+      # 生产着启用SSL,需要启用的环境放开,不需要启用的环境注释掉下面一行 ,需要更改
       security:
         protocol: SSL
     consumer:
@@ -53,8 +55,8 @@ spring:
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       # 值的反序列化方式
       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
-#      properties:
-#        max.partition.fetch.bytes: 50000000
+      properties:
+        max.partition.fetch.bytes: 50000000
       #   消费者启用SSL,需要启用的环境放开,不需要启用的环境注释掉下面一行 ,需要更改
       security:
         protocol: SSL
@@ -64,20 +66,22 @@ spring:
       #listner负责ack,每调用一次,就立即commit
       ack-mode: manual_immediate
       missing-topics-fatal: false
+    properties:
+      ssl.endpoint.identification.algorithm:
 iot:
   # 程序一式两份,一份部署在云端,一份部署在边缘端,一式两份之间通过kafka通讯,实现内外网之间的iot数据通讯
   type:
     # 程序部署位置类型,部署在云端一侧的,需要设置cloud为true,edge为false,需要更改
-    cloud: false
+    cloud: true
     # 程序部署位置类型,部署在边缘端一侧的,需要设置cloud为false,edge为true,需要更改
-    edge: false
+    edge: true
   server:
     #同原来iot-collect中的util.xml配置,用于部署在需要更改edge边缘端给iot-project通讯使用,需要更改
     collect:
       tcp_enable: true
       tcp_bind_ip: 0.0.0.0
       #需要更改
-      tcp_bind_port: 8851
+      tcp_bind_port: 8861
       udp_enable: false
       udp_bind_ip: 0.0.0.0
       udp_bind_port: 8851
@@ -105,7 +109,7 @@ iot:
       #iot-collect IP,需要更改
       target_ip: 127.0.0.1
       #iot-collect port,需要更改
-      target_port: 8851
+      target_port: 8855
       max_size: 1000
       sleepcount: 1000
       separate_begin: (