curator zkclient

原生api的不足：

（1）连接的创建是异步的，需要开发人员自行编码实现等待。

（2）连接没有自动的超进重连机制。

（3）zk本身没提供序列化机制，需要开发人员自行指定，从而实现数据的序列化和反序列化。

（4）Watcher注册一次只会生效一次，需要不断的重复注册。

（5）不支持递归创建树形节点。

使用Java操作zookeeper时，一般有两种方式：zkclient和curator，相比较来说，curator的使用较为简单。

CuratorFrameworkFactory 的静态方法创建客户端

（a）static CuratorFramework newClient(String connectString , int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)

（b）static CuratorFramework newClient(String connectString, RetryPolicy retryPolicy)

参数说明：

connectString 分开的ip:port对。
retryPolicy：重试策略
sessionTimeoutMs：会话超时时间，默认60000ms
connectionTimeoutMs：连接创建超时时间，单位为毫秒，默认是15000ms

实现接口RetryPolicy可以自定义重试策略：

boolean allowRetry(int retryCount, long elapsedTimeMs, RetrySleeper sleeper)

retryCount：已经重试的次数，如果第一次重试，此值为0.
elapsedTimeMs：重试花费的时间
sleeper：类似于Thread.sleep，用于sleep指定时间。
返回值：如果还会继续重试，则返回true

默认重试策略：

（1）RetryNTimes(int n, int sleepMsBetweenRetries)

n：最大重试次数。
sleepMsBetweenRetry：重试间隔的时间。

（2）RetryOneTime(int sleepMsBetweenRetry)

sleepMsBetweenRetry：重试间隔的时间

（3）RetryUntilElapsed(int maxElapsedTimeMs, int sleepMsBetweenRetries)

重试的时间超过最大时间后，就不再重试。

maxElapsedTimeMs：最大重试时间。
sleepMsBetweenRetriees:每次重试的间隔时间。

（4）ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs)

baseSleepTimeMs：初始sleep时间。
maxRetries：最大重试次数。
maxSleepMs:最大重试时间。

当前应该sleep的时间：baseSleepTimeMs*Math.max(1,random.nextInt(1<<retryCount+1))，随着重试次数，增加重试时间间隔，指数增加。

Fluent风格的API

一种面向对象的开发方式，目的是提高代码的可读性。

例子：

zkclient = CuratorFramework.builder().connectString(connectString).sessionTimeoutMs(5000).retryPolicy(retryPolicy)。

创建节点

CreateBuilder create()

createingParentsIfNeeded()递归创建父目录。
withMode(CreateMode mode)设置节点属性，如果是递归创建，创建模式为临时节点，则只有叶子节点是临时节点，非叶子都为持久节点。
withACL(List aclList)
forPath(String path) 指定路径。

删除节点

DeleteBuilder delete()

withVersion(int version)
guaranteed() 确保节点被删除
forPath(String path) 指定路径
deletingChildrenIfNeeded 递归删除所有子节点

读取数据

GetDataBuilder getData()

storingStatIn(org.apache.zookeeper.data.Stat stat) 把服务器端获取的状态数据存储到stat对象。
Byte[] forPath(String path) 节点路径

设置watcher

（1）NodeCache

监听数据节点的内容变更，如果指定的节点不存在，则节点创建后，会触发这个监听。

（2）PathChildrenCache

监听指定节点的子节点变化情况。

