ZooKeeper分布式协调服务

引言

ZooKeeper是分布式协调服务。本文介绍其核心原理和应用。

一、核心概念

ZNode: 数据节点
- 持久节点
- 临时节点
- 顺序节点

Watch: 监听机制
Session: 会话
ACL: 访问控制

二、基本操作

ZooKeeper zk = new ZooKeeper("localhost:2181", 5000, event -> {
    System.out.println("Event: " + event.getType());
});

// 创建节点
zk.create("/mynode", "data".getBytes(), 
    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

// 读取节点
byte[] data = zk.getData("/mynode", false, null);

// 更新节点
zk.setData("/mynode", "new data".getBytes(), -1);

// 删除节点
zk.delete("/mynode", -1);

// 监听节点
zk.exists("/mynode", event -> {
    System.out.println("Node changed");
});

三、应用场景

3.1 服务注册与发现

// 服务注册
String servicePath = "/services/userService";
zk.create(servicePath, 
    "192.168.1.100:8080".getBytes(),
    ZooDefs.Ids.OPEN_ACL_UNSAFE,
    CreateMode.EPHEMERAL);

// 服务发现
List<String> services = zk.getChildren("/services", true);

3.2 分布式锁

public class DistributedLock {
    private ZooKeeper zk;
    private String lockPath = "/locks/mylock";
    
    public void lock() throws Exception {
        String path = zk.create(lockPath + "/lock-",
            new byte[0],
            ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL);
        
        while (true) {
            List<String> children = zk.getChildren("/locks/mylock", false);
            Collections.sort(children);
            
            if (path.endsWith(children.get(0))) {
                return; // 获得锁
            }
            
            // 等待前一个节点删除
            String prevNode = null;
            for (int i = 0; i < children.size(); i++) {
                if (path.endsWith(children.get(i))) {
                    prevNode = children.get(i - 1);
                    break;
                }
            }
            
            CountDownLatch latch = new CountDownLatch(1);
            zk.exists("/locks/mylock/" + prevNode, event -> {
                latch.countDown();
            });
            latch.await();
        }
    }
    
    public void unlock() throws Exception {
        zk.delete(lockPath, -1);
    }
}

四、选举机制

// Leader选举
public class LeaderElection {
    public void electLeader() throws Exception {
        String path = zk.create("/election/candidate-",
            new byte[0],
            ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL);
        
        List<String> candidates = zk.getChildren("/election", false);
        Collections.sort(candidates);
        
        if (path.endsWith(candidates.get(0))) {
            System.out.println("I am the leader!");
        } else {
            System.out.println("I am a follower");
        }
    }
}