Browse Source

圣地蓝设备日志备份队列消费功能

master
bgy 4 days ago
parent
commit
18eebaa64e
1 changed files with 48 additions and 0 deletions
  1. +48
    -0
      src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java

+ 48
- 0
src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java View File

@ -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);
}
}
}
/** /**
* 更新设备下发命令日志 * 更新设备下发命令日志
* *


Loading…
Cancel
Save