Browse Source

消费者组动态配置

luoguangyi 3 years ago
parent
commit
2d8a08f41d

+ 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.GROUP_IOT)
+    @KafkaListener(topics = KafkaProducer.TOPIC_COLLECT2EDGE, groupId = "${persagy.group.iot: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.GROUP_IOT)
+    @KafkaListener(topics = KafkaProducer.TOPIC_CONTROL2EDGE, groupId = "${persagy.group.iot: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.GROUP_IOT)
+    @KafkaListener(topics = KafkaProducer.TOPIC_EDGE2COLLECT, groupId = "${persagy.group.iot: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.GROUP_IOT)
+    @KafkaListener(topics = KafkaProducer.TOPIC_EDGE2CONTROL, groupId = "${persagy.group.iot: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 {

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

@@ -26,7 +26,7 @@ public class KafkaProducer {
     public static final String TOPIC_COLLECT2EDGE = "persagy.iot.collect2edge";
     public static final String TOPIC_CONTROL2EDGE = "persagy.iot.control2edge";
     //group
-    public static final String GROUP_IOT = "persagy.group.iot";
+    public static final String GROUP_IOT = "group_iot";
 
     @Autowired
     private KafkaTemplate<String, Object> kafkaTemplate;