java API来操作zookeeper

我不是女神ヾ 2022-05-26 04:28 159阅读 0赞

package com.irisian.zookeeper;  
  
import java.io.IOException;  
import java.util.concurrent.CountDownLatch;  
  
import org.apache.zookeeper.AsyncCallback;  
import org.apache.zookeeper.CreateMode;  
import org.apache.zookeeper.KeeperException;  
import org.apache.zookeeper.WatchedEvent;  
import org.apache.zookeeper.Watcher;  
import org.apache.zookeeper.Watcher.Event.EventType;  
import org.apache.zookeeper.Watcher.Event.KeeperState;  
import org.apache.zookeeper.ZooDefs.Ids;  
import org.apache.zookeeper.ZooKeeper;  
import org.apache.zookeeper.data.Stat;  
import org.junit.After;  
import org.junit.Before;  
import org.junit.Test;  
  
public class ZookeeperTest \{  
    private ZooKeeper zk = null;  
    /\*\* zookeeper地址 \*/  
    private String CONNECT\_ADDR = "irisian01:2181,irisian01:2181,irisian01:2181";  
    /\*\* session超时时间 \*/  
    private int SESSION\_OUTTIME = 2000;// ms  
    /\*\* 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 \*/  
    private CountDownLatch connectedSemaphore = new CountDownLatch(1);  
  
    @Before  
    public void init() throws IOException, InterruptedException \{  
        zk = new ZooKeeper(CONNECT\_ADDR, SESSION\_OUTTIME, new Watcher() \{  
            public void process(WatchedEvent event) \{  
                // 获取事件的状态  
                KeeperState keeperState = event.getState();  
                EventType eventType = event.getType();  
                // 如果是建立连接  
                if (KeeperState.SyncConnected == keeperState) \{  
                    if (EventType.None == eventType) \{  
                        // 如果建立连接成功，则发送信号量，让后续阻塞程序向下执行  
                        connectedSemaphore.countDown();  
                        System.out.println("zk 建立连接");  
                    \}  
                \}  
            \}  
        \});  
        // 进行阻塞  
        connectedSemaphore.await();  
    \}  
  
    @Test  
    public void add() throws KeeperException, InterruptedException \{  
        zk.create("/name", "wumanxin".getBytes(), Ids.OPEN\_ACL\_UNSAFE, CreateMode.PERSISTENT);  
    \}  
  
    @Test  
    public void get() throws Exception \{  
        byte\[\] data = zk.getData("/name", false, null);  
        System.out.println(new String(data));  
    \}  
  
    @Test  
    public void mod() throws KeeperException, InterruptedException \{  
        zk.setData("/name", "wumanxin2018".getBytes(), -1);  
    \}  
  
    // 判断节点是否存在  
    @Test  
    public void isExist() throws KeeperException, InterruptedException \{  
        Stat exists = zk.exists("/name", false);  
        if (exists != null) \{  
            System.out.println("name存在");  
        \} else \{  
            System.out.println("name不存在");  
        \}  
    \}  
  
    // 同步删除  
    @Test  
    public void del() throws InterruptedException, KeeperException \{  
        zk.delete("/name", -1);  
    \}  
  
    // 异步删除  
    @Test  
    public void asyDel() \{  
        // 异步删除节点  
        zk.delete("/name", -1, new AsyncCallback.VoidCallback() \{  
            public void processResult(int rc, String path, Object ctx) \{  
                System.out.println("rc=====" + rc);  
                System.out.println("path======" + path);  
                System.out.println("ctc======" + path);  
            \}  
        \}, "回调值");  
  
    \}  
  
    @After  
    public void close() throws InterruptedException \{  
        zk.close();  
    \}  
\}