Curator（ZoooKeeper客户端）使用详解

时间  2019-11-17
标签 curator zoookeeper 客户端 使用 详解 繁體版
原文   原文链接

更多精彩文章。html

《微服务不是所有，只是特定领域的子集》java

《“分库分表" ？选型和流程要慎重，不然会失控》node

这么多监控组件，总有一款适合你apache

《Linux生产环境上，最经常使用的一套“vim“技巧》vim

《使用Netty，咱们到底在开发些什么？》后端

最有用系列：api

《Linux生产环境上，最经常使用的一套“vim“技巧》缓存

《Linux生产环境上，最经常使用的一套“Sed“技巧》bash

《Linux生产环境上，最经常使用的一套“AWK“技巧》服务器

欢迎Linux和java后端的同窗关注公众号。

注：该文档主要是基于官方文档的说明，具体可查看：curator.apache.org/index.html

Curator是netflix公司开源的一套zookeeper客户端，目前是Apache的顶级项目。与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。Curator解决了不少zookeeper客户端很是底层的细节开发工做，包括链接重连、反复注册wathcer和NodeExistsException 异常等。

Curator由一系列的模块构成，对于通常开发者而言，经常使用的是curator-framework和curator-recipes，下面对此依次介绍。

1.maven依赖问题

最新版本的curator 4.0支持zookeeper 3.4.x和3.5，可是须要注意curator传递进来的依赖，须要和实际服务器端使用的版本相符，以咱们目前使用的zookeeper 3.4.6为例

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>4.0.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.0.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.6</version>
</dependency>
复制代码

2.curator-framework

public static CuratorFramework getClient() {
    return CuratorFrameworkFactory.builder()
            .connectString("127.0.0.1:2181")
            .retryPolicy(new ExponentialBackoffRetry(1000, 3))
            .connectionTimeoutMs(15 * 1000) //链接超时时间，默认15秒
            .sessionTimeoutMs(60 * 1000) //会话超时时间，默认60秒
            .namespace("arch") //设置命名空间
            .build();
}
 
public static void create(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
    client.create().creatingParentsIfNeeded().forPath(path, payload);
}
 
public static void createEphemeral(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
    client.create().withMode(CreateMode.EPHEMERAL).forPath(path, payload);
}
 
public static String createEphemeralSequential(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
    return client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, payload);
}
 
public static void setData(final CuratorFramework client, final String path, final byte[] payload) throws Exception {
    client.setData().forPath(path, payload);
}
 
public static void delete(final CuratorFramework client, final String path) throws Exception {
    client.delete().deletingChildrenIfNeeded().forPath(path);
}
 
public static void guaranteedDelete(final CuratorFramework client, final String path) throws Exception {
    client.delete().guaranteed().forPath(path);
}
 
public static String getData(final CuratorFramework client, final String path) throws Exception {
    return new String(client.getData().forPath(path));
}
 
public static List<String> getChildren(final CuratorFramework client, final String path) throws Exception {
    return client.getChildren().forPath(path);
}
复制代码

3.curator-recipes

curator-recipes 提供了一些zk的典型使用场景的参考，主要介绍一下开发中经常使用的组件。

事件监听

zookeeper原生支持经过注册watcher来进行事件监听，可是其使用不是特别方便，须要开发人员本身反复注册watcher，比较繁琐。Curator引入Cache来实现对zookeeper服务端事务的监听。Cache是Curator中对事件监听的包装，其对事件的监听其实能够近似看做是一个本地缓存视图和远程Zookeeper视图的对比过程。同时Curator可以自动为开发人员处理反复注册监听，从而大大简化原生api开发的繁琐过程。

1）Node Cache

public static void nodeCache() throws Exception {
    final String path = "/nodeCache";
    final CuratorFramework client = getClient();
    client.start();
 
    delete(client, path);
    create(client, path, "cache".getBytes());
 
    final NodeCache nodeCache = new NodeCache(client, path);
    nodeCache.start(true);
    nodeCache.getListenable()
            .addListener(() -> System.out.println("node data change, new data is " + new String(nodeCache.getCurrentData().getData())));
 
    setData(client, path, "cache1".getBytes());
    setData(client, path, "cache2".getBytes());
 
    Thread.sleep(1000);
 
    client.close();
}
复制代码

NodeCache能够监听指定的节点，注册监听器后，节点的变化会通知相应的监听器

2）Path Cache

Path Cache 用来监听ZNode的子节点事件，包括added、updateed、removed，Path Cache会同步子节点的状态，产生的事件会传递给注册的PathChildrenCacheListener。

