ZAB 协议详解#

// ZooKeeper Watch 示例
ZooKeeper zk = new ZooKeeper("localhost:2181", 3000, watcher);

zk.exists("/config", new Watcher() {
    @Override
    public void process(WatchedEvent event) {
        System.out.println("/config 发生了变化: " + event);
        // 重新设置 Watch，继续监听
    }
});

// 触发 Watch
zk.setData("/config", "new-value".getBytes(), -1);

// ZAB 事务处理伪代码
public class zab.Proposer {
    public void handleRequest(Request request) {
        // 生成新 zxid：epoch << 32 | counter
        long newZxid = generateZxid(currentEpoch, counter++);

        // 构造事务
        Proposal proposal = new Proposal(newZxid, request);

        // 按顺序广播（ZAB 保证串行处理）
        broadcast(proposal);
    }
}

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 基于 ZooKeeper 的分布式锁实现
 * 利用 ZAB 的顺序保证，确保锁的公平性
 */
public class ZooKeeperDistributedLock implements Watcher {
    private final ZooKeeper zk;
    private final String lockPath;
    private String currentNode;
    private final CountDownLatch latch = new CountDownLatch(1);

    public ZooKeeperDistributedLock(ZooKeeper zk, String lockName) {
        this.zk = zk;
        this.lockPath = "/locks/" + lockName;
    }

    /**
     * 尝试获取锁
     * @return true 获取成功，false 被其他节点持有
     */
    public boolean tryLock() throws Exception {
        // 创建临时顺序节点
        currentNode = zk.create(
            lockPath + "/lock-",
            new byte[0],
            ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL
        );

        // 获取所有子节点并排序
        List<String> nodes = zk.getChildren(lockPath, false);
        Collections.sort(nodes);

        String smallestNode = nodes.get(0);

        // 如果自己是最小的，获得锁
        if (currentNode.endsWith(smallestNode)) {
            return true;
        }

        // 否则监听前一个节点
        int myIndex = nodes.indexOf(currentNode.substring(lockPath.length() + 1));
        String previousNode = nodes.get(myIndex - 1);

        // 注册 Watch：当前一个节点消失时，尝试重新获取锁
        zk.exists(lockPath + "/" + previousNode, this);

        // 等待被唤醒（通过 Watch 机制）
        latch.await();
        return true;
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.NodeDeleted) {
            latch.countDown(); // 前一个节点消失，唤醒尝试重新获取
        }
    }

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