package com.zk.dev.zkClient.day1;  import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;  /** * @see 测试curator框架例子 * @Author:xuehan * @Date:2016年5月14日下午8:44:49 */
public class CuratorUtils {  public String connectString = "localhost:2181";  CuratorFramework  zkclient = null ;  public CuratorUtils(){  /** * connectString连接字符串中间用分号隔开,sessionTimeoutMs session过期时间,connectionTimeoutMs连接超时时间,retryPolicyc连接重试策略 */  //CuratorFrameworkFactory.newClient(connectString, sessionTimeoutMs, connectionTimeoutMs, retryPolicy)  // fluent风格aip   //    CuratorFrameworkFactory.builder().sessionTimeoutMs(5000).connectString(connectString).namespace("/test").build();  // 重连策略,没1一秒重试一次,最大重试次数3次  RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);  zkclient = CuratorFrameworkFactory.builder().connectString(connectString).sessionTimeoutMs(5000).retryPolicy(retryPolicy).namespace("tests").build();  zkclient.start();  }  /** * 递归创建节点 * @param path * @param data * @throws Exception */  public void createNode(String path, byte[] data) throws Exception{  zkclient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).withACL(Ids.OPEN_ACL_UNSAFE).forPath(path, data);  }  /** * 递归删除节点 * @param path * @throws Exception */  public void delNode(String path) throws Exception{  zkclient.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);  }   public void zkClose(){  zkclient.close();  }  public void delNodeCallBack(String path) throws Exception{  zkclient.delete().guaranteed().deletingChildrenIfNeeded().inBackground(new DeleteCallBack()).forPath(path);  }      public void dataChanges(String path) throws Exception{  final NodeCache  dataWatch =  new NodeCache(zkclient, path);  dataWatch.start(true);  dataWatch.getListenable().addListener(new NodeCacheListener(){  public void nodeChanged() throws Exception {  System.out.println("path==>" + dataWatch.getCurrentData().getPath() + "==data==>" + new String(dataWatch.getCurrentData().getData()));  }  });  zkclient.delete().guaranteed().deletingChildrenIfNeeded().inBackground(new DeleteCallBack()).forPath(path);  }      public void addChildWatcher(String path) throws Exception{  final PathChildrenCache pc = new PathChildrenCache(zkclient, path, true);  pc.start(StartMode.POST_INITIALIZED_EVENT);  System.out.println("节点个数===>" + pc.getCurrentData().size());  pc.getListenable().addListener(new  PathChildrenCacheListener() {  public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {  System.out.println("事件监听到"  + event.getData().getPath());  if(event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)){  System.out.println("客户端初始化节点完成"  + event.getData().getPath());  }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)){  System.out.println("添加节点完成"  + event.getData().getPath());  }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){  System.out.println("删除节点完成"  + event.getData().getPath());  }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){  System.out.println("修改节点完成"  + event.getData().getPath());  }  }  });  }  public static void main(String[] args) throws Exception{  CuratorUtils cu = new CuratorUtils();
//      cu.createNode("/test/sb/aa/bb", "erhu".getBytes());
//      cu.delNode("/test");  cu.zkclient.setData().forPath("/aa", "love is not".getBytes());  cu.addChildWatcher("/aa");  try{  Thread.sleep(20000000);  }catch(Exception e){};  }  }

最后欢迎大家访问我的个人网站：1024s

curator zkclient相关推荐

高性能微服务 Dubbo RPC Redis 分布式框架
开发目标兼容电脑.平板和手机的浏览器支持分布式部署,支持Redis集群友好的代码结构及注释,便于阅读及二次开发提供Swagger API接口,方便团队协作和功能测试 Shiro认证授权,包含菜 ...
亲！你的专属Dubbo启动注册中心，请注意查收！
现在来说一下dubbo注册中心的大致结构,请看下图: Java学习笔记开源框架--dubbo启动之注册中心(Registry) 服务注册对于服务提供方,它需要发布服务,而且由于应用系统的复杂性,服务 ...
LTS分布式任务调度文档
LTS介绍 LTS(light-task-scheduler)主要用于解决分布式任务调度问题,支持实时任务,定时任务,Cron任务,Repeat任务.有较好的伸缩性,扩展性,健壮稳定性而被多家公司使用 ...
LTS分布式任务调度文档-转载
转载:https://www.cnblogs.com/dion-90/articles/8674591.html LTS介绍 LTS(light-task-scheduler)主要用于解决分布式任务调 ...
Zookeeper实战之客户端Curator vs zkClient
1. zookeeper应用开发 Zookeeper应用开发,需要使用Zookeeper的java 客户端API ,去连接和操作Zookeeper 集群. 可以供选择的java 客户端API 有:Zo ...
zookeeper学习笔记-zkclient,curator使用
开源客户端,原生api的不足连接的创建是异步的,需要开发人员自行编码实现等待连接没有自动的超时重连机制 Zk本身没提供序列化机制,需要开发人员自行指定,从而实现数据的序列化和反序列化 Wat ...
Zookeeper客户端ZkClient、Curator的使用，史上最详细的教程来啦~
1 前言本文主要介绍了操作Zookeeper的几种客户端的基础使用,希望对老铁们会有所帮助. 可以去操作zookeeper创建.删除.查询.修改znode节点 2 Zookeeper服务器客户端分类 ...
Zookeeper开源客户端框架Curator的简单使用
为什么80%的码农都做不了架构师?>>> Curator最初由Netflix的Jordan Zimmerman开发, Curator提供了一套Java类库, 可以更容易的使用Z ...
聊聊、Zookeeper 客户端 Curator
[Curator] 和 ZkClient 一样,Curator 也是开源客户端,Curator 是 Netflix 公司开源的一套框架. <dependency><groupId ...

curator zkclient

curator zkclient相关推荐

最新文章

热门文章