public static void pathChildrenCache() throws Exception {
        final String path = "/pathChildrenCache";
        final CuratorFramework client = getClient();
        client.start();
 
        final PathChildrenCache cache = new PathChildrenCache(client, path, true);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
 
        cache.getListenable().addListener((client1, event) -> {
            switch (event.getType()) {
                case CHILD_ADDED:
                    System.out.println("CHILD_ADDED:" + event.getData().getPath());
                    break;
                case CHILD_REMOVED:
                    System.out.println("CHILD_REMOVED:" + event.getData().getPath());
                    break;
                case CHILD_UPDATED:
                    System.out.println("CHILD_UPDATED:" + event.getData().getPath());
                    break;
                case CONNECTION_LOST:
                    System.out.println("CONNECTION_LOST:" + event.getData().getPath());
                    break;
                case CONNECTION_RECONNECTED:
                    System.out.println("CONNECTION_RECONNECTED:" + event.getData().getPath());
                    break;
                case CONNECTION_SUSPENDED:
                    System.out.println("CONNECTION_SUSPENDED:" + event.getData().getPath());
                    break;
                case INITIALIZED:
                    System.out.println("INITIALIZED:" + event.getData().getPath());
                    break;
                default:
                    break;
            }
        });
 
//        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
        Thread.sleep(1000);
 
        client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
        Thread.sleep(1000);
 
        client.delete().forPath(path + "/c1");
        Thread.sleep(1000);
 
        client.delete().forPath(path); //监听节点自己的变化不会通知
        Thread.sleep(1000);
 
        client.close();
    }
复制代码

3）Tree Cache

Path Cache和Node Cache的“合体”，监视路径下的建立、更新、删除事件，并缓存路径下全部孩子结点的数据。

public static void treeCache() throws Exception {
    final String path = "/treeChildrenCache";
    final CuratorFramework client = getClient();
    client.start();
 
    final TreeCache cache = new TreeCache(client, path);
    cache.start();
 
    cache.getListenable().addListener((client1, event) -> {
        switch (event.getType()){
            case NODE_ADDED:
                System.out.println("NODE_ADDED:" + event.getData().getPath());
                break;
            case NODE_REMOVED:
                System.out.println("NODE_REMOVED:" + event.getData().getPath());
                break;
            case NODE_UPDATED:
                System.out.println("NODE_UPDATED:" + event.getData().getPath());
                break;
            case CONNECTION_LOST:
                System.out.println("CONNECTION_LOST:" + event.getData().getPath());
                break;
            case CONNECTION_RECONNECTED:
                System.out.println("CONNECTION_RECONNECTED:" + event.getData().getPath());
                break;
            case CONNECTION_SUSPENDED:
                System.out.println("CONNECTION_SUSPENDED:" + event.getData().getPath());
                break;
            case INITIALIZED:
                System.out.println("INITIALIZED:" + event.getData().getPath());
                break;
            default:
                break;
        }
    });
 
    client.create().withMode(CreateMode.PERSISTENT).forPath(path);
    Thread.sleep(1000);
 
    client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
    Thread.sleep(1000);
 
    setData(client, path, "test".getBytes());
    Thread.sleep(1000);
 
    client.delete().forPath(path + "/c1");
    Thread.sleep(1000);
 
    client.delete().forPath(path);
    Thread.sleep(1000);
 
    client.close();
}
复制代码

选举

curator提供了两种方式，分别是Leader Latch和Leader Election。

1）Leader Latch

随机从候选着中选出一台做为leader，选中以后除非调用close()释放leadship，不然其余的后选择没法成为leader

public class LeaderLatchTest {
 
    private static final String PATH = "/demo/leader";
 
    public static void main(String[] args) {
        List<LeaderLatch> latchList = new ArrayList<>();
        List<CuratorFramework> clients = new ArrayList<>();
        try {
            for (int i = 0; i < 10; i++) {
                CuratorFramework client = getClient();
                client.start();
                clients.add(client);
 
                final LeaderLatch leaderLatch = new LeaderLatch(client, PATH, "client#" + i);
                leaderLatch.addListener(new LeaderLatchListener() {
                    @Override
                    public void isLeader() {
                        System.out.println(leaderLatch.getId() + ":I am leader. I am doing jobs!");
                    }
 
                    @Override
                    public void notLeader() {
                        System.out.println(leaderLatch.getId() + ":I am not leader. I will do nothing!");
                    }
                });
                latchList.add(leaderLatch);
                leaderLatch.start();
            }
            Thread.sleep(1000 * 60);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            for (CuratorFramework client : clients) {
                CloseableUtils.closeQuietly(client);
            }
 
            for (LeaderLatch leaderLatch : latchList) {
                CloseableUtils.closeQuietly(leaderLatch);
            }
        }
    }
 
    public static CuratorFramework getClient() {
        return CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .connectionTimeoutMs(15 * 1000) //链接超时时间，默认15秒
                .sessionTimeoutMs(60 * 1000) //会话超时时间，默认60秒
                .namespace("arch") //设置命名空间
                .build();
    }
 
}
复制代码

