package jnpf.dmsService;
|
|
import jnpf.audit.sdk.annotation.AuditLog;
|
import jnpf.dmsEntity.fenfa.DmsBupinghengRetry;
|
import jnpf.dmsMapper.DmsBupinghengRetryMapper;
|
import jnpf.exception.DataException;
|
import lombok.RequiredArgsConstructor;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
import org.springframework.scheduling.annotation.EnableScheduling;
|
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.stereotype.Service;
|
import java.util.UUID;
|
|
@Slf4j
|
@Service
|
@EnableScheduling
|
@RequiredArgsConstructor
|
@ConditionalOnProperty(name = "dms.imbalance.retry.enabled", havingValue = "true")
|
public class DmsBupinghengRetryJob {
|
private final DmsBupinghengRetryMapper mapper;
|
private final DmsWenjianBupinghengService service;
|
@Value("${dms.imbalance.retry.batch-size:100}")
|
private int batchSize = 100;
|
@Value("${dms.imbalance.retry.max-attempts:5}")
|
private int maxAttempts = 5;
|
|
@Scheduled(cron = "${dms.imbalance.retry.cron:0 */5 * * * ?}")
|
@AuditLog(action = "FILE_DISTRIBUTION_IMBALANCE_SCAN", bizCodeExpr = "''", recordSuccess = false)
|
public void retryPending() {
|
// 每次 executeScheduled 经过独立代理事务；领取和失败回执不会随业务回滚丢失。
|
for (DmsBupinghengRetry candidate : mapper.selectPending(Math.max(1, Math.min(batchSize, 1000)), Math.max(1, maxAttempts))) {
|
String token = UUID.randomUUID().toString();
|
try {
|
if (mapper.claim(candidate, token, Math.max(1, maxAttempts)) != 1) continue;
|
DmsBupinghengRetry task = mapper.selectClaim(candidate, token);
|
if (task == null) continue;
|
try {
|
service.executeScheduled(task.getTenantId(), task.getDocumentId(), task.getVersion());
|
mapper.finish(task, token, true, null, 0);
|
} catch (Exception error) {
|
boolean stop = error instanceof DataException || task.getAttempts() >= Math.max(1, maxAttempts);
|
int delay = Math.min(3600, 300 * (1 << Math.min(task.getAttempts() - 1, 4)));
|
String message = error.getMessage() == null ? error.getClass().getSimpleName() : error.getMessage();
|
mapper.finish(task, token, stop, message.substring(0, Math.min(message.length(), 1000)), delay);
|
log.warn("不平衡补偿失败 tenant={} document={} attempts={} stopped={}",
|
task.getTenantId(), task.getDocumentId(), task.getAttempts(), stop);
|
}
|
} catch (Exception error) {
|
// 回执写入失败由持久租约到期恢复；不能终止本批其余登记单。
|
log.error("不平衡补偿调度异常 tenant={} document={}", candidate.getTenantId(), candidate.getDocumentId(), error);
|
}
|
}
|
}
|
}
|
