Zookeeper集成java

2025-02-12

源生API

//Maven配置
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.6</version>
</dependency>

private static final int SESSION_TIMEOUT = 10000;
private static final String CONNECTION_STRING = "127.0.0.1:2181";
private static final String ZK_PATH = "/leader";
private ZooKeeper zk = null;

private CountDownLatch connectedSemaphore = new CountDownLatch(1);

/**
 * 创建ZK连接
 *
 * @param connectString  ZK服务器地址列表
 * @param sessionTimeout Session超时时间
 */
public void createConnection(String connectString, int sessionTimeout) {
    this.releaseConnection();
    try {
        // this代表有监听
        zk = new ZooKeeper(connectString, sessionTimeout, this);
        connectedSemaphore.await();
    } catch (InterruptedException e) {
        System.out.println("连接创建失败，发生 InterruptedException");
        e.printStackTrace();
    } catch (IOException e) {
        System.out.println("连接创建失败，发生 IOException");
        e.printStackTrace();
    }
}

监听：

// 实现Watcher接口
public class ZooKeeperWatcher implements Watcher  {
    @Override
    public void process(WatchedEvent watchedEvent) {
        
    }
}

ZKClient

//Maven配置
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>

连接：

1	ZkClient zkc = new ZkClient(new ZkConnection(CONNECT_ADDR), SESSION_OUTTIME);

监听：

public void testZkClientWatcher2() throws Exception {
       ZkClient zkc = new ZkClient(new ZkConnection(CONNECT_ADDR), SESSION_OUTTIME);

       zkc.createPersistent("/super", "1234");

       //对父节点添加监听子节点变化。
       zkc.subscribeChildChanges("/super", new IZkChildListener() {
           @Override
           public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
               System.out.println("parentPath: " + parentPath);
               System.out.println("currentChilds: " + currentChilds);
           }
       });

       //对父节点添加监听子节点内容变化。
       zkc.subscribeDataChanges("/super", new IZkDataListener() {
           @Override
           public void handleDataDeleted(String path) throws Exception {
               System.out.println("删除的节点为:" + path);
           }

           @Override
           public void handleDataChange(String path, Object data) throws Exception {
               System.out.println("变更的节点为:" + path + ", 变更内容为:" + data);
           }
       });

       Thread.sleep(3000);
       zkc.writeData("/super", "456", -1);
       Thread.sleep(1000);

       zkc.delete("/super");
       Thread.sleep(Integer.MAX_VALUE);

   }

Curator

//Maven配置
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>4.0.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.0.0</version>
</dependency>

启动：

public void init(){
    //创建 CuratorFrameworkImpl实例
    client = CuratorFrameworkFactory.newClient(SERVER, SESSION_TIMEOUT, CONNECTION_TIMEOUT, retryPolicy);

    //启动
    client.start();
}

  public void setListenterTwoOne() throws Exception{
    ExecutorService pool = Executors.newCachedThreadPool();
    PathChildrenCache childrenCache = new PathChildrenCache(client, "/test", true);
    PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
            System.out.println("开始进行事件分析:-----");
            ChildData data = event.getData();
            switch (event.getType()) {
                case CHILD_ADDED:
                    System.out.println("CHILD_ADDED : "+ data.getPath() +"  数据:"+ data.getData());
                    break;
                case CHILD_REMOVED:
                    System.out.println("CHILD_REMOVED : "+ data.getPath() +"  数据:"+ data.getData());
                    break;
                case CHILD_UPDATED:
                    System.out.println("CHILD_UPDATED : "+ data.getPath() +"  数据:"+ data.getData());
                    break;
                case INITIALIZED:
                    System.out.println("INITIALIZED : "+ data.getPath() +"  数据:"+ data.getData());
                    break;
                default:
                    break;
            }
        }
    };
    childrenCache.getListenable().addListener(childrenCacheListener);
    System.out.println("Register zk watcher successfully!");
    childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

    //创建一个节点
    client.create().orSetData().withMode(CreateMode.PERSISTENT).forPath("/test","test".getBytes());

    client.create().orSetData().withMode(CreateMode.EPHEMERAL).forPath("/test/node01","enjoy".getBytes());
    Thread.sleep(1000);
    client.create().orSetData().withMode(CreateMode.EPHEMERAL).forPath("/test/node02","deer".getBytes());
    Thread.sleep(1000);
    client.create().orSetData().withMode(CreateMode.EPHEMERAL).forPath("/test/node02","demo".getBytes());
    Thread.sleep(1000);
    client.delete().forPath("/test/node02");
    Thread.sleep(10000);
}