快速入门Zookeeper

Zookeeper

ZooKeeper作为一个强大的开源分布式协调服务，扮演着分布式系统中至关重要的角色。它提供了一个中心化的服务，用于维护配置信息、命名、提供分布式同步以及提供组服务等。通过其高性能和可靠的特性，ZooKeeper能够确保在复杂的分布式环境中，各个节点和服务之间的协调和通信得以顺利进行。无论是在大规模的Web服务、云基础设施管理，还是大数据处理框架中，ZooKeeper都是实现服务发现、负载均衡、数据发布/订阅等关键功能的首选工具。它的出现极大地简化了分布式系统设计中的一致性问题，使得开发者可以更加专注于业务逻辑的实现。

安装配置

学习笔记：快速入门ZooKeeper技术

zookeeper命令操作

zookeeper数据模型

服务端命令操作
• 启动 ZooKeeper 服务: ./zkServer.sh start
• 查看 ZooKeeper 服务状态: ./zkServer.sh status
• 停止 ZooKeeper 服务: ./zkServer.sh stop
• 重启 ZooKeeper 服务: ./zkServer.sh restart
客户端命令操作

连接ZooKeeper服务端：

./zkCli.sh –server ip:port

断开链接

quit

查看节点信息

ls /dubbo

创建节点并赋值（没有加任何参数就是持久化的）

create /节点path value

设置节点值

set /节点path value

删除节点值

delete /path

不能重复创建节点，但是可以创建子节点往下延申

create /app1/p1
create /app1/p2

子节点存在的时候，不允许直接删除父节点
直接删除子节点的节点

deleteall /节点path

创建临时顺序节点

create -e /节点path value

退出会话，重新打开xshell查询会发现临时节点没有了

创建持久化顺序节点

create -s /节点path value

如果生成多次，会发现顺序产生多个节点

查看创建顺序临时节点

create -es /节点路径

查看节点详细信息

ls -s /节点path(ls2 /)

之前使用过Dubbo，这里可以看一下服务提供方的ip地址

JAVA API操作

注意
因为Curator是ZooKeeper的Java客户端库，所以二者的版本要对应起来可以看下官网的说明。
Zookeeper 3.5X 的版本可以用Curator4.0版本
低版本的Curator不能用高版本的Zookeeper，反之则可以。
导入pom.xml

<dependencies>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.10</version>
            <scope>test</scope>
        </dependency>

        <!--curator-->
        <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>
        <!--日志-->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.21</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.21</version>
        </dependency>

    </dependencies>


    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

建立链接

public class CuratorTest {
    /**
     * 建立连接
     */
    @Test
    public void testConnection() {
        // 1.第一种方式
        // 重试策略，该策略重试设定的次数，每次重试之间的睡眠时间都会增加
        /**
         * 参数：
         * baseSleepTimeMs – 重试之间等待的初始时间量
         * 最大重试 次数 – 重试的最大次数
         */
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);

        /**
         * Create a new client
         *
         * @param connectString       list of servers to connect to 连接字符串，地址+端口 例如“192. 168.149.135:2181, 192.168.149.135”
         * @param sessionTimeoutMs    session timeout 会话超时时间 单位毫秒
         * @param connectionTimeoutMs connection timeout 连接超时时间 单位毫秒
         * @param retryPolicy         retry policy to use 策略
         * @return client
         */

        CuratorFramework client =
                CuratorFrameworkFactory.newClient("192. 168.149.135:2181", 60 * 1000, 15 * 1000, retryPolicy);

        client.start();
    }
}

 /**
     * 建立连接 方式二
     */
    @Test
    public void testConnection2() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);

        // 第二种方式
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().connectString("192. 168.149.135:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("itheima");;
        
        builder.build().start();

    }

注意这里建议把namespace加上，相当于是根目录

添加节点

public class CuratorTest {
	// 声明为成员变量 
    private CuratorFramework client;
    /**
     * 建立连接 方式二
     */
    @Before
    public void testConnection2() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);

        // 第二种方式
        client = CuratorFrameworkFactory.builder().connectString("192. 168.149.135:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("itheima").build();

        client.start();
    }

    /**
     * 创建节点
     * 持久、临时、顺序
     * 设置数据
     * <p>
     * 1.基本创建
     * 2.创建节点，带有数据
     * 3.设置节点的类型
     * 4.创建多级节点
     */
    @Test
    public void testCreate() throws Exception {
        // client操作
    }

    @After
    /**
     * 关闭连接
     */
    public void close() {
        if (null != client) {
            client.close();
        }
    }
}

