ZooKeeper 4. Zookeeper实战（开发重点）

4.1 分布式安装部署

1. 新建文件夹/opt/zk_cluster

2. 将zookeeper压缩包加压到文件夹下，并重命名为cluster1

   tar -zxvf apache-zookeeper-3.5.6-bin.tar.gz -C ./zk_cluster/

3. 在cluster1里新建文件夹zkData

4. 在zkData文件夹下新建文件myid，内容为1

   echo 1 >> cluster1/zkData/myid

5. 复制conf目录下的zoo_sample.cfg文件为zoo.cfg

   cp zoo_sample.cfg zoo.cfg

6. 修改zoo.cfg的内容

   dataDir=/opt/zk_cluster/cluster1/zkData
   clientPort=2187
   ###########Cluster##################
   server.1=192.168.181.128:12887:13887
   server.2=192.168.181.128:12888:13888
   server.3=192.168.181.128:12889:13889

7. 将cluster1复制为cluster2、cluster3

   cp -r cluster1/ cluster2
   cp -r cluster1/ cluster3

8. 将cluster2和cluster3中的zkData/myid内的内容修改为2、3

9. 修改cluster2和cluster3中的zoo.cfg的内容

10. 验证

    启动三个ZooKeeper，查看状态

      ./cluster1/bin/zkServer.sh start
      ./cluster1/bin/zkServer.sh status

    [root@hwjsLinux zk_cluster]# ./cluster1/bin/zkServer.sh status
    /usr/bin/java
    ZooKeeper JMX enabled by default
    Using config: /opt/zk_cluster/cluster1/bin/../conf/zoo.cfg
    Client port found: 2187. Client address: localhost.
    Mode: follower

4.2 客户端命令行操作

1. 启动客户端

   ./zkCli.sh -server localhost:2187

2. 显示所有操作命令

   help

3. 查看当前znode中所包含的内容

   [zk: localhost:2181(CONNECTED) 0] ls /
   [zookeeper]

4. 查看当前节点详细数据

   [zk: localhost:2181(CONNECTED) 1] ls2 /
   [zookeeper]
   cZxid = 0x0
   ctime = Thu Jan 01 08:00:00 CST 1970
   mZxid = 0x0
   mtime = Thu Jan 01 08:00:00 CST 1970
   pZxid = 0x0
   cversion = -1
   dataVersion = 0
   aclVersion = 0
   ephemeralOwner = 0x0
   dataLength = 0
   numChildren = 1

5. 分别创建2个普通节点

   [zk: localhost:2181(CONNECTED) 3] create /sanguo "jinlian"
   Created /sanguo
   [zk: localhost:2181(CONNECTED) 4] create /sanguo/shuguo "liubei"
   Created /sanguo/shuguo

6. 获得节点的值

   [zk: localhost:2181(CONNECTED) 5] get /sanguo
   jinlian
   cZxid = 0x100000003
   ctime = Wed Aug 29 00:03:23 CST 2018
   mZxid = 0x100000003
   mtime = Wed Aug 29 00:03:23 CST 2018
   pZxid = 0x100000004
   cversion = 1
   dataVersion = 0
   aclVersion = 0
   ephemeralOwner = 0x0
   dataLength = 7
   numChildren = 1
   [zk: localhost:2181(CONNECTED) 6] get /sanguo/shuguo
   liubei
   cZxid = 0x100000004
   ctime = Wed Aug 29 00:04:35 CST 2018
   mZxid = 0x100000004
   mtime = Wed Aug 29 00:04:35 CST 2018
   pZxid = 0x100000004
   cversion = 0
   dataVersion = 0
   aclVersion = 0
   ephemeralOwner = 0x0
   dataLength = 6
   numChildren = 0

7. 创建短暂节点

   [zk: localhost:2181(CONNECTED) 7] create -e /sanguo/wuguo "zhouyu"
   Created /sanguo/wuguo

   （1）在当前客户端是能查看到的

   [zk: localhost:2181(CONNECTED) 3] ls /sanguo 
   [wuguo, shuguo]

   （2）退出当前客户端然后再重启客户端

   [zk: localhost:2181(CONNECTED) 12] quit
   [atguigu@hadoop104 zookeeper-3.4.10]$ bin/zkCli.sh

   （3）再次查看根目录下短暂节点已经删除

   [zk: localhost:2181(CONNECTED) 0] ls /sanguo
   [shuguo]

8. 创建带序号的节点

? （1）先创建一个普通的根节点/sanguo/weiguo

[zk: localhost:2181(CONNECTED) 1] create /sanguo/weiguo "caocao"
Created /sanguo/weiguo

? （2）创建带序号的节点

[zk: localhost:2181(CONNECTED) 2] create -s /sanguo/weiguo/xiaoqiao "jinlian"
Created /sanguo/weiguo/xiaoqiao0000000000
[zk: localhost:2181(CONNECTED) 3] create -s /sanguo/weiguo/daqiao "jinlian"
Created /sanguo/weiguo/daqiao0000000001
[zk: localhost:2181(CONNECTED) 4] create -s /sanguo/weiguo/diaocan "jinlian"
Created /sanguo/weiguo/diaocan0000000002

? 如果原来没有序号节点，序号从0开始依次递增。如果原节点下已有2个节点，则再排序时从2开始，以此类推。

1. 修改节点数据值

   [zk: localhost:2181(CONNECTED) 6] set /sanguo/weiguo "simayi"

2. 节点的值变化监听

? （1）在hadoop104主机上注册监听/sanguo节点数据变化