2）Leader Election

经过LeaderSelectorListener能够对领导权进行控制， 在适当的时候释放领导权，这样每一个节点都有可能得到领导权。 而LeaderLatch则一直持有leadership， 除非调用close方法，不然它不会释放领导权。

public class LeaderSelectorTest {
    private static final String PATH = "/demo/leader";
 
    public static void main(String[] args) {
        List<LeaderSelector> selectors = new ArrayList<>();
        List<CuratorFramework> clients = new ArrayList<>();
        try {
            for (int i = 0; i < 10; i++) {
                CuratorFramework client = getClient();
                client.start();
                clients.add(client);
 
                final String name = "client#" + i;
                LeaderSelector leaderSelector = new LeaderSelector(client, PATH, new LeaderSelectorListenerAdapter() {
                    @Override
                    public void takeLeadership(CuratorFramework client) throws Exception {
                        System.out.println(name + ":I am leader.");
                        Thread.sleep(2000);
                    }
                });
 
                leaderSelector.autoRequeue();
                leaderSelector.start();
                selectors.add(leaderSelector);
            }
            Thread.sleep(Integer.MAX_VALUE);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            for (CuratorFramework client : clients) {
                CloseableUtils.closeQuietly(client);
            }
 
            for (LeaderSelector selector : selectors) {
                CloseableUtils.closeQuietly(selector);
            }
 
        }
    }
 
    public static CuratorFramework getClient() {
        return CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .connectionTimeoutMs(15 * 1000) //链接超时时间，默认15秒
                .sessionTimeoutMs(60 * 1000) //会话超时时间，默认60秒
                .namespace("arch") //设置命名空间
                .build();
    }
 
}
复制代码

分布式锁

1）可重入锁Shared Reentrant Lock

Shared意味着锁是全局可见的， 客户端均可以请求锁。 Reentrant和JDK的ReentrantLock相似， 意味着同一个客户端在拥有锁的同时，能够屡次获取，不会被阻塞。 它是由类InterProcessMutex来实现。 它的构造函数为：

public InterProcessMutex(CuratorFramework client, String path)
复制代码

经过acquire得到锁，并提供超时机制：

/**
* Acquire the mutex - blocking until it's available. Note: the same thread can call acquire * re-entrantly. Each call to acquire must be balanced by a call to release() */ public void acquire(); /** * Acquire the mutex - blocks until it's available or the given time expires. Note: the same thread can
* call acquire re-entrantly. Each call to acquire that returns true must be balanced by a call to release()
* Parameters:
* time - time to wait
* unit - time unit
* Returns:
* true if the mutex was acquired, false if not
*/
public boolean acquire(long time, TimeUnit unit);
复制代码

经过release()方法释放锁。 InterProcessMutex 实例能够重用。 Revoking ZooKeeper recipes wiki定义了可协商的撤销机制。 为了撤销mutex, 调用下面的方法：

/**
* 将锁设为可撤销的. 当别的进程或线程想让你释放锁时Listener会被调用。
* Parameters:
* listener - the listener
*/
public void makeRevocable(RevocationListener<T> listener)
复制代码

2）不可重入锁Shared Lock

使用InterProcessSemaphoreMutex，调用方法相似，区别在于该锁是不可重入的，在同一个线程中不可重入

3）可重入读写锁Shared Reentrant Read Write Lock

相似JDK的ReentrantReadWriteLock. 一个读写锁管理一对相关的锁。 一个负责读操做，另一个负责写操做。 读操做在写锁没被使用时可同时由多个进程使用，而写锁使用时不容许读 (阻塞)。 此锁是可重入的。一个拥有写锁的线程可重入读锁，可是读锁却不能进入写锁。 这也意味着写锁能够降级成读锁， 好比请求写锁 —>读锁 —->释放写锁。 从读锁升级成写锁是不成的。 主要由两个类实现：

InterProcessReadWriteLock
InterProcessLock
复制代码

4）信号量Shared Semaphore

一个计数的信号量相似JDK的Semaphore。 JDK中Semaphore维护的一组许可(permits)，而Cubator中称之为租约(Lease)。注意，全部的实例必须使用相同的numberOfLeases值。 调用acquire会返回一个租约对象。 客户端必须在finally中close这些租约对象，不然这些租约会丢失掉。 可是， 可是，若是客户端session因为某种缘由好比crash丢掉， 那么这些客户端持有的租约会自动close， 这样其它客户端能够继续使用这些租约。 租约还能够经过下面的方式返还：

public void returnAll(Collection<Lease> leases)
public void returnLease(Lease lease)
复制代码

注意一次你能够请求多个租约，若是Semaphore当前的租约不够，则请求线程会被阻塞。 同时还提供了超时的重载方法:

