etcd 分布式锁实现#

import io.etcd.jetcd.Client;
import io.etcd.jetcd.LockKey;
import io.etcd.jetcd.LockResponse;

public class EtcdDistributedLock {
    private final Client client;
    private final String lockName;
    private final LockKey lockKey;
    private LockResponse lockResponse;
    private byte[] leaseId;

    public EtcdDistributedLock(Client client, String lockName) {
        this.client = client;
        this.lockName = lockName;
        this.lockKey = client.getLockClient().getLockKey(lockName);
    }

    /**
     * 获取锁
     * @param waitTime  最大等待时间（秒）
     * @param leaseTime Lease 过期时间（秒）
     */
    public boolean tryLock(long waitTime, long leaseTime) throws Exception {
        // 1. 创建 Lease（带 TTL）
        Lease lease = client.getLeaseClient();
        long ttl = lease.grant(leaseTime);
        this.leaseId = Lease.toLeaseID(ttl);

        // 2. 尝试获取锁（原子操作：在指定 key 前缀下创建唯一 key）
        // 内部实现：
        //   - 在 /lock/{lockName}/ 路径下创建顺序 key
        //   - key 格式：/lock/{lockName}/{revision}
        //   - 使用 Txn 事务保证原子性
        try {
            lockResponse = client.getLockClient()
                .lock(lockName, ("instance-" + UUID.randomUUID()).getBytes(), leaseId)
                .get(waitTime, TimeUnit.SECONDS);
            return true;
        } catch (Exception e) {
            // 获取失败，撤销 Lease
            lease.revoke(Lease.toLeaseID(lease.grant(leaseTime).getID()));
            throw e;
        }
    }

    /**
     * 释放锁
     * 只需要撤销 Lease，key 会自动删除
     */
    public void unlock() throws Exception {
        if (lockResponse != null) {
            // Lease 撤销后，关联的 key 自动被删除
            client.getLeaseClient().revoke(leaseId);
        }
    }
}

// Lease 的核心能力

// 1. 创建带 TTL 的 Lease
long leaseId = lease.grant(30); // TTL=30秒

// 2. key 关联 Lease
kv.put(key, value, leaseId); // key 只在 lease 存活期间有效

// 3. 续期（KeepAlive）
// 业务执行时间超过 lease TTL 时，需要续期
CompletableFuture<LeaseKeepAliveResponse> resp = lease.keepAlive(leaseId);
resp.thenAccept(r -> {
    System.out.println("续期成功，TTL=" + r.getTTL());
});

// 4. 撤销 Lease
lease.revoke(leaseId); // key 立即被删除

// 5. Lease 自动过期
// 如果不续期，TTL 到期后 etcd 自动删除所有关联的 key

// 非公平锁的问题
// 场景：3 个客户端同时尝试获取 /lock/order
// 由于 etcd 的写操作经过 Raft，如果 C2 的写请求先于 C1 到达
// revision 的顺序就不完全等于客户端的到达顺序

// 实际情况：
// C1 的写请求在 T1 发起，但由于 Raft Leader 的处理延迟
// C2 在 T2 发起的写请求可能先被 Raft 复制并提交
// 导致 revision(C2) < revision(C1)

// 结果：C2 拿到了锁，但 C1 先发起请求
// ——这就是非公平性

// 基于 Txn 实现公平锁：先检查队列，再加锁
public class FairEtcdLock {
    private final KV kv;
    private final String lockPath;
    private final String myKey;

    public boolean acquire(long waitTime) throws Exception {
        // 1. 创建唯一 key（包含客户端标识 + 时间戳）
        myKey = lockPath + "/" + UUID.randomUUID().toString();

        // 2. 使用 Txn 确保原子性
        // Txn 的语义：先检查前缀下是否存在更小的 key
        // 如果不存在，才写入自己的 key
        Txn txn = kv.txn();
        txn.If(
            // 条件：不存在比当前 revision 更小的 key
            // 这保证了 FIFO 顺序
            new Range(kv, lockPath, myKey)
        ).Then(
            // 如果条件满足（自己是队列第一个），写入锁 key
            new Put(kv, myKey, lockValue, leaseId)
        ).Else(
            // 否则等待 Watch
            new Watch(kv, lockPath, myKey)
        );

        return txn.commit();
    }
}

// ❌ 错误：只在获取锁时调用一次 keepAlive
lock.tryLock(30, 30);
processOrder(); // 假设这个方法执行 45 秒
// Lease 在 30 秒后过期，但业务还没执行完
// 锁被其他客户端拿走，但当前客户端还在临界区
// → 两个客户端同时在临界区！

// ✅ 正确：异步持续续期
long leaseId = lease.grant(30);
CompletableFuture<Long> keepAlive = lease.keepAlive(leaseId);
keepAlive.thenAccept(response -> {
    // etcd 自动定期续期，直到显式撤销
    System.out.println("续期成功，TTL=" + response.getTTL());
});

// ❌ 错误：Watch 不处理断连
Watch watch = client.getWatchClient().watch(
    key,
    response -> {
        for (WatchEvent event : response.getEvents()) {
            // 处理事件
        }
    }
);
// 当 etcd 集群发生选举时，Watch 连接会断开
// 如果不处理，客户端会永远等待

// ✅ 正确：Watch 需要处理断连并重建
Watch watch = client.getWatchClient().watch(
    key,
    response -> {
        if (response.getEvents().isEmpty()) {
            // Watch 已过期，需要重新建立
            rebuildWatch();
            return;
        }
        for (WatchEvent event : response.getEvents()) {
            // 处理事件
        }
    }
);

// ❌ 错误：并发无上限
for (int i = 0; i < 200; i++) {
    final int idx = i;
    executor.submit(() -> {
        // 200 个并发请求同时打 etcd
        // etcd QPS 被打满，后续请求超时
        lock.tryLock();
    });
}

// ✅ 正确：限制并发度 + 指数退避
Semaphore semaphore = new Semaphore(50); // 限制同时竞争数
for (int i = 0; i < 200; i++) {
    executor.submit(() -> {
        if (!semaphore.tryAcquire()) {
            return; // 跳过，等待下次调度
        }
        try {
            lock.tryLock(10, 30); // 带超时
        } catch (Exception e) {
            // 指数退避重试
            retryWithBackoff();
        } finally {
            semaphore.release();
        }
    });
}

// ❌ 错误：把 Lease 和 Put 分开执行，不是原子的
long leaseId = lease.grant(30);
kv.put(key, value, leaseId); // 如果 Put 失败，Lease 已经创建了
// → Lease 泄漏

// ✅ 正确：使用 Txn 将 Lease 和 Put 打包
Txn txn = kv.txn();
txn.If(new Compare(lease, "GRANT", 0))  // Lease 已存在
    .Then(new Put(key, value, leaseId))
    .Else(new Put(key, value, lease.grant(30))); // 创建并使用新 Lease
txn.commit();