|
|
@@ -42,7 +42,9 @@ public class WorkHoursCalcService {
|
|
|
private static final int MAX_RETRY = 2;
|
|
|
private static final int EXISTING_QUERY_MAX_ATTEMPTS = 3;
|
|
|
private static final long[] EXISTING_QUERY_RETRY_DELAYS_MS = {2_000L, 5_000L};
|
|
|
+ private static final long RATE_LIMIT_COOLDOWN_MS = 2_000L;
|
|
|
private RetrySleeper existingQueryRetrySleeper = Thread::sleep;
|
|
|
+ private RetrySleeper deferredWriteSleeper = Thread::sleep;
|
|
|
|
|
|
// prd 增量同步只覆盖最近 N 天工作日窗口 (含 today), 避免服务器停机/单次失败导致数据永久丢失
|
|
|
// fixme 窗口内已存在的记录一律 skip (补漏语义, 不刷新已写入字段); 字段刷新走独立接口 (backfill / cleanup)
|
|
|
@@ -824,9 +826,10 @@ public class WorkHoursCalcService {
|
|
|
Set<String> existKeys) {
|
|
|
AtomicInteger successCount = new AtomicInteger(0);
|
|
|
AtomicInteger failCount = new AtomicInteger(0);
|
|
|
- // fixme: 宜搭写接口有突发 QPS 上限,10 线程裸跑会零星触发「请求过于频繁」导致单条记录被丢(按人散落缺日,
|
|
|
- // 如吕加冕缺 6-16/6-25)。与 backfillCfEmployee 同款处方:20 QPS 全局限流 + 重试退避,补齐丢失记录。
|
|
|
+ // fixme 宜搭写接口失败时不能原地重试:首个请求可能已成功落库但响应丢失,连续 upsert 会生成重复记录。
|
|
|
+ // 首轮只写一次并记录失败;全部首轮任务结束后,逐条查询确认不存在,再补写一次。
|
|
|
RateLimiter yidaLimiter = RateLimiter.create(20.0);
|
|
|
+ Queue<DeferredWrite> deferredWrites = new ConcurrentLinkedQueue<>();
|
|
|
ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
|
|
|
final Set<String> existSet = existKeys == null ? Collections.emptySet() : existKeys;
|
|
|
final LocalDate today = LocalDate.now();
|
|
|
@@ -898,31 +901,18 @@ public class WorkHoursCalcService {
|
|
|
continue;
|
|
|
}
|
|
|
}
|
|
|
- boolean written = false;
|
|
|
- for (int retry = 0; retry <= MAX_RETRY; retry++) {
|
|
|
- try {
|
|
|
- yidaLimiter.acquire();
|
|
|
- upsertDailyHours(empId, managerIds, workDay, info);
|
|
|
- written = true;
|
|
|
- break;
|
|
|
- } catch (Exception e) {
|
|
|
- if (retry < MAX_RETRY) {
|
|
|
- log.warn("员工{} {} 写入失败(第{}次重试)", empId, workDay, retry + 1);
|
|
|
- // fixme: 「请求过于频繁」是瞬时突发限流,退避后再试才能补齐,紧贴重试会命中同一限流窗口
|
|
|
- try {
|
|
|
- Thread.sleep(1000L * (retry + 1));
|
|
|
- } catch (InterruptedException ie) {
|
|
|
- Thread.currentThread().interrupt();
|
|
|
- }
|
|
|
- } else {
|
|
|
- log.error("员工{} {} 写入失败(已重试{}次)", empId, workDay, MAX_RETRY, e);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- if (written) {
|
|
|
+ try {
|
|
|
+ yidaLimiter.acquire();
|
|
|
+ upsertDailyHours(empId, managerIds, workDay, info);
|
|
|
successCount.incrementAndGet();
|
|
|
- } else {
|
|
|
- failCount.incrementAndGet();
|
|
|
+ } catch (Exception e) {
|
|
|
+ boolean rateLimited = isRateLimitError(e);
|
|
|
+ deferredWrites.add(new DeferredWrite(empId, managerIds, workDay, info));
|
|
|
+ log.warn("员工{} {} 首轮写入失败,已记录并跳过,待首轮全部完成后查询再补写: rateLimited={}, code={}, message={}",
|
|
|
+ empId, workDay, rateLimited, queryErrorCode(e), e.getMessage());
|
|
|
+ if (rateLimited) {
|
|
|
+ sleepAfterRateLimit(empId, workDay);
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
}));
|
|
|
@@ -935,6 +925,7 @@ public class WorkHoursCalcService {
|
|
|
log.error("线程执行异常", e);
|
|
|
}
|
|
|
}
|
|
|
+ retryDeferredWrites(deferredWrites, yidaLimiter, successCount, failCount);
|
|
|
if (skippedFuture.get() > 0) {
|
|
|
log.info("未来日期过滤: 跳过{}条 workDay > today({}) 的记录", skippedFuture.get(), today);
|
|
|
}
|
|
|
@@ -963,6 +954,77 @@ public class WorkHoursCalcService {
|
|
|
return new int[]{successCount.get(), failCount.get()};
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 首轮全部结束后处理失败项;每项必须先查询,确认员工+日期不存在才允许补写一次。
|
|
|
+ *
|
|
|
+ * @param deferredWrites 首轮失败项
|
|
|
+ * @param yidaLimiter 宜搭写请求全局限流器
|
|
|
+ * @param successCount 成功计数
|
|
|
+ * @param failCount 最终失败计数
|
|
|
+ */
|
|
|
+ private void retryDeferredWrites(Queue<DeferredWrite> deferredWrites,
|
|
|
+ RateLimiter yidaLimiter,
|
|
|
+ AtomicInteger successCount,
|
|
|
+ AtomicInteger failCount) {
|
|
|
+ if (deferredWrites.isEmpty()) return;
|
|
|
+ log.info("首轮写入全部完成,开始处理{}条失败项(先查询、后补写)", deferredWrites.size());
|
|
|
+ DeferredWrite item;
|
|
|
+ while ((item = deferredWrites.poll()) != null) {
|
|
|
+ int existingCount;
|
|
|
+ try {
|
|
|
+ existingCount = queryRequiredHoursCount(item.employeeId, item.workDay);
|
|
|
+ } catch (Exception e) {
|
|
|
+ failCount.incrementAndGet();
|
|
|
+ log.error("员工{} {} 补写前查询失败,本轮不再写入: code={}, message={}",
|
|
|
+ item.employeeId, item.workDay, queryErrorCode(e), e.getMessage(), e);
|
|
|
+ if (isRateLimitError(e)) {
|
|
|
+ sleepAfterRateLimit(item.employeeId, item.workDay);
|
|
|
+ }
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (existingCount == 1) {
|
|
|
+ successCount.incrementAndGet();
|
|
|
+ log.info("员工{} {} 首轮请求已实际落库,查询确认存在,跳过补写", item.employeeId, item.workDay);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (existingCount > 1) {
|
|
|
+ failCount.incrementAndGet();
|
|
|
+ log.error("员工{} {} 补写前已存在{}条重复记录,本轮不再写入", item.employeeId, item.workDay, existingCount);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ try {
|
|
|
+ yidaLimiter.acquire();
|
|
|
+ upsertDailyHours(item.employeeId, item.managerIds, item.workDay, item.personnelInfo);
|
|
|
+ successCount.incrementAndGet();
|
|
|
+ log.info("员工{} {} 延后补写成功", item.employeeId, item.workDay);
|
|
|
+ } catch (Exception e) {
|
|
|
+ failCount.incrementAndGet();
|
|
|
+ log.error("员工{} {} 延后补写失败,本轮不再重试: code={}, message={}",
|
|
|
+ item.employeeId, item.workDay, queryErrorCode(e), e.getMessage(), e);
|
|
|
+ if (isRateLimitError(e)) {
|
|
|
+ sleepAfterRateLimit(item.employeeId, item.workDay);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 限流后等待冷却;只影响后续任务,不原地重试当前写请求。
|
|
|
+ *
|
|
|
+ * @param employeeId 员工 ID
|
|
|
+ * @param workDay 应填报日期
|
|
|
+ */
|
|
|
+ private void sleepAfterRateLimit(String employeeId, LocalDate workDay) {
|
|
|
+ try {
|
|
|
+ deferredWriteSleeper.sleep(RATE_LIMIT_COOLDOWN_MS);
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ log.warn("员工{} {} 限流冷却等待被中断", employeeId, workDay);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* 清理同一员工同一日期的重复应报工时(一次性接口)。
|
|
|
* prd: dryRun 默认由 Controller 设为 true;离职日后的记录交由 cleanupAfterOffline 独立处理,避免重复删除。
|
|
|
@@ -1548,6 +1610,40 @@ public class WorkHoursCalcService {
|
|
|
return keys;
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 查询指定员工、指定日期当前实际存在的应填报工时数量。
|
|
|
+ * ppExt: 仅用于写失败后的延后补偿,数量为 0 才允许再次写入;1 视为首轮已落库,>1 视为重复异常。
|
|
|
+ *
|
|
|
+ * @param employeeId 员工 ID
|
|
|
+ * @param workDay 应填报日期
|
|
|
+ * @return 当前匹配记录数
|
|
|
+ */
|
|
|
+ private int queryRequiredHoursCount(String employeeId, LocalDate workDay) {
|
|
|
+ ZoneId zone = ZoneId.systemDefault();
|
|
|
+ long startMs = workDay.atStartOfDay(zone).toInstant().toEpochMilli();
|
|
|
+ long endMs = workDay.plusDays(1).atStartOfDay(zone).toInstant().toEpochMilli() - 1;
|
|
|
+ Map<String, Object> search = new LinkedHashMap<>();
|
|
|
+ search.put("employeeField_mmd8onl4", Collections.singletonList(employeeId));
|
|
|
+ search.put("dateField_mmd8onl5", Arrays.asList(startMs, endMs));
|
|
|
+
|
|
|
+ DDR_New result = queryExistingHoursPage(
|
|
|
+ whConf.getYidaAppType(),
|
|
|
+ whConf.getYidaSystemToken(),
|
|
|
+ JSON.toJSONString(search),
|
|
|
+ workDay,
|
|
|
+ workDay,
|
|
|
+ 1,
|
|
|
+ YDConf.PAGE_SIZE_LIMIT);
|
|
|
+ if (result == null) {
|
|
|
+ throw new IllegalStateException("查询员工单日应填报工时返回空响应");
|
|
|
+ }
|
|
|
+ long totalCount = result.getTotalCount();
|
|
|
+ Object data = result.getData();
|
|
|
+ long returnedCount = data instanceof Collection ? ((Collection<?>) data).size() : 0L;
|
|
|
+ long confirmedCount = Math.max(totalCount, returnedCount);
|
|
|
+ return confirmedCount > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) confirmedCount;
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* 查询一页已存在工时;宜搭临时服务异常时仅重试当前页,避免前面已完成的分页重复执行。
|
|
|
*/
|
|
|
@@ -1607,6 +1703,39 @@ public class WorkHoursCalcService {
|
|
|
|| containsRetryableQueryMarker(mcException.getMessage());
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 判断异常是否为明确限流,仅限流触发冷却等待;普通写失败只记录并跳过。
|
|
|
+ *
|
|
|
+ * @param error 调用异常
|
|
|
+ * @return 是否为限流异常
|
|
|
+ */
|
|
|
+ private static boolean isRateLimitError(Throwable error) {
|
|
|
+ for (Throwable cause = error; cause != null; cause = cause.getCause()) {
|
|
|
+ if (cause instanceof McException) {
|
|
|
+ McException mcException = (McException) cause;
|
|
|
+ if (containsRateLimitMarker(mcException.getCode())
|
|
|
+ || containsRateLimitMarker(mcException.getMessage())) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ } else if (containsRateLimitMarker(cause.getMessage())) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static boolean containsRateLimitMarker(String value) {
|
|
|
+ if (value == null || value.trim().isEmpty()) return false;
|
|
|
+ String normalized = value.trim().toLowerCase(Locale.ROOT);
|
|
|
+ return "429".equals(normalized)
|
|
|
+ || "90002".equals(normalized)
|
|
|
+ || normalized.contains("too many request")
|
|
|
+ || normalized.contains("rate limit")
|
|
|
+ || normalized.contains("throttl")
|
|
|
+ || normalized.contains("请求过于频繁")
|
|
|
+ || normalized.contains("限流");
|
|
|
+ }
|
|
|
+
|
|
|
private static boolean containsRetryableQueryMarker(String value) {
|
|
|
if (value == null || value.trim().isEmpty()) return false;
|
|
|
String normalized = value.trim().toLowerCase(Locale.ROOT);
|
|
|
@@ -1646,6 +1775,26 @@ public class WorkHoursCalcService {
|
|
|
void sleep(long delayMs) throws InterruptedException;
|
|
|
}
|
|
|
|
|
|
+ /** 首轮写失败后延迟到第二阶段处理的不可变任务快照。 */
|
|
|
+ private static final class DeferredWrite {
|
|
|
+ private final String employeeId;
|
|
|
+ private final List<String> managerIds;
|
|
|
+ private final LocalDate workDay;
|
|
|
+ private final Map<String, Object> personnelInfo;
|
|
|
+
|
|
|
+ private DeferredWrite(String employeeId,
|
|
|
+ List<String> managerIds,
|
|
|
+ LocalDate workDay,
|
|
|
+ Map<String, Object> personnelInfo) {
|
|
|
+ this.employeeId = employeeId;
|
|
|
+ this.managerIds = managerIds == null
|
|
|
+ ? Collections.emptyList()
|
|
|
+ : new ArrayList<>(managerIds);
|
|
|
+ this.workDay = workDay;
|
|
|
+ this.personnelInfo = personnelInfo;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
// ==================== 数据查询 ====================
|
|
|
|
|
|
/**
|