巡检bug

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

@@ -28,17 +28,19 @@ public class KafkaConsumerService {
      * 消费单条消息,topics 可以监听多个topic，如：topics = {"topic1", "topic2"}
      * @param message 消息
      */
-    @KafkaListener(id = "consumerSingle", topics = "#{'${kafka.topics}'.split(',')}", concurrency = "2")
+    @KafkaListener(id = "consumerSingle", idIsGroup = false, topics = "#{'${kafka.topics}'.split(',')}", concurrency = "2")
     public void consumerSingle(String message,Acknowledgment ack) {
         JSONObject messageObj = JSONObject.fromObject(message);
-        String topic = messageObj.getString("topic");
-        JSONObject data = messageObj.getJSONObject("data");
         try {
+            String topic = messageObj.getString("topic");
+            JSONObject data = messageObj.getJSONObject("data");
             emqKeeper.getMqttClient().publish(topic, data.toString().getBytes("UTF-8"), 0,false);
         } catch (MqttException e) {
             e.printStackTrace();
         } catch (UnsupportedEncodingException e) {
             e.printStackTrace();
+        } catch (Exception e) {
+            log.info("单条消息 ====> message: {}", message);
         }
         log.info("单条消息 ====> message: {}", message);
         ack.acknowledge();