ZooKeeper 分布式锁实现#

// 持久节点：创建后一直存在，直到主动删除
zk.create("/config", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

// 临时节点：与客户端 session 绑定，session 断开后自动删除
zk.create("/worker-001", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

// 顺序节点：在创建时自动追加递增序号
zk.create("/lock/lock-", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
// 创建结果可能是：/lock/lock-0000000001

public class ZkDistributedLock implements Watcher {
    private final ZooKeeper zk;
    private final String lockPath;
    private final String currentNode;
    private final String waitNode;
    private final CountDownLatch latch = new CountDownLatch(1);
    private final AtomicBoolean locked = new AtomicBoolean(false);

    public ZkDistributedLock(ZooKeeper zk, String lockPath) throws Exception {
        this.zk = zk;
        this.lockPath = lockPath;

        // 1. 创建临时顺序节点
        currentNode = zk.create(
            lockPath + "/lock-",
            new byte[0],
            ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL
        );

        // 2. 检查自己是否是最小节点
        if (checkMinNode()) {
            locked.set(true);
            return;
        }

        // 3. 不是最小节点，监听前一个节点
        waitNode = getPrevNode();
        if (waitNode != null) {
            zk.getData(waitNode, true, null);
        }
    }

    private boolean checkMinNode() throws Exception {
        List<String> nodes = zk.getChildren(lockPath, false);
        Collections.sort(nodes);

        String minNode = nodes.get(0);
        return currentNode.endsWith(minNode);
    }

    private String getPrevNode() throws Exception {
        List<String> nodes = zk.getChildren(lockPath, false);
        Collections.sort(nodes);

        String currentShortName = currentNode.substring(currentNode.lastIndexOf('/') + 1);
        int idx = Collections.binarySearch(nodes, currentShortName);
        if (idx > 0) {
            return lockPath + "/" + nodes.get(idx - 1);
        }
        return null;
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.NodeDeleted) {
            try {
                // 前一个节点被删除，重新检查自己是否是最小
                if (checkMinNode()) {
                    locked.set(true);
                    latch.countDown();
                }
            } catch (Exception e) {
                // ignore
            }
        }
    }

    public void unlock() throws Exception {
        if (locked.get()) {
            zk.delete(currentNode, -1);
        }
    }
}

// 问题代码：每次被唤醒后都要重新 getData 注册 Watch
@Override
public void process(WatchedEvent event) {
    if (event.getType() == Event.EventType.NodeDeleted) {
        try {
            if (checkMinNode()) {
                locked.set(true);
                latch.countDown();
            } else {
                // 重新注册 Watch——这是羊群效应的触发点
                waitNode = getPrevNode();
                zk.getData(waitNode, true, null);
            }
        } catch (Exception e) { }
    }
}

// 使用信号量限制同时竞争锁的客户端数量
Semaphore semaphore = new Semaphore(50); // 最多50个客户端同时竞争

public boolean tryLock() {
    if (!semaphore.tryAcquire()) {
        return false; // 直接放弃，不注册 Watch
    }

    try {
        return acquireZkLock();
    } catch (Exception e) {
        semaphore.release();
        return false;
    }
}

// 收到通知后，不立即行动，随机等待一段时间
@Override
public void process(WatchedEvent event) {
    if (event.getType() == Event.EventType.NodeDeleted) {
        // 随机退避 0~500ms，减少并发冲突
        int jitter = new Random().nextInt(500);
        Thread.sleep(jitter);

        if (checkMinNode()) {
            locked.set(true);
            latch.countDown();
        }
    }
}

// 创建 Curator 客户端
CuratorFramework client = CuratorFrameworkFactory.newClient(
    "zk1:2181,zk2:2181,zk3:2181",
    5000,  // sessionTimeoutMs
    5000,  // connectionTimeoutMs
    new ExponentialBackoffRetry(1000, 3)
);
client.start();

// 创建可重入分布式锁
InterProcessMutex lock = new InterProcessMutex(client, "/lock/my-app");

// 加锁
try {
    lock.acquire(30, TimeUnit.SECONDS); // 等待30秒
    // 临界区
    processOrder();
} catch (Exception e) {
    throw new RuntimeException("获取锁失败", e);
} finally {
    try {
        lock.release(); // 释放锁
    } catch (Exception e) {
        // log error
    }
}

// InterProcessMutex 的可重入原理
// 在 ZK 中存储：thread UUID -> lock count
private String lockPath;

// 获取锁
public void acquire() throws Exception {
    if (!internalLock(-1, null)) {
        throw new IOException("Lost connection while trying to acquire lock: " + lockPath);
    }
}

private boolean internalLock(long time, TimeUnit unit) throws Exception {
    // 1. 尝试在 ZK 中创建锁节点
    String lockPath = client.create()
        .creatingParentsIfNeeded()
        .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
        .forPath(lockPath + "/lock-");

    // 2. 检查当前线程是否已经持有锁（可重入）
    // 原理：在 /locks 下存储 thread UUID -> count 的映射
    // 如果当前线程 UUID 已存在，增加 count 并直接返回成功
    if (isCurrentThreadMutexHolder(lockPath)) {
        incrementMutexHolderCount(lockPath);
        return true;
    }

    // 3. 按标准流程等待锁（检查最小序号 + Watch 前一个节点）
    return waitForLock(lockPath, lockPathData);
}

// 坑1：默认使用 OPEN_ACL_UNSAFE，生产环境应该配置 ACL
InterProcessMutex lock = new InterProcessMutex(client, "/lock/my-app");

// 坑2：锁自动续期依赖 session，如果 session 断开需要重试
// Curator 默认会续期，sessionTimeout 内不会自动释放锁
// 但极端情况下（网络分区）仍可能丢锁

// 坑3：InterProcessMutex 不支持公平锁和非公平锁的切换
// 默认是公平锁（按顺序获取），高并发下性能受限

// 选主代码：主节点崩溃后自动切换
InterProcessMutex leaderElection = new InterProcessMutex(client, "/leader");

if (leaderElection.acquire(0, TimeUnit.MILLISECONDS)) {
    // 拿到锁的实例成为主节点
    becomeLeader();
} else {
    // 其他实例等待主节点释放
    waitForLeadership();
}

// 当主节点崩溃时，session 断开，临时节点消失
// 下一个拿到锁的实例自动成为新主节点

// 监听配置变更：所有客户端 Watch 同一个配置节点
String configPath = "/config/app-config";

while (true) {
    byte[] data = client.getData().usingWatcher(new ConfigWatcher()).forPath(configPath);
    applyConfig(new String(data));

    // 配置更新时，Watch 触发，重新读取
    // 不需要轮询，由 ZK 服务端推送变更
}

private static class ConfigWatcher implements Watcher {
    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.NodeDataChanged) {
            // 触发重新读取配置
        }
    }
}

// 正确的 sessionTimeout 配置
CuratorFramework client = CuratorFrameworkFactory.newClient()
    .sessionTimeoutMs(30000)    // session 超时 30s
    .connectionTimeoutMs(10000) // 连接超时 10s
    .retryPolicy(new ExponentialBackoffRetry(1000, 3))
    .build();