《Java 并发编程·线程与内存模型》11.3 调度漂移、幂等与补偿

// 下一次 = 上一次的「计划时刻」+ period
scheduler.scheduleAtFixedRate(task, 0, 200, TimeUnit.MILLISECONDS);

// 下一次 = 本次「执行结束」+ delay
scheduler.scheduleWithFixedDelay(task, 0, 200, TimeUnit.MILLISECONDS);

理想网格：0, 200, 400, 600, ...        （第 k 次 = k × period）
漂移(k) = 实际开始(k) - k × period

Runnable task = () -> {
    long at = (System.nanoTime() - t0) / 1_000_000;
    starts.add(at);
    try {
        Thread.sleep(80);                 // 模拟执行耗时
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    done.countDown();
};

== fixedRate  period=200ms work=80ms ==
  第 1 次 开始 +   1ms（理想 +   0ms，落后 1ms）
  第 2 次 开始 + 203ms（理想 + 200ms，落后 3ms，距上次 202ms）
  第 3 次 开始 + 401ms（理想 + 400ms，落后 1ms，距上次 198ms）
  第 4 次 开始 + 604ms（理想 + 600ms，落后 4ms，距上次 203ms）
  第 5 次 开始 + 801ms（理想 + 800ms，落后 1ms，距上次 197ms）
  第 6 次 开始 +1003ms（理想 +1000ms，落后 3ms，距上次 202ms）

== fixedDelay period=200ms work=80ms ==
  第 1 次 开始 +   0ms（理想 +   0ms，落后 0ms）
  第 2 次 开始 + 282ms（理想 + 200ms，落后 82ms，距上次 282ms）
  第 3 次 开始 + 565ms（理想 + 400ms，落后 165ms，距上次 283ms）
  第 4 次 开始 + 856ms（理想 + 600ms，落后 256ms，距上次 291ms）
  第 5 次 开始 +1156ms（理想 + 800ms，落后 356ms，距上次 300ms）
  第 6 次 开始 +1444ms（理想 +1000ms，落后 444ms，距上次 288ms）

== fixedRate period=100ms work=300ms ==
  第 1 次 开始 +    1ms（理想网格 +   0ms，落后    1ms）
  第 2 次 开始 +  301ms（理想网格 + 100ms，落后  201ms）
  第 3 次 开始 +  601ms（理想网格 + 200ms，落后  401ms）
  第 4 次 开始 +  901ms（理想网格 + 300ms，落后  601ms）
  第 5 次 开始 + 1210ms（理想网格 + 400ms，落后  810ms）

private volatile long lastSlot = -1;

private void guardedTick() {
    long slot = System.nanoTime() / TimeUnit.MILLISECONDS.toNanos(PERIOD_MS);
    long behind = slot - lastSlot;
    if (lastSlot >= 0 && behind > MAX_CATCH_UP) {
        // 落后超过 3 个周期：只做一次，把基准拉回现在
        lastSlot = slot;
        doWork();
        return;
    }
    if (slot <= lastSlot) {
        return;                      // 同一周期的重复触发，直接丢弃
    }
    lastSlot = slot;
    doWork();
}

private final AtomicLong processedSlot = new AtomicLong(-1);

private void mergedTick() {
    long slot = System.currentTimeMillis() / PERIOD_MS;
    long prev = processedSlot.get();
    if (slot <= prev) {
        return;                                  // 这个位格已经处理过
    }
    if (!processedSlot.compareAndSet(prev, slot)) {
        return;                                  // 并发触发时只有一个赢家
    }
    doWorkForSlot(slot);
}

public final class IdempotentJob {
    private final ConcurrentHashMap<String, Boolean> seen = new ConcurrentHashMap<>();

    public boolean runOnce(String key, Runnable work) {
        if (seen.putIfAbsent(key, Boolean.TRUE) != null) {
            return false;
        }
        work.run();
        return true;
    }
}

== 幂等键去重 ==
  key=daily-fine@12 -> 执行
  key=daily-fine@12 -> 跳过（重复投递）
  key=daily-fine@12 -> 跳过（重复投递）
  key=daily-fine@13 -> 执行
  key=daily-fine@13 -> 跳过（重复投递）
  真实执行 2 次，抑制重复 3 次

CREATE UNIQUE INDEX uk_job_key ON job_execution (job_key);

public final class OverdueFineJob {
    private final LoanRepository loans;
    private final FineLedger ledger;                    // 罚款台账端口，不属于领域契约
    private final Function<LocalDate, List<String>> overdueIds;

    public OverdueFineJob(LoanRepository loans, FineLedger ledger,
                          Function<LocalDate, List<String>> overdueIds) {
        this.loans = loans;
        this.ledger = ledger;
        this.overdueIds = overdueIds;
    }

    public void runFor(LocalDate day) {
        String key = "overdue-fine:" + day;
        if (!ledger.tryAcquire(key)) {                  // 唯一索引插入，冲突即已执行
            return;
        }
        for (String loanId : overdueIds.apply(day)) {
            Loan loan = loans.findById(loanId).orElse(null);
            if (loan == null) {
                continue;
            }
            try {
                Instant deadline = day.atStartOfDay(ZoneId.systemDefault()).toInstant();
                long amount = ChronoUnit.DAYS.between(loan.dueAt(), deadline) * 1L;
                ledger.charge(loan.loanId(), day, amount);       // 每天 1 元
            } catch (RuntimeException e) {
                ledger.recordPending(loan.loanId(), day);        // 失败留下补偿记录
            }
        }
    }

    /** 补偿：下一轮先处理上一轮留下的 pending，再跑当天 */
    public void compensate(LocalDate day) {
        for (PendingFine p : ledger.pendingBefore(day)) {
            try {
                ledger.charge(p.loanId(), p.day(), p.amount());
                ledger.clearPending(p);
            } catch (RuntimeException e) {
                // 仍然失败，留到下一轮；pending 表本身就是重试队列
            }
        }
    }
}