(1).创建一个基础的节点
我们先创建一个基础的节点
CuratorTest.java

    @Test
    public void testCreate() throws Exception {
        String path = client.create().forPath("/app4");
        System.out.println(path);
    }

(2).创建一个带有数据的节点
CuratorTest.java

    @Test
    public void testCreateValue() throws Exception {
        String path = client.create().forPath("/app5", "heima".getBytes());
        System.out.println(path);
    }

(3).设置节点的类型
CuratorTest.java
创建临时节点

    /**
     * 设置节点类型
     *
     * @throws Exception
     */
    @Test
    public void testCreateType() throws Exception {
        String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app6", "number6".getBytes());
        System.out.println(path);
    }

(4).创建多级节点
CuratorTest.java

 /**
     * 多级节点
     *
     * @throws Exception
     */
    @Test
    public void testCreateManyTypes() throws Exception {
        // creatingParentsIfNeeded()如果父节点不存在，就创建一个节点
        String path = client.create().creatingParentsIfNeeded().forPath("/app8/p8", "number8".getBytes());
        System.out.println(path);
    }

查询节点
(1).查询数据
CuratorTest.java

    /**
     * 获取、查询
     * 1.查询数据
     * 2.查询子节点
     * 3.查询节点的状态 ls -s
     */
    @Test
    public void testGet() throws Exception {
        byte[] bytes = client.getData().forPath("/app7");
        System.out.println(new String(bytes));
    }

(2).查询子节点
CuratorTest.java

    /**
     * 获取、查询
     * 2.查询子节点
     */
    @Test
    public void testGetSon() throws Exception {
        List<String> path = client.getChildren().forPath("/app8");
        System.out.println(path);
    }

(3).查询节点的状态 ls -s
CuratorTest.java

    /**
     * 3.查询节点的状态 ls -s
     */
    @Test
    public void testGetStatus() throws Exception {
        Stat stat = new Stat();
        System.out.println("查询前:" + stat);
        client.getData().storingStatIn(stat).forPath("/app8/p8");
        System.out.println("查询后:" + stat);
    }

修改节点
修改数据
修改前用Zookeeper查询

    /**
     * 修改数据
     *
     * @throws Exception
     */
    @Test
    public void testSet() throws Exception {
        client.setData().forPath("/app7", "dong77".getBytes());
    }

根据版本修改数据⭐(推荐)
CuratorTest.java

    /**
     * 按照版本修改
     *
     * @throws Exception
     */
    @Test
    public void testSetForVersion() throws Exception {
        int version = 0;
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath("/app7");
        version = stat.getVersion();
        System.out.println("当前version是：" + version);
        client.setData().withVersion(version).forPath("/app7", "luka7".getBytes());
    }

删除节点

    /**
     * 1.删除单个节点
     *
     * @throws Exception
     */
    @Test
    public void testDelete() throws Exception {
        client.delete().forPath("/app5");
    }

删除带有子节点的节点
删除前节点app7是有子节点的

    /**
     * 2.删除带有子节点的节点
     *
     * @throws Exception
     */
    @Test
    public void testDeleteWithSon() throws Exception {
        client.delete().deletingChildrenIfNeeded().forPath("/app7");
    }

必须成功的删除节点⭐(推荐)：防止网络抖动，本质抖动。
删除前app4还在

    /**
     * 3.必须成功的删除节点
     *
     * @throws Exception
     */
    @Test
    public void testDeleteSucceed() throws Exception {
        client.delete().guaranteed().forPath("/app4");
    }

回调
删除前有节点

/**
     * 4.回调
     *
     * @throws Exception
     */
    @Test
    public void testDeleteReturnFun() throws Exception {
        BackgroundCallback callback = new BackgroundCallback() {
            @Override
            public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                System.out.println("我被删除了");
                System.out.println(event);
            }
        };
        client.delete().inBackground(callback).forPath("/app8");
    }

Watch事件监听

