Apache Curator 完全指南- ZooKeeper Java客户端

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>5.8.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.8.0</version>
</dependency>

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorClientDemo {
    private static final String ZK_ADDRESS = "127.0.0.1:2181,127.0.0.1:2182";
    private static final int SESSION_TIMEOUT = 5000;
    private static final int CONNECTION_TIMEOUT = 5000;

    public static CuratorFramework createClient() {
        // 重试策略：初始休眠1秒，最多重试3次，休眠时间指数增长
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(ZK_ADDRESS)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .connectionTimeoutMs(CONNECTION_TIMEOUT)
                .retryPolicy(retryPolicy)
                .namespace("curator_demo")  // 命名空间：所有操作自动加此前缀
                .build();
        
        client.start();
        System.out.println("ZK客户端启动成功：" + client.isStarted());
        return client;
    }
}

// 指数退避：初始间隔1秒，最多重试3次
RetryPolicy policy1 = new ExponentialBackoffRetry(1000, 3);

// 固定间隔：每5秒重试一次，共重试3次
RetryPolicy policy2 = new RetryNTimes(3, 5000);

// 限时重试：每3秒重试一次，总耗时超过10秒则放弃
RetryPolicy policy3 = new RetryUntilElapsed(10000, 3000);

CuratorFramework client = createClient();

// 1. 创建持久节点（自动创建父节点）
client.create()
      .creatingParentsIfNeeded()
      .withMode(CreateMode.PERSISTENT)
      .forPath("/config/database", "jdbc:mysql://localhost:3306".getBytes());

// 2. 创建临时节点（会话结束自动删除）
client.create()
      .creatingParentsIfNeeded()
      .withMode(CreateMode.EPHEMERAL)
      .forPath("/services/user-service-001", "{\"port\":8080}".getBytes());

// 3. 创建持久顺序节点（自动生成序号）
String path = client.create()
        .creatingParentsIfNeeded()
        .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
        .forPath("/queue/task-", "task data".getBytes());
System.out.println("创建的顺序节点路径：" + path); 
// 输出类似：/queue/task-0000000001

// 读取节点数据
byte[] data = client.getData().forPath("/config/database");
System.out.println("数据：" + new String(data));

// 读取节点数据 + 状态信息
Stat stat = new Stat();
byte[] dataWithStat = client.getData()
        .storingStatIn(stat)
        .forPath("/config/database");
System.out.println("版本号：" + stat.getVersion());

// 更新节点数据
client.setData().forPath("/config/database", "new connection string".getBytes());

// 带版本控制的更新（防止并发覆盖）
int expectedVersion = stat.getVersion();
client.setData()
      .withVersion(expectedVersion)
      .forPath("/config/database", "safe update".getBytes());
// 若节点版本号已变化，此操作将失败

// 删除单个节点（节点必须没有子节点）
client.delete().forPath("/config/database");

// 递归删除（删除节点及其所有子节点）
client.delete().deletingChildrenIfNeeded().forPath("/config");

// 版本控制删除
client.delete().withVersion(0).forPath("/config/database");

// 保证删除（即使失败也会持续重试直到成功）
client.delete().guaranteed().forPath("/config/database");

NodeCache nodeCache = new NodeCache(client, "/config/database");
nodeCache.getListenable().addListener(() -> {
    ChildData currentData = nodeCache.getCurrentData();
    if (currentData != null) {
        System.out.println("节点数据变更：" + new String(currentData.getData()));
    }
});
nodeCache.start(true);  // true表示启动时立即加载缓存数据

PathChildrenCache cache = new PathChildrenCache(client, "/services", true);
cache.getListenable().addListener((client, event) -> {
    switch (event.getType()) {
        case CHILD_ADDED:
            System.out.println("服务上线：" + event.getData().getPath());
            break;
        case CHILD_REMOVED:
            System.out.println("服务下线：" + event.getData().getPath());
            break;
        case CHILD_UPDATED:
            System.out.println("服务更新：" + event.getData().getPath());
            break;
    }
});
cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);

TreeCache treeCache = new TreeCache(client, "/config");
treeCache.getListenable().addListener((client, event) -> {
    System.out.println("事件类型：" + event.getType() + "，路径：" + event.getData().getPath());
});
treeCache.start();

import org.apache.curator.framework.recipes.locks.InterProcessMutex;

public class DistributedLockDemo {
    public static void main(String[] args) throws Exception {
        CuratorFramework client = createClient();
        
        // 创建分布式锁
        InterProcessMutex lock = new InterProcessMutex(client, "/locks/order_lock");
        
        try {
            // 获取锁（阻塞，直到获取成功）
            lock.acquire();
            
            // 临界区：执行需要互斥的业务逻辑
            System.out.println("成功获取锁，执行业务逻辑...");
            // 扣减库存、生成订单等操作
            
        } finally {
            // 释放锁
            lock.release();
        }
        
        client.close();
    }
}

// 尝试获取锁，最多等待10秒，超时则放弃
if (lock.acquire(10, TimeUnit.SECONDS)) {
    try {
        // 业务逻辑
    } finally {
        lock.release();
    }
}

import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;

public class LeaderElectionDemo {
    public static void main(String[] args) {
        CuratorFramework client = createClient();
        String myId = "node-001";
        
        LeaderSelector selector = new LeaderSelector(client, "/leader/election", 
            new LeaderSelectorListener() {
                @Override
                public void takeLeadership(CuratorFramework client) throws Exception {
                    // 成为Leader后的业务逻辑
                    System.out.println(myId + " 已成为Leader！");
                    // 执行只有Leader能做的任务...
                    Thread.sleep(30000);  // 模拟工作
                    // 方法返回即放弃Leader身份
                }
            });
        
        selector.setId(myId);
        selector.autoRequeue();  // 自动重新参与选举
        selector.start();
    }
}

import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.ServiceProvider;

// 服务注册
ServiceInstance<Void> instance = ServiceInstance.<Void>builder()
        .name("user-service")
        .address("192.168.1.100")
        .port(8080)
        .build();

ServiceDiscovery<Void> discovery = ServiceDiscoveryBuilder.builder(Void.class)
        .client(client)
        .basePath("/services")
        .build();

discovery.registerService(instance);
discovery.start();

// 服务发现
ServiceProvider<Void> provider = discovery.serviceProviderBuilder()
        .serviceName("user-service")
        .build();
provider.start();

// 获取可用服务实例
Collection<ServiceInstance<Void>> instances = provider.getAllInstances();
for (ServiceInstance<Void> inst : instances) {
    System.out.println("发现服务：" + inst.getAddress() + ":" + inst.getPort());
}

client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
    @Override
    public void stateChanged(CuratorFramework client, ConnectionState state) {
        switch (state) {
            case CONNECTED:
                System.out.println("连接成功");
                break;
            case SUSPENDED:
                System.out.println("连接暂时中断，操作将被排队");
                break;
            case RECONNECTED:
                System.out.println("连接已恢复");
                break;
            case LOST:
                System.out.println("会话已过期，临时节点已丢失，需要重建");
                break;
        }
    }
});

CuratorFramework client = null;
try {
    client = CuratorFrameworkFactory.builder()
            .connectString(zkAddress)
            .retryPolicy(new ExponentialBackoffRetry(1000, 3))
            .build();
    client.start();
    // 使用 client 进行操作...
} finally {
    if (client != null) {
        client.close();  // 务必关闭，释放资源
    }
}