[zk: localhost:2181(CONNECTED) 26] [zk: localhost:2181(CONNECTED) 8] get /sanguo watch

? （2）在hadoop103主机上修改/sanguo节点的数据

[zk: localhost:2181(CONNECTED) 1] set /sanguo "xisi"

? （3）观察hadoop104主机收到数据变化的监听

WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/sanguo

1. 节点的子节点变化监听（路径变化）

? （1）在hadoop104主机上注册监听/sanguo节点的子节点变化

[zk: localhost:2181(CONNECTED) 1] ls /sanguo watch
[aa0000000001, server101]

? （2）在hadoop103主机/sanguo节点上创建子节点

[zk: localhost:2181(CONNECTED) 2] create /sanguo/jin "simayi"
Created /sanguo/jin

? （3）观察hadoop104主机收到子节点变化的监听

WATCHER::
WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/sanguo

1. 删除节点

[zk: localhost:2181(CONNECTED) 4] delete /sanguo/jin

1. 递归删除节点

[zk: localhost:2181(CONNECTED) 15] rmr /sanguo/shuguo

1. 查看节点状态

[zk: localhost:2181(CONNECTED) 17] stat /sanguo
cZxid = 0x100000003
ctime = Wed Aug 29 00:03:23 CST 2018
mZxid = 0x100000011
mtime = Wed Aug 29 00:21:23 CST 2018
pZxid = 0x100000014
cversion = 9
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 1

4.3 API应用

1. pom.xml

        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.10</version>
        </dependency>

2. 测试代码

   public class TestZookeeper {
   
    // private String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private String connectString="192.168.181.128:2187,192.168.181.128:2188,192.168.181.128:2189";
    private int sessionTimeout = 2000;
    private ZooKeeper zkClient;
   
    @Before
    public void init() throws IOException{
   
        zkClient = new ZooKeeper(connectString, sessionTimeout , new Watcher() {
   
            @Override
            public void process(WatchedEvent event) {
   
                System.out.println("---------start----------");
                List<String> children;
                try {
                    children = zkClient.getChildren("/", true);
   
                    for (String child : children) {
                        System.out.println(child);
                    }
                    System.out.println("---------end----------");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
   
    // 1 创建节点
    @Test
    public void createNode() throws KeeperException, InterruptedException{
   
        String path = zkClient.create("/atguigu", "dahaigezuishuai".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
   
        System.out.println(path);
   
    }
   
    // 2 获取子节点 并监控节点的变化
    @Test
    public void getDataAndWatch() throws KeeperException, InterruptedException{
   
        List<String> children = zkClient.getChildren("/", true);
   
        for (String child : children) {
            System.out.println(child);
        }
   
        Thread.sleep(Long.MAX_VALUE);
    }
   
    // 3 判断节点是否存在
    @Test
    public void exist() throws KeeperException, InterruptedException{
        Stat stat = zkClient.exists("/atguigu", false);
        System.out.println(stat==null? "not exist":"exist");
    }
   }

4.4 监听服务器节点动态上下线案例

1. 需求

   某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。

2. 具体实现

   服务端代码

   public class DistributeServer {
   
    public static void main(String[] args) throws Exception {
        String hostname = "test4";
   
        DistributeServer server = new DistributeServer();
   
        // 1 连接zookeeper集群
        server.getConnect();
   
        // 2 注册节点
        server.regist(hostname);
   
        // 3 业务逻辑处理
        server.business();
    }
   
    private void business() throws InterruptedException {
   
        Thread.sleep(Long.MAX_VALUE);
    }
   
    private void regist(String hostname) throws KeeperException, InterruptedException {
   
        String path = zkClient.create("/servers/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
   
        System.out.println(hostname +" is online ");
   
    }
   
    // private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private String connectString="192.168.181.128:2187,192.168.181.128:2188,192.168.181.128:2189";
    private int sessionTimeout = 2000;
    private ZooKeeper zkClient;
   
    private void getConnect() throws IOException {
   
        zkClient = new ZooKeeper(connectString , sessionTimeout , new Watcher() {
   
            @Override
            public void process(WatchedEvent event) {
                // TODO Auto-generated method stub
   
            }
        });
    }
   }

   客户端代码

   public class DistributeClient {
   
    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
   
        DistributeClient client = new DistributeClient();
   
        // 1 获取zookeeper集群连接
        client.getConnect();
   
        // 2 注册监听
        client.getChlidren();
   
        // 3 业务逻辑处理
        client.business();
   
    }
   
    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }
   
    private void getChlidren() throws KeeperException, InterruptedException {
   
        List<String> children = zkClient.getChildren("/servers", true);
   
        // 存储服务器节点主机名称集合
        ArrayList<String> hosts = new ArrayList<>();
   
        for (String child : children) {
   
            byte[] data = zkClient.getData("/servers/"+child, false, null);
   
            hosts.add(new String(data));
        }
   
        // 将所有在线主机名称打印到控制台
        System.out.println(hosts);
   
    }
   
    // private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private String connectString="192.168.181.128:2187,192.168.181.128:2188,192.168.181.128:2189";
    private int sessionTimeout = 2000;
    private ZooKeeper zkClient;
   
    private void getConnect() throws IOException {
   
        zkClient = new ZooKeeper(connectString , sessionTimeout , new Watcher() {
   
            @Override
            public void process(WatchedEvent event) {
   
                try {
                    getChlidren();
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
   
    }
   }

3. 测试

   客户端随着服务端的启动、停止动态获得服务器信息。

20200209 ZooKeeper 4. Zookeeper实战（开发重点）

原文：https://www.cnblogs.com/huangwenjie/p/12287578.html