日志记录提交

2026-08-05 14:20:04 +08:00
parent 264458920b
commit 06f7f9d2f2
83 changed files with 3256 additions and 27 deletions
@@ -0,0 +1,62 @@
+package com.baiyue.common.audit;
+
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQHeaders;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.scheduling.annotation.Scheduled;
+
+/** 带租约、退避、死信和自动补偿的审计 Outbox 发布器。 */
+public class AuditOutboxPublisher {
+    private static final Logger log = LoggerFactory.getLogger(AuditOutboxPublisher.class);
+    private static final int[] BACKOFF_SECONDS = {5, 15, 60, 300, 900};
+    private final AuditOutboxRepository repository;
+    private final RocketMQTemplate rocketMQ;
+    private final AuditProperties properties;
+    private final String owner;
+
+    public AuditOutboxPublisher(AuditOutboxRepository repository, RocketMQTemplate rocketMQ,
+                                AuditProperties properties, String owner) {
+        this.repository = repository;
+        this.rocketMQ = rocketMQ;
+        this.properties = properties;
+        this.owner = owner;
+    }
+
+    @Scheduled(fixedDelayString = "${baiyue.audit.outbox.poll-interval-ms:5000}")
+    public void publish() {
+        AuditProperties.Outbox config = properties.getOutbox();
+        repository.releaseExpiredClaims();
+        repository.compensateDead(config.getCompensationDelayMinutes(), config.getMaxCompensations());
+        repository.deleteSent(config.getRetentionDays());
+        for (String eventId : repository.publishableIds(config.getBatchSize())) {
+            if (!repository.claim(eventId, owner, config.getLeaseSeconds())) {
+                continue;
+            }
+            AuditOutboxRepository.OutboxRow row = repository.find(eventId);
+            if (row == null) {
+                continue;
+            }
+            try {
+                var message = MessageBuilder.withPayload(row.payloadJson())
+                    .setHeader(RocketMQHeaders.KEYS, eventId)
+                    .setHeader("eventId", eventId)
+                    .build();
+                rocketMQ.syncSend(properties.getRocketmq().getTopic() + ":AUDIT", message,
+                    properties.getRocketmq().getSendTimeoutMs());
+                repository.sent(eventId, owner);
+            } catch (RuntimeException exception) {
+                int index = Math.min(Math.max(row.attemptCount(), 1) - 1, BACKOFF_SECONDS.length - 1);
+                repository.failed(eventId, owner, config.getMaxAttempts(), BACKOFF_SECONDS[index],
+                    safe(exception.getMessage()));
+                log.warn("审计事件投递失败 eventId={} attempt={}", eventId, row.attemptCount(), exception);
+            }
+        }
+    }
+
+    private static String safe(String value) {
+        String message = value == null ? "RocketMQ 发送失败" : value;
+        return message.length() <= 1000 ? message : message.substring(0, 1000);
+    }
+}