(1).NodeCache给指定一个节点注册监听器
用代码实现一下：
命名空间下面可能没有子节点的话，过一会儿就会删除命名空间
CuratorTest.java

 /**
     * NodeCache : 只是监听某一个特定的节点
     *
     * @throws Exception
     */
    @Test
    public void testNodeCache() throws Exception {
        while (true) {
            /**
             * Params:
             * client – curztor client 客户端
             * path – the full path to the node to cache  要缓存的节点的完整路径
             * dataIsCompressed – if true, data in the path is compressed 如果为 true，则路径中的数据被压缩,默认不压缩
             */
            // 1.创建NodeCache对象
            NodeCache nodeCache = new NodeCache(client, "/app1");

            // 2.注册监听8
            nodeCache.getListenable().addListener(new NodeCacheListener() {
                @Override
                public void nodeChanged() throws Exception {
                    System.out.println("节点变化了");
                }
            });

            // 3.开启监听 如果为 true，将在此方法返回之前调用， rebuild() 以便获得节点的初始视图
            nodeCache.start(true);
        }

    }

现在想知道节点变成了什么，这里需要修改代码
CuratorTest.java

 /**
     * NodeCache : 只是监听某一个特定的节点
     *
     * @throws Exception
     */
    @Test
    public void testNodeCache() throws Exception {

        /**
         * Params:
         * client – curztor client 客户端
         * path – the full path to the node to cache  要缓存的节点的完整路径
         * dataIsCompressed – if true, data in the path is compressed 如果为 true，则路径中的数据被压缩,默认不压缩
         */
        // 1.创建NodeCache对象
        NodeCache nodeCache = new NodeCache(client, "/app1");

        // 2.注册监听
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("节点变化了");
                // 获取修改节点后的数据
                byte[] data = nodeCache.getCurrentData().getData();
                System.out.println("修改后数据是：" + new String(data));
            }
        });

        // 3.开启监听 如果为 true，将在此方法返回之前调用， rebuild() 以便获得节点的初始视图
        nodeCache.start(true);
        while (true) {
        }


    }

(2).PathChildrenCache(监听某个节点的子节点们 )
测试前先给app1节点创建3个子节点

/**
     * PathChildrenCache : 监控一个ZNode的所有子节点.
     *
     * @throws Exception
     */
    @Test
    public void testPathChildrenCache() throws Exception {
        /**
         *  如果为 true，则除了统计信息之外，还会缓存节点内容
         */
        // 1.创建监听对象
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app1", true);

        // 2.绑定监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                System.out.println("子节点变化了");
                System.out.println(event);
            }
        });
        pathChildrenCache.start();

        while (true) {

        }
    }

那现在我想看到具体子节点的变化情况和变化的值呢。修改CuratorTest.java修改PathChildrenCache构造函数，最后一个值是false,然后输出监听信息。

/**
     * PathChildrenCache : 监控一个ZNode的所有子节点.
     *
     * @throws Exception
     */
    @Test
    public void testPathChildrenCache() throws Exception {
        /**
         *  如果为 true，则除了统计信息之外，还会缓存节点内容
         */
        // 1.创建监听对象
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app1", false);

        // 2.绑定监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                System.out.println("子节点变化了");
                System.out.println(event);

                // 1.获取类型
                PathChildrenCacheEvent.Type type = event.getType();
                // 2.判断类型是否是update
                if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
                    byte[] bytes = event.getData().getData();
                    System.out.println("数据被更新了：" + new String(bytes));
                } else if (type.equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {
                    byte[] bytes = event.getData().getData();
                    System.out.println("数据被添加了：" + new String(bytes));
                } else if (type.equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                    byte[] bytes = event.getData().getData();
                    System.out.println("数据被删除了：" + new String(bytes));
                }

            }
        });
        pathChildrenCache.start();

        while (true) {

        }
    }

(3).TreeCache(监听某个节点自己和所有的子节点们)
CuratorTest.java

/**
     * TreeCache : 可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合
     *
     * @throws Exception
     */
    @Test
    public void testTreeCache() throws Exception {
        TreeCache treeCache = new TreeCache(client, "/app1");
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                System.out.println("整个树节点有变化");
                System.out.println(event);
            }
        });
        treeCache.start();

        while (true) {
            
        }

    }

