63 lines
2.9 KiB
Java
63 lines
2.9 KiB
Java
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);
|
|
}
|
|
}
|