ZooKeeper快速入门

clientPort=2181
dataDir=/data
dataLogDir=/datalog
tickTime=2000
initLimit=5
syncLimit=2
//3088用于Zookeeper选举, 2088用于Leader和Follower或Observer交换数据
server.1=127.0.xx.1:2088:3088
server.2=127.0.xx.1:2089:3089
server.3=127.0.xx.1:2090:3090

docker pull zookeeper:3.4.10
docker run --name zookeeper01 -p 2181:2181 -p 2888:2888 -p 3888:3888 -v $PWD/zookeeper01/data:/data  -v $PWD/zookeeper01/datalog:/datalog -d zookeeper:3.4.10 
docker run --name zookeeper02 -p 2182:2181 -p 2889:2888 -p 3889:3888 -v $PWD/zookeeper02/data:/data  -v $PWD/zookeeper02/datalog:/datalog -d zookeeper:3.4.10 
docker run --name zookeeper03 -p 2183:2181 -p 2890:2888 -p 3890:3888 -v $PWD/zookeeper03/data:/data  -v $PWD/zookeeper02/datalog:/datalog -d zookeeper:3.4.10 
//通过bash命令
docker exec -ti zookeeper01 /bin/bash
docker exec -ti zookeeper02 /bin/bash
docker exec -ti zookeeper03 /bin/bash

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.6</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.4.2</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.4.2</version>
</dependency>

public class SessionNew implements Watcher {
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static ZooKeeper zookeeper = null;
    private static Stat stat = new Stat();

    public static void main(String[] args) throws IOException {

        try {
            // 1.创建会话
            zookeeper = new ZooKeeper("xx.xx.xx.xx:2181", 5000, new SessionNew());
            System.out.println(zookeeper.getState());
            connectedSemaphore.await();
            // 2.创建节点
            String path1 = zookeeper.create("/zk-test-ephemeral-", "123".getBytes(), Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL);
            System.out.println(path1);
            String path2 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(path2);
            /*          String path1sub1 = zookeeper.create(path1+"/sub1", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
                                CreateMode.EPHEMERAL_SEQUENTIAL);
                        String path1sub2 = zookeeper.create(path1+"/sub2", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
                                CreateMode.EPHEMERAL_SEQUENTIAL);
                        List<String> childList = zookeeper.getChildren(path1, true);
                        System.out.println(childList);*/
            System.out.println(new String(zookeeper.getData(path1, true, stat)));

            zookeeper.setData(path1, "testData01".getBytes(), -1);
            // zookeeper.getData(path, watch, stat);
        } catch (Exception e) {
            System.out.println("连接已经建立");
            e.printStackTrace();
        }

    }

    @Override
    public void process(WatchedEvent event) {
        System.out.println("Receive watched event: " + event);
        if (KeeperState.SyncConnected == event.getState()) {
            if (EventType.None == event.getType() && null == event.getPath()) {
                connectedSemaphore.countDown();
            } else if (EventType.NodeDataChanged == event.getType()) {
                try {
                    System.out.println(new String(zookeeper.getData(event.getPath(), true, stat)));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

    public static void main(String[] args) throws Exception {
        // 1.创建会话
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // CuratorFramework client = CuratorFrameworkFactory.newClient(url01,
        // 5000, 3000, retryPolicy);
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString(url01).sessionTimeoutMs(5000)
                .namespace("firstDemo").retryPolicy(retryPolicy).build();
        client.start();
        // 2.创建节点并设置初始值
        String path = "/zk-book/c1";
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, "init".getBytes());
        // 3.读取数据
        Stat stat = new Stat();
        byte[] data = client.getData().storingStatIn(stat).forPath(path);
        System.out.println(String.format("第一次的版本: %s", stat.getVersion()));
        System.out.println(new String(data, "UTF-8"));
        // 4.更新数据
        client.setData().withVersion(stat.getVersion()).forPath(path, "second".getBytes());
        // data = client.getData().storingStatIn(stat).forPath(path);
        data = client.getData().forPath(path);
        System.out.println(String.format("第2次的版本: %s", stat.getVersion()));
        System.out.println(new String(data, "UTF-8"));
        try {
            client.setData().withVersion(stat.getVersion()).forPath(path, "third".getBytes());
        } catch (Exception ex) {
            System.out.println("更新失败，乐观锁！");
        }

        // 5.删除数据
        client.delete().deletingChildrenIfNeeded().withVersion(stat.getVersion()).forPath(path);
        Thread.sleep(Integer.MAX_VALUE);
    }