也想输出树节点的信息，继续修改CuratorTest.java

 /**
     * TreeCache : 可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合
     *
     * @throws Exception
     */
    @Test
    public void testTreeCache() throws Exception {
        TreeCache treeCache = new TreeCache(client, "/app1");
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                System.out.println("整个树节点有变化");
                System.out.println(event);

                TreeCacheEvent.Type type = event.getType();
                if (type.equals(TreeCacheEvent.Type.NODE_UPDATED)) {
                    byte[] bytes = event.getData().getData();
                    System.out.println("树节点数据被更新了：" + new String(bytes));
                } else if (type.equals(TreeCacheEvent.Type.NODE_ADDED)) {
                    byte[] bytes = event.getData().getData();
                    System.out.println("树节点数据被添加了：" + new String(bytes));
                } else if (type.equals(TreeCacheEvent.Type.NODE_REMOVED)) {
                    byte[] bytes = event.getData().getData();
                    System.out.println("树节点数据被删除了：" + new String(bytes));
                } else {
                    System.out.println("树节点无操作");
                }

            }
        });
        treeCache.start();

        while (true) {

        }

    }

分布式锁

原理
当客户端想要获得锁，则创建节点，使用完锁，则删除该节点。

模拟12306售票案例

创建一个测试类LockTest.java

package com.itheima;

/**
 * @ClassName: LockTest
 * @Description:
 * @Author: wty
 * @Date: 2023/3/14
 */

public class LockTest {
    public static void main(String[] args) {
        Tick12306 tick12306 = new Tick12306();
        // 创建客户端
        Thread t1 = new Thread(tick12306, "携程");
        Thread t2 = new Thread(tick12306, "飞猪");
        Thread t3 = new Thread(tick12306, "去哪儿");
        
        t1.start();
        t2.start();
        t3.start();
    }
}

创建实体类模拟12306抢票操作

package com.itheima;

/**
 * @ClassName: Tick12306
 * @Description:
 * @Author: wty
 * @Date: 2023/3/14
 */

public class Tick12306 implements Runnable {
    // 数据库的票数
    private int tickets = 100;

    @Override
    public void run() {
        while (true) {
            if (tickets > 0) {
                System.out.println("线程:" + Thread.currentThread().getName() + "买走了票，剩余" + (--tickets));
            }
        }
    }
}

解决方案：用分布式锁解决
添加工具类ClientConnection.java

public class ClientConnection {
    public static CuratorFramework getConnection() {
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(3000, 10);

        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192. 168.149.135:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000).retryPolicy(retry).build();
        client.start();
        return client;
    }
}

修改Tick12306.java

public class Tick12306 implements Runnable {
    private InterProcessMutex lock;
    // 数据库的票数
    private int tickets = 100;

    public Tick12306() {
        CuratorFramework client = ClientConnection.getConnection();
        lock = new InterProcessMutex(client, "/itheima/lock");
    }

    @Override
    public void run() {
        while (true) {
            // 获取锁
            try {
                lock.acquire(3, TimeUnit.SECONDS);
                if (tickets > 0) {
                    System.out.println("线程:" + Thread.currentThread().getName() + "买走了票，剩余" + (--tickets));
                    Thread.sleep(100);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 释放锁
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

zookeeper 集群

配置每一个Zookeeper 的dataDir 和 clientPort 分别为2181 2182 2183

修改/usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg





模拟集群异常
（1）首先我们先测试如果是从服务器挂掉，会怎么样
把3号服务器停掉，观察1号和2号，发现状态并没有变化。

由此得出结论，3个节点的集群，从服务器挂掉，集群正常
（2）我们再把1号服务器（从服务器）也停掉，查看2号（主服务器）的状态，发现已经停止运行了。

由此得出结论，3个节点的集群，2个从服务器都挂掉，主服务器也无法运行。因为可运行的机器没有超过集群总数量的半数。
（3）我们再次把1号服务器启动起来，发现2号服务器又开始正常工作了。而且依然是领导者。

（4）我们把3号服务器也启动起来，把2号服务器停掉,停掉后观察1号和3号的状态。

（5）我们再次测试，当我们把2号服务器重新启动起来启动后，会发生什么？2号服务器会再次成为新的领导吗？结果2号服务器启动后依然是跟随者（从服务器），3号服务器依然是领导者（主服务器），没有撼动3号服务器的领导地位。由此我们得出结论，当领导者产生后，再次有新服务器加入集群，不会影响到现任领导者。

集群角色