ebike-gather/src/main/java/org/cdzy/gather/kafka/KafkaConsumer.java

@@ -64,6 +64,7 @@ public class KafkaConsumer {
             }
             mqttPoolClient.getPoolStatus();
             // 使用注入的 MqttPoolClient 发送消息
+            Thread.sleep(10000);
             boolean sent = mqttPoolClient.sendMessage(topic, MqttQoS.AT_LEAST_ONCE, command);
 
             log.info("[MQTT发送结果] 主题: {}, 成功: {}", topic, sent);

ebike-gather/src/main/java/org/cdzy/gather/mqtt/MqttHandler.java

@@ -6,8 +6,6 @@ import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.mqtt.*;
 import io.netty.handler.timeout.IdleStateEvent;
 import lombok.extern.slf4j.Slf4j;
-import org.cdzy.gather.config.SpringContextHolder;
-import org.cdzy.gather.kafka.KafkaProducer;
 
 import java.nio.charset.StandardCharsets;
 
@@ -106,12 +104,6 @@ public class MqttHandler extends ChannelInboundHandlerAdapter {
      */
     private void processReceivedMessage(String topic, String content) {
         try {
-            KafkaProducer kafkaProducer = SpringContextHolder.getBean(KafkaProducer.class);
-           if (topic.contains("rsp")) {
-               kafkaProducer.send("msg_rsp",content);
-           } else if (topic.contains("rpt")) {
-               kafkaProducer.send("msg_rpt",content);
-           }
             log.info("处理MQTT消息 - 主题: {}, 内容: {}", topic, content);
             if (topic.startsWith("sensor/")) {
                 handleSensorData(topic, content);