《Java 并发编程·性能诊断与事故》11.2 分布式锁的租约与续期

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

public final class LeaseLab {

    /** 人造单调时钟，单位毫秒；advance 表示所有线程都停住了这么久。 */
    static final class ManualClock {
        private final AtomicLong millis = new AtomicLong();
        long now() { return millis.get(); }
        void advance(long delta) { millis.addAndGet(delta); }
    }

    record Lease(String client, long fence, long expiresAt) {}

    /** 租约服务：单机锁保护 owner/fence/expiresAt，语义对齐 SET NX PX。 */
    static final class LeaseService {
        private final ReentrantLock guard = new ReentrantLock();
        private final ManualClock clock;
        private String owner;
        private long fence;
        private long expiresAt = -1;

        LeaseService(ManualClock clock) { this.clock = clock; }

        Lease acquire(String client, long ttlMillis) {
            guard.lock();
            try {
                if (owner != null && clock.now() < expiresAt) {
                    return null;                 // 仍被有效持有
                }
                owner = client;
                fence++;
                expiresAt = clock.now() + ttlMillis;
                return new Lease(client, fence, expiresAt);
            } finally {
                guard.unlock();
            }
        }

        /** 只有“还是我、且没过期”才允许续期；看门狗走的就是这条路。 */
        boolean renew(Lease lease, long ttlMillis) {
            guard.lock();
            try {
                if (!lease.client().equals(owner) || lease.fence() != fence) {
                    return false;                // 已被别人接管
                }
                if (clock.now() >= expiresAt) {
                    return false;                // 已过期，续期窗口关闭
                }
                expiresAt = clock.now() + ttlMillis;
                return true;
            } finally {
                guard.unlock();
            }
        }
    }
}

/** 存储侧：只认递增的 fence，小于等于已见最大值的写入一律拒绝。 */
static final class FencedStore {
    private long lastFence;
    private String value = "<empty>";
    private int staleRejected;

    synchronized boolean write(long fence, String v) {
        if (fence <= lastFence) {
            staleRejected++;
            return false;
        }
        lastFence = fence;
        value = v;
        return true;
    }

    synchronized String value() { return value; }
    synchronized int staleRejected() { return staleRejected; }
}

private static void resurrectionDemo(boolean useFence) {
    ManualClock clock = new ManualClock();
    LeaseService svc = new LeaseService(clock);
    FencedStore store = new FencedStore();

    Lease a = svc.acquire("A", 100);
    store.write(a.fence(), "A-old");            // A 已写过一次

    clock.advance(250);                         // A 停住：STW / 分区，看门狗也停
    Lease b = svc.acquire("B", 100);            // B 合法接管
    boolean bOk = store.write(b.fence(), "B-new");

    boolean aOk = useFence
            ? store.write(a.fence(), "A-stale") // A 复活，带旧 fence
            : writeUnfenced(store, "A-stale");

    System.out.printf("useFence=%-5s owner=%s A.fence=%d B.fence=%d bWrote=%s aWrote=%s final=%s rejected=%d%n",
            useFence, svc.owner(), a.fence(), b.fence(), bOk, aOk, store.value(), store.staleRejected());
}

--- 1. 租约 + 续期（看门狗）---
acquire fence=1 renewed=true serverOwner=A
now=110 localRecordExpiry=100 serverExpiry=130 validByRecord=false validByServer=true
--- 2. 旧持有者复活：无 fencing vs 有 fencing ---
useFence=false owner=B A.fence=1 B.fence=2 bWrote=true aWrote=true final=A-stale rejected=0
useFence=true  owner=B A.fence=1 B.fence=2 bWrote=true aWrote=false final=B-new rejected=1
--- 3. 客户端时钟漂移导致的误判 ---
clientDeadline=100 serverNow=150 clientNow=-50 clientThinksValid=true serverSaysValid=false

UPDATE book
   SET stock = stock - 1,
       fence = ?
 WHERE isbn = ?
   AND fence < ?;

Lease lease = svc.acquire("A", 100);
long clientDeadline = lease.expiresAt();   // 获取时两端时钟一致，本地记下绝对截止点
serverClock.advance(150);                  // 服务端：已过期
long clientOffset = -200;                  // 之后 NTP 把客户端时钟回拨 200ms
long clientNow = serverClock.now() + clientOffset;
boolean clientThinksValid = clientNow < clientDeadline;   // true
boolean serverSaysValid = serverClock.now() < lease.expiresAt();  // false

续期返回 false 或超时
  -> 设置本地 leaseLost 标志（volatile）
  -> 业务线程在每个写操作前检查该标志，发现丢失立即中止
  -> 回滚未提交的事务 / 走补偿
  -> 不再尝试 reclaim，重新排队抢锁

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

interface LeaseStore {
    Lease acquire(String key, long ttlMillis);
    boolean renew(Lease lease, long ttlMillis);
    void release(Lease lease);
}

record Lease(String key, String owner, long fence, long expiresAt) {}

final class LeaseGuard implements AutoCloseable {
    private final LeaseStore store;
    private final Lease lease;
    private final long ttlMillis;
    private final AtomicBoolean lost = new AtomicBoolean();
    private final ScheduledExecutorService renewer =
            Executors.newSingleThreadScheduledExecutor(r ->
                    Thread.ofPlatform().name("lease-renewer").daemon().unstarted(r));

    LeaseGuard(LeaseStore store, Lease lease, long ttlMillis) {
        this.store = store;
        this.lease = lease;
        this.ttlMillis = ttlMillis;
        long period = Math.max(1, ttlMillis / 3);
        renewer.scheduleAtFixedRate(() -> {
            if (lost.get()) {
                return;                          // 已经丢了，不再续
            }
            if (!store.renew(lease, ttlMillis)) {
                lost.set(true);                  // 业务线程看到后必须立即停写
            }
        }, period, period, TimeUnit.MILLISECONDS);
    }

    boolean lost() { return lost.get(); }

    @Override
    public void close() {
        renewer.shutdownNow();
        store.release(lease);
    }
}