From 18eebaa64ebaea9ae6254f966a52c2a7ae7eb8d8 Mon Sep 17 00:00:00 2001
From: bgy <bgy@topsail.com>
Date: Sat, 10 Oct 2026 15:58:25 +0800
Subject: [PATCH] =?UTF-8?q?=E5=9C=A3=E5=9C=B0=E8=93=9D=E8=AE=BE=E5=A4=87?=
 =?UTF-8?q?=E6=97=A5=E5=BF=97=E5=A4=87=E4=BB=BD=E9=98=9F=E5=88=97=E6=B6=88?=
 =?UTF-8?q?=E8=B4=B9=E5=8A=9F=E8=83=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../topsail/influxdb/rabbitmq/AmqpListener.java    | 48 ++++++++++++++++++++++
 1 file changed, 48 insertions(+)

diff --git a/src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java b/src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java
index d870272..58f6bb8 100644
--- a/src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java
+++ b/src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java
@@ -203,7 +203,55 @@ public class AmqpListener {
             }
         }
     }
+    @RabbitListener(queues = "shengdilandevicelogback")
+    public void deviceLogMqBackListener(@Payload String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception {
+        if (StringUtils.isEmpty(message)) {
+            channel.basicAck(deliveryTag, false);
+            return;
+        }
+
+        try {
+            DeviceLogData deviceLogData = JSON.parseObject(message, DeviceLogData.class);
+            deviceLogService.saveDeviceLogToInfluxdb(deviceLogData);
+
+            // 成功写入后确认消息
+            channel.basicAck(deliveryTag, false);
+        } catch (JSONException je) {
+            // ⚠️ 毒丸保护：JSON解析异常不会通过重试恢复，直接丢弃消息并记录日志
+            LOG.error("JSON解析异常（不可恢复），消息将被丢弃: {}", je.getMessage(), je);
+            LOG.error("被丢弃的消息内容: {}", message);
+            channel.basicAck(deliveryTag, false);
+        } catch (Exception e) {
+            LOG.error("处理设备日志时发生异常: {}", e.getMessage(), e);
+
+            boolean messageHandled = false;
+
+            try {
+                boolean isInfluxTimeout = isInfluxDBTimeoutException(e);
 
+                if (isInfluxTimeout) {
+                    LOG.warn("检测到 InfluxDB 超时/连接异常，将消息发送到备份队列: {}", e.getMessage());
+                    amqpService.SendMessage("shengdilandevicelogback", message);
+                    LOG.info("消息已发送到备份队列，确认原消息");
+                    messageHandled = true;
+                } else {
+                    LOG.error("非超时异常，消息重新入队等待重试: {}", e.getMessage());
+                    channel.basicNack(deliveryTag, false, true);
+                    messageHandled = true;
+                    return;
+                }
+            } catch (Exception sendError) {
+                LOG.error("发送备份队列失败，原消息将重新入队: {}", sendError.getMessage(), sendError);
+                channel.basicNack(deliveryTag, false, true);
+                messageHandled = true;
+                return;
+            }
+
+            if (messageHandled) {
+                channel.basicAck(deliveryTag, false);
+            }
+        }
+    }
     /**
      * 更新设备下发命令日志
      *