fix(message):调整kafka消息提交错误

amos-boot-utils/amos-boot-utils-message/src/main/java/com/yeejoin/amos/message/kafka/KafkaConsumerService.java

@@ -69,12 +69,11 @@ public class KafkaConsumerService {
                     data = messageObj;
                     data.put("datatype", "state");
                 }
-                //log.info("接收到Roma消息对象: {}", data);
                 emqKeeper.getMqttClient().publish(MQTT_TOPIC, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
             } catch (MqttException e) {
                 log.error("消息转发失败" + e.getMessage(), e);
-                ack.acknowledge();
             }
+            ack.acknowledge();
         }
     }
 
@@ -84,7 +83,7 @@ public class KafkaConsumerService {
      * @param message 省级消息
      * @param ack     ack
      */
-    @KafkaListener(id = "provinceMessage", groupId = "province", topics = "#{'${kafka.risk.topics}'.split(',')}", concurrency = "2")
+    @KafkaListener(id = "provinceMessage", groupId = "province", topics = "#{'${kafka.topics}'.split(',')}", concurrency = "2")
     public void consumerSingle1(String message, Acknowledgment ack) {
         if(isZxj) {
             Optional<?> messages = Optional.ofNullable(message);
@@ -105,8 +104,8 @@ public class KafkaConsumerService {
                     }
                 } catch (MqttException e) {
                     log.error("消息转发失败" + e.getMessage(), e);
-                    ack.acknowledge();
                 }
+                ack.acknowledge();
             }
         }
     }