version: "3.1"
services:
  zk1:
    image: zookeeper
    restart: always
    container_name: zk1
    ports:
      - 2181:2181
    environment:
      ZOO_MY_ID: 1
      ZOO_SERVERS: server.1=zk1:2888:3888;2181 server.2=zk2:2888:3888;2181 server.3=zk3:2888:3888;2181
  zk2:
    image: zookeeper
    restart: always
    container_name: zk2
    ports:
      - 2182:2181
    environment:
      ZOO_MY_ID: 2
      ZOO_SERVERS: server.1=zk1:2888:3888;2181 server.2=zk2:2888:3888;2181 server.3=zk3:2888:3888;2181
  zk3:
    image: zookeeper
    restart: always
    container_name: zk3
    ports:
      - 2183:2181
    environment:
      ZOO_MY_ID: 3
      ZOO_SERVERS: server.1=zk1:2888:3888;2181 server.2=zk2:2888:3888;2181 server.3=zk3:2888:3888;2181

• 启动zookeeper集群，并分别查看三个容器的角色，在容器的bin目录下执行 ./zkServer.sh status 进行查看

• my_id谁大就谁做leader的前提是，没有leader的情况下会选大的做leader
  把zk3宕机，则会选举zk2作为新的leader
  给zk2创建新的节点，则zk1可以同步到新创建的节点信息
  把zk2宕机了，则整个集群就瘫痪了
  重启zk3，因为zk1中有数据，所以zxid比zk3大，因此选zk1作为leader

7.java操作Zookeeper

7.1：java连接Zookeeper

• 创建Maven工程
• 导入依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.6.0</version>
    </dependency>

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

    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
</dependencies>

public class ZkUtil {
    public static CuratorFramework cf(){
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,2);
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                .connectString("192.168.206.142:2181,192.168.206.142:2182,192.168.206.142:2183")
                .retryPolicy(retryPolicy)
                .build();
        cf.start();
        return cf;
    }
}

public class Demo2 {
    CuratorFramework cf = ZkUtil.cf();
    // 获取子节点
    @Test
    public void getChildren() throws Exception {
        List<String> strings = cf.getChildren().forPath("/");
        for (String string : strings) {
            System.out.println(string);
        }
    }
    // 获取节点数据
    @Test
    public void getData() throws Exception {
        byte[] bytes = cf.getData().forPath("/qf");
        System.out.println(new String(bytes,"UTF-8"));
    }
}

@Test
public void create() throws Exception {
    cf.create().withMode(CreateMode.PERSISTENT).forPath("/qf2","uuuu".getBytes());
}

@Test
public void update() throws Exception {
    cf.setData().forPath("/qf2","oooo".getBytes());
}

@Test
public void delete() throws Exception {
    cf.delete().deletingChildrenIfNeeded().forPath("/qf2");
}

@Test
public void stat() throws Exception {
    Stat stat = cf.checkExists().forPath("/qf");
    System.out.println(stat);
}

public class Demo3 {
    CuratorFramework cf = ZkUtil.cf();
    @Test
    public void listen() throws Exception {
        //1. 创建NodeCache对象，指定要监听的znode
        NodeCache nodeCache = new NodeCache(cf,"/qf");
        nodeCache.start();
        //2. 添加一个监听器
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                byte[] data = nodeCache.getCurrentData().getData();
                Stat stat = nodeCache.getCurrentData().getStat();
                String path = nodeCache.getCurrentData().getPath();
                System.out.println("监听的节点是：" + path);
                System.out.println("节点现在的数据是：" + new String(data,"UTF-8"));
                System.out.println("节点状态是：" + stat);
            }
        });
        System.out.println("开始监听！！");
        //3. System.in.read();
        System.in.read();
    }
}