使用Curator来控制ZooKeeper

ゞ浴缸里的玫瑰 2022-12-05 10:25 149阅读 0赞

# 使用Curator来控制ZooKeeper #

## 前言 ##

作为一个了解分布式架构的开发者，一定会用到`ZooKeeper`,它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。虽然`ZooKeeper`功能强大，但是`ZooKeeper`客户端的调用比较复杂，对我们开发者相对而言还是不够友好的，而`Apache Curator` 就是为了简化`ZooKeeper`客户端调用而生，利用它，可以更好的使用`ZooKeeper`。

## 正文 ##

## Apache Curator ##

**Apache Curator**是一个比较完善的`ZooKeeper`客户端框架，通过封装的一套高级`API` 简化了`ZooKeeper`的操作。通过查看官方文档，可以发现`Curator`主要解决了三类问题：

1.  封装`ZooKeeper client`与`ZooKeeper server`之间的连接处理。
2.  提供了一套`Fluent`风格的操作`API`。
3.  提供`ZooKeeper`各种应用场景(`recipe`， 比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装。

`Curator`主要从以下几个方面降低了`zk`使用的复杂性：

*  **重试机制**:提供可插拔的重试机制, 它将给捕获所有可恢复的异常配置一个重试策略，并且内部也提供了几种标准的重试策略(比如指数补偿)
 *  **连接状态监控**: `Curator`初始化之后会一直对zk连接进行监听，一旦发现连接状态发生变化将会作出相应的处理
 *  **zk客户端实例管理**:`Curator`会对`zk`客户端到`server`集群的连接进行管理，并在需要的时候重建`zk`实例，保证与`zk`集群连接的可靠性
 *  **各种使用场景支持**:`Curator`实现了`zk`支持的大部分使用场景（甚至包括`zk`自身不支持的场景），这些实现都遵循了`zk`的最佳实践，并考虑了各种极端情况

**Curator的组件**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center]

**Maven的依赖**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 1]

我们在实际的应用时，最常用的是`curator-recipes`，它可以实现：

*  **锁**：包括共享锁、共享可重入锁、读写锁等。
 *  **选举**：`Leader`选举算法。
 *  **Barrier**：阻止分布式计算直至某个条件被满足的“栅栏”，可以看做`JDK Concurrent`包中`Barrier`的分布式实现。
 *  **缓存**：三种`Cache`及监听机制。
 *  **持久化结点**：连接或`Session`终止后仍然在`ZooKeeper`中存在的结点。
 *  **队列**：分布式队列、分布式优先级队列等。

**Maven引入`curator-recipes`的依赖**

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

三种`Cache`及监听机制，值得是`Curator`提供了三种`Watcher`(`Cache`)来监听结点的变化：

*  **Path Cache**：监视一个路径下1）孩子结点的创建、2）删除，3）以及结点数据的更新。产生的事件会传递给注册的`PathChildrenCacheListener`。
 *  **Node Cache**：监视一个结点的创建、更新、删除，并将结点的数据缓存在本地。
 *  **Tree Cache**：`Path Cache`和`Node Cache`的“合体”，监视路径下的创建、更新、删除事件，并缓存路径下所有孩子结点的数据。

我们这里以最常使用的分布式锁的实现为例，来简单感受`Curator`的魅力。

## Curator实现分布式锁 ##

**InterProcessMutexDemo.class：一个分布式锁的简单实现**

*  `Thread.currentThread().sleep(5000L)`这里休眠5秒中，但不释放资源，下一个其他线程必须要在该线程后释放资源后，才能`lock.acquire()`，即获取分布式锁
 *  同时针对这个分布式锁简单实现引入依赖时要注意`curator`和`ZooKeeper`版本关系，否则可能会报错。

public class InterProcessMutexDemo { 
    
    
        public static void main(String[] args) { 
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
            final CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").sessionTimeoutMs(5000).connectionTimeoutMs(10000).retryPolicy(retryPolicy).namespace("text").build();
            client.start();
    
    
            InterProcessMutex lock = new InterProcessMutex(client,"/lock");
            try { 
                lock.acquire();
                System.err.println("生成订单号");
                Thread.currentThread().sleep(5000L);
            } catch (Exception e) { 
            } finally { 
                try { 
                    lock.release();
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
    
    
            try { 
                lock.acquire();
                System.err.println("生成订单号");
                Thread.currentThread().sleep(Long.MAX_VALUE);
            } catch (Exception e) { 
            } finally { 
                try { 
                    lock.release();
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
        }
    }

**控制台输出一下结果：**

![在这里插入图片描述][20200909233821280.png_pic_center]

## Curator实现分布式数据管理 ##

**CuratorUtils：`Curator`工具类**

public class CuratorUtils { 
    
    
        private static String connectString = "127.0.0.1:2181";
    
    
        /** * 创建节点 * @param path * @param value * @throws Exception */
        public static void createNode(String path,Object value) throws Exception { 
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(2000, 30000);
            CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
            curatorFramework.start();
            curatorFramework.create().
                    creatingParentsIfNeeded().//递归创建,如果没有父节点,自动创建父节点
                    withMode(CreateMode.PERSISTENT).//设置节点类型
                    withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).
                    forPath(path, FastJsonUtils.convertObjectToJSON(value).getBytes());
            curatorFramework.close();
        }
    
        /** * 获取节点 * @param path * @param clazz * @return * @throws Exception */
        public static Object getNode(String path, Class<?> clazz) throws Exception { 
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(2000, 30000);
            CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
            curatorFramework.start();
    
            byte[] data = curatorFramework.
                    getData().
                    forPath(path);
    
            curatorFramework.getData().usingWatcher(new Watcher() { 
                public void process(WatchedEvent watchedEvent) { 
                    System.out.println("watch机制触发" + watchedEvent);
                }
            });
            curatorFramework.close();
            return FastJsonUtils.convertJsonToObject(new String(data),clazz);
        }
    
    
        /** * 设置结点 * @param path * @param value * @throws Exception */
        public static void SetNode(String path,Object value) throws Exception { 
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(2000, 30000);
            CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
            curatorFramework.start();
            Stat stat = curatorFramework.setData().
                    withVersion(-1).
                    forPath(path, FastJsonUtils.convertObjectToJSON(value).getBytes());
            curatorFramework.close();
        }
    
    
        /** * 删除结点 * @param path * @throws Exception */
        public static void deleteNode(String path) throws Exception { 
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(2000, 30000);
            CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
            curatorFramework.start();
            curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
            curatorFramework.close();
        }
    
    
        public static void main(String[] args) { 
            try { 
                String path ="/services";
                ResponseResult responseResult =ResponseResult.success("haha");
                //创建结点
                createNode(path,responseResult);
                //获取结点
                ResponseResult result =(ResponseResult) getNode(path,ResponseResult.class);
                System.out.println(result);
    
            }catch (Exception e){ 
                System.out.println("zookeeper操作异常"+e.getMessage());
            }
    
        }
    
    }

**获取结点中的数据，控制台输出：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 2]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/bba19d90233b4a2d9cc8034348ce8db0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/76ce5c744112422d838e57590946af16.png
[20200909233821280.png_pic_center]: /images/20221123/40c3751261fd4954a40fdc563193a9d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/0d1168e5e27044e98415d3a3bee967cb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/a43e401880d74f6ead26aaf14320e596.png