diff --git a/inspect-main/inspect-main-api/src/main/java/com/inspect/api/domain/ReportPushMessage.java b/inspect-main/inspect-main-api/src/main/java/com/inspect/api/domain/ReportPushMessage.java index 6361247..d56a094 100644 --- a/inspect-main/inspect-main-api/src/main/java/com/inspect/api/domain/ReportPushMessage.java +++ b/inspect-main/inspect-main-api/src/main/java/com/inspect/api/domain/ReportPushMessage.java @@ -20,7 +20,4 @@ public class ReportPushMessage implements Serializable { /** 报告详情(reportInfo JSON 字符串,发布时组装,消费者只负责下载文件并上传) */ private String reportInfo; - - /** 消费重试次数(上传失败后 MQ 重投递计数) */ - private int retryCount; } diff --git a/inspect-main/inspect-main-api/src/main/java/com/inspect/api/service/ReportUploadConsumer.java b/inspect-main/inspect-main-api/src/main/java/com/inspect/api/service/ReportUploadConsumer.java index 0bec7a2..67d515c 100644 --- a/inspect-main/inspect-main-api/src/main/java/com/inspect/api/service/ReportUploadConsumer.java +++ b/inspect-main/inspect-main-api/src/main/java/com/inspect/api/service/ReportUploadConsumer.java @@ -16,13 +16,17 @@ import java.io.InputStream; * 巡检报告上传消费者:从 report.upload.queue 取消息,下载报告 docx, * 通过 {@link ReportUploadService} HTTP 上传到外部平台(POST /jb-nessp-external/api/uploadReport)。 *
- * 上传失败抛出异常使消息重投递(重试次数由消息 retryCount 控制,超过 {@code mq-max-retries} 丢弃)。 + * 重试策略:上传失败就在当前消费里重试,最多 {@code inspect.report-push.mq-max-retries} 次(含首次); + * 仍然失败则正常返回(消息 ack 出队 = 丢弃),不抛异常,避免 MQ 自动重投递造成无限重投递。 */ @Slf4j @Component public class ReportUploadConsumer { - /** 上传失败最大重试次数(超过后丢弃,避免无限重投) */ + /** 两次尝试之间的间隔(毫秒) */ + private static final long RETRY_INTERVAL_MILLIS = 3000; + + /** 上传失败最大尝试次数(含首次),超过后直接丢弃消息 */ @Value("${inspect.report-push.mq-max-retries:3}") private int maxRetries; @@ -47,24 +51,33 @@ public class ReportUploadConsumer { log.warn("[REPORT_PUSH_MQ] 报告上传消息为空,丢弃"); return; } - try { - boolean ok = upload(message); - if (ok) { - log.info("[REPORT_PUSH_MQ] 报告上传成功: reportId={}", message.getReportId()); - } else { - throw new RuntimeException("report upload failed, will be requeued"); + int maxAttempts = Math.max(1, maxRetries); + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + try { + if (upload(message)) { + log.info("[REPORT_PUSH_MQ] 报告上传成功: reportId={}, 第{}/{}次尝试", + message.getReportId(), attempt, maxAttempts); + return; + } + log.warn("[REPORT_PUSH_MQ] 报告上传失败, 第{}/{}次尝试: reportId={}", + attempt, maxAttempts, message.getReportId()); + } catch (Exception e) { + log.warn("[REPORT_PUSH_MQ] 报告上传异常, 第{}/{}次尝试: reportId={}, err={}", + attempt, maxAttempts, message.getReportId(), e.getMessage()); } - } catch (Exception e) { - message.setRetryCount(message.getRetryCount() + 1); - if (message.getRetryCount() > maxRetries) { - log.error("[REPORT_PUSH_MQ] 报告上传超过最大重试次数({}), 丢弃: reportId={}, err: {}", - maxRetries, message.getReportId(), e.getMessage()); - return; + if (attempt < maxAttempts) { + try { + Thread.sleep(RETRY_INTERVAL_MILLIS); + } catch (InterruptedException ie) { + // 应用停机等中断场景:抛出让消息重回队列,避免丢消息 + Thread.currentThread().interrupt(); + throw new RuntimeException("report upload retry interrupted", ie); + } } - log.warn("[REPORT_PUSH_MQ] 报告上传失败(第{}次), 消息重投递: reportId={}, err: {}", - message.getRetryCount(), message.getReportId(), e.getMessage()); - throw new RuntimeException("report upload failed, will be requeued", e); } + // 超过最大尝试次数:正常返回即 ack 出队,消息被丢弃,不会再被投递 + log.error("[REPORT_PUSH_MQ] 报告上传超过最大尝试次数({}), 丢弃消息: reportId={}, filePath={}", + maxAttempts, message.getReportId(), message.getFilePath()); } /**