zookeeper(三）zookeeper的多种客户端

stay hungry,stay you

826人浏览 · 2020-03-15 19:09:20

stay hungry,stay you · 2020-03-15 19:09:20 发布

（一）使用ZooKeeper原生API

命名空间：

Chroot特性允许每个客户端设置一个命名空间，如果一个Zookeeper客户端设置了Chroot，那么该客户端对服务器的任何操作，都将被限定在自己的命名空间下。
如果我们希望为应用分配/apps/X下的所有子节点，那么该应用可以将所有Zookeeper客户端的Chroot设置为/apps/X。一旦设置了Chroot后，那么对于这个客户端来说，所有的节点路径都已/apps/X为跟节点。
客户端可以在connectString中以添加后缀的方式来设置，如：

192.168.56.101:2181,192.168.56.101:2182,192.168.56.101:2183/apps/X

负载均衡策略：

StaticHostProvider是ZK默认的一种非常简单的负载均衡策略，它的表现形式其实类似“Random Robin”策略。StaticHostProvider会从客户端输入构成服务器地址，然后通过其next方法从serverAddress中获取一个服务器地址时，会先将服务器地址打散然后拼装成一个环形循环列表。
如原始地址访问字符为：“host1,host2,host3,host4,host5”经过打散重新拼装后会构成环形列表：
在这里插入图片描述
初始化的时候currentIndex和lastIndex都是-1。每次尝试获取一个服务器地址的时候，都会将currentIndex向前移动1位，如果发现游标移动超过了整个地址列表的长度，那么就重置0，回到开始的位置重新开始。对于那些服务器地址列表提供的比较少的场景，StaticHostProvider如果发现当前游标位置和上次使用过的地址一样，即当currentIndex和lastIndex相同时，就进行spinDelay毫秒时间的等待。

具体源码如下：

// 初始化时currentIndex和lastIndex都为-1
// lastIndex表示当前正在使用的服务器地址位置
private int lastIndex = -1;

// currentIndex表示环形队列中当前遍历到的那个元素位置
private int currentIndex = -1;

private void init(Collection<InetSocketAddress> serverAddresses) {
    if (serverAddresses.isEmpty()) {
        throw new IllegalArgumentException(
            "A HostProvider may not be empty!");
    }

    this.serverAddresses.addAll(serverAddresses);
    // 打散集合
    Collections.shuffle(this.serverAddresses);
}

// 获取服务器地址函数
public InetSocketAddress next(long spinDelay) {
    // currentIndex向前移动一位并与服务器数量做“%”操作
    currentIndex = ++currentIndex % serverAddresses.size();
    // 如果currentIndex和上次访问的服务器相同，则休眠spinDelay毫秒
    if (currentIndex == lastIndex && spinDelay > 0) {
        try {
            Thread.sleep(spinDelay);
        } catch (InterruptedException e) {
            LOG.warn("Unexpected exception", e);
        }
    } else if (lastIndex == -1) {
        // We don't want to sleep on the first ever connect attempt.
        lastIndex = 0;
    }

    // 根据currentIndex获取地址
    InetSocketAddress curAddr = serverAddresses.get(currentIndex);
    try {
        // 把地址解析成字符串
        String curHostString = getHostString(curAddr);
        // 根据字符解析所有网络地址
        List<InetAddress> resolvedAddresses = new ArrayList<InetAddress>(Arrays.asList(this.resolver.getAllByName(curHostString)));
        // 如果List为空，代表没有可以解析到的其他主机
        if (resolvedAddresses.isEmpty()) {
            return curAddr;
        }
        // 如果List不为空，把解析到的集合再次打散
        Collections.shuffle(resolvedAddresses);
        // 返回打散集合的第一条数据
        return new InetSocketAddress(resolvedAddresses.get(0), curAddr.getPort());
    } catch (UnknownHostException e) {
        return curAddr;
    }
}

// 获取到服务器地址后，连接服务器时调用
public void onConnected(){
    // 让lastIndex等于currentIndex，相当于lastIndex移动到currentIndex的位置来表示最后一次访问的服务器
    lastIndex = currentIndex;
}

在使用ZooKeeper的Java客户端时，经常需要处理几个问题：

Watcher反复注册
Session超时重连
异常处理

要解决上述的几个问题，可以自己解决，也可以采用第三方的java客户端来完成。

（二）使用ZkClient

创建会话：

public ZkClient(final String zkServers, final int sessionTimeout, 
       final int connectionTimeout, final ZkSerializer zkSerializer, 
       final long operationRetryTimeout)

创建节点：
同步，可以递归创建

public String create(String path,Object data,final List<ACL> acl,CreateMode mode)
public void createPersistent(String path,boolean createParents,List<ACL> acl)
public void createPersistent(String path, Object data, List<ACL> acl)
public String createPersistentSequential(String path,Object data,List<ACL> acl)
public void createEphemeral(String path, Object data, List<ACL> acl)
public String createEphemeralSequential(String path,Object data,List<ACL> acl)

删除节点：
同步，可以提供递归删除

public boolean delete(String path,int version)
public boolean deleteRecursive(String path)

获取节点：
同步，避免不存在异常

public List<String> getChildren(String path)
public <T> T readData(String path, boolean returnNullIfPathNotExists)
public <T> T readData(String path, Stat stat)

更新节点：
同步，实现CAS，状态返回

public void writeData(String path, Object datat, int expectedVersion)
public Stat writeDataReturnStat(String path,Object datat,int expectedVersion)

判断节点是否存在：

public boolean exists(String path)

事件监听：

public List<String> subscribeChildChanges(String path, IZkChildListener listener)

事件通知：

public void handleChildChange(String parentPath, List<String> currentChilds)

（三）使用Curator

Curator是Netflix公司开源的一个Zookeeper客户端，与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。

封装ZooKeeper client与ZooKeeper server之间的连接处理；
提供了一套Fluent风格的操作API；
提供ZooKeeper各种应用场景(recipe，比如共享锁服务，集群领导选举机制)的抽象封装。

Curator几个组成部分

Client：是ZooKeeper客户端的一个替代品，提供了一些底层处理和相关的工具方法。
Framework：用来简化ZooKeeper高级功能的使用，并增加了一些新的功能，比如管理到ZooKeeper集群的连接，重试处理。
Recipes：实现了通用ZooKeeper的recipe，该组件建立在Framework的基础之上
Utilities：各种ZooKeeper的工具类。
Errors：异常处理，连接，恢复等。
Extensions：recipe扩展

RetryPolicy 连接策略

RetryOneTime：只重连一次.
RetryNTime：指定重连的次数N.
RetryUtilElapsed：指定最大重连超时时间和重连时间间隔，间歇性重连直到超时或者链接成功。
ExponentialBackoffRetry：基于"backoff（退避）"方式重连，和RetryUtilElapsed的区别是重连的时间间隔是动态的。
BoundedExponentialBackoffRetry：同ExponentialBackoffRetry，增加了最大重试次数的控制。

Curator的API

创建会话：

CuratorFrameworkFactory.newClient(String connectString, int sessionTimeoutMs, 
				int connectionTimeoutMs, RetryPolicy retryPolicy)

CuratorFrameworkFactory.builder().connectString("192.168.11.56:2180")  
		    .sessionTimeoutMs(30000).connectionTimeoutMs(30000)  
		    .canBeReadOnly(false)  
		    .retryPolicy(new ExponentialBackoffRetry(1000, Integer.MAX_VALUE))
		    .build();

创建节点：

client.create().creatingParentIfNeeded()
        .withMode(CreateMode.PERSISTENT)
        .withACL(aclList)
        .forPath(path, "hello, zk".getBytes());

删除节点：

client.delete().guaranteed().deletingChildrenIfNeeded().withVersion(version).forPath(path)

获取节点：

client.getData().storingStatIn(stat).forPath(path);

client.getChildren().forPath(path);

更新节点：

client.setData().withVersion(version).forPath(path, data)

判断节点是否存在：

client.checkExists().forPath(path)；

设置权限：

Build.authorization(String scheme, byte[] auth)
client.setACL().withVersion(version)
        .withACL(ZooDefs.Ids.CREATOR_ALL_ACL)
        .forPath(path);

监听器：