public Lease acquire()
public Collection<Lease> acquire(int qty)
public Lease acquire(long time, TimeUnit unit)
public Collection<Lease> acquire(int qty, long time, TimeUnit unit)
复制代码

主要类有:

InterProcessSemaphoreV2
Lease
SharedCountReader
复制代码

5）多锁对象Multi Shared Lock

Multi Shared Lock是一个锁的容器。 当调用acquire， 全部的锁都会被acquire，若是请求失败，全部的锁都会被release。 一样调用release时全部的锁都被release(失败被忽略)。 基本上，它就是组锁的表明，在它上面的请求释放操做都会传递给它包含的全部的锁。 主要涉及两个类：

InterProcessMultiLock
InterProcessLock
复制代码

它的构造函数须要包含的锁的集合，或者一组ZooKeeper的path。

public InterProcessMultiLock(List<InterProcessLock> locks)
public InterProcessMultiLock(CuratorFramework client, List<String> paths)
复制代码

栅栏barrier

1）DistributedBarrier构造函数中barrierPath参数用来肯定一个栅栏，只要barrierPath参数相同(路径相同)就是同一个栅栏。一般状况下栅栏的使用以下：
1.主导client设置一个栅栏
2.其余客户端就会调用waitOnBarrier()等待栅栏移除，程序处理线程阻塞
3.主导client移除栅栏，其余客户端的处理程序就会同时继续运行。
DistributedBarrier类的主要方法以下：
setBarrier() - 设置栅栏
waitOnBarrier() - 等待栅栏移除
removeBarrier() - 移除栅栏

2）双栅栏Double Barrier
双栅栏容许客户端在计算的开始和结束时同步。当足够的进程加入到双栅栏时，进程开始计算，当计算完成时，离开栅栏。双栅栏类是DistributedDoubleBarrier DistributedDoubleBarrier类实现了双栅栏的功能。它的构造函数以下：

// client - the client
// barrierPath - path to use
// memberQty - the number of members in the barrier
public DistributedDoubleBarrier(CuratorFramework client, String barrierPath, int memberQty)
复制代码

memberQty是成员数量，当enter方法被调用时，成员被阻塞，直到全部的成员都调用了enter。当leave方法被调用时，它也阻塞调用线程，直到全部的成员都调用了leave。
注意：参数memberQty的值只是一个阈值，而不是一个限制值。当等待栅栏的数量大于或等于这个值栅栏就会打开！
与栅栏(DistributedBarrier)同样,双栅栏的barrierPath参数也是用来肯定是不是同一个栅栏的，双栅栏的使用状况以下：
1.从多个客户端在同一个路径上建立双栅栏(DistributedDoubleBarrier),而后调用enter()方法，等待栅栏数量达到memberQty时就能够进入栅栏。
2.栅栏数量达到memberQty，多个客户端同时中止阻塞继续运行，直到执行leave()方法，等待memberQty个数量的栅栏同时阻塞到leave()方法中。
3.memberQty个数量的栅栏同时阻塞到leave()方法中，多个客户端的leave()方法中止阻塞，继续运行。
DistributedDoubleBarrier类的主要方法以下： enter()、enter(long maxWait, TimeUnit unit) - 等待同时进入栅栏
leave()、leave(long maxWait, TimeUnit unit) - 等待同时离开栅栏
异常处理：DistributedDoubleBarrier会监控链接状态，当链接断掉时enter()和leave方法会抛出异常。

计数器Counters

利用ZooKeeper能够实现一个集群共享的计数器。 只要使用相同的path就能够获得最新的计数器值， 这是由ZooKeeper的一致性保证的。Curator有两个计数器， 一个是用int来计数，一个用long来计数。

1）SharedCount

这个类使用int类型来计数。 主要涉及三个类。

* SharedCount
* SharedCountReader
* SharedCountListener
复制代码

SharedCount表明计数器， 能够为它增长一个SharedCountListener，当计数器改变时此Listener能够监听到改变的事件，而SharedCountReader能够读取到最新的值， 包括字面值和带版本信息的值VersionedValue。

2）DistributedAtomicLong

除了计数的范围比SharedCount大了以外， 它首先尝试使用乐观锁的方式设置计数器， 若是不成功(好比期间计数器已经被其它client更新了)， 它使用InterProcessMutex方式来更新计数值。 此计数器有一系列的操做：

  • get(): 获取当前值
  • increment()： 加一
  • decrement(): 减一
  • add()： 增长特定的值
  • subtract(): 减去特定的值
  • trySet(): 尝试设置计数值
  • forceSet(): 强制设置计数值

你必须检查返回结果的succeeded()， 它表明此操做是否成功。 若是操做成功， preValue()表明操做前的值， postValue()表明操做后的值。

End

Curator抽象和简化了不少复杂的zookeeper操做，推荐替代zkclient包进行开发。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程