Apache curator-client详解

111

    Apache curator框架中curator-client组件可以作为zookeeper client来使用,它提供了zk实例创建/重连机制等,简单便捷.不过直接使用curator-client并不能减少太多的开发量,因为它相对比较底层,稍后我们继续了解curator-framework组件提供的更多的便捷特性.
 
一.核心API
    1. CuratorZookeeperClient类: zookeeper客户端,根据指定的配置信息创建zookeeper实例.
    2. RetryPolicy接口: 重连策略,当zookeeper失去链接时使用的"重连策略": 
        <> RetryOneTime: 只重连一次.
        <> RetryNTime: 指定重连的次数N.
        <> RetryUtilElapsed: 指定最大重连超时时间和重连时间间隔,间歇性重连直到超时或者链接成功.
        <> ExponentialBackoffRetry: 基于"backoff"方式重连,和RetryUtilElapsed的区别是重连的时间间隔是动态的.
 

时间间隔 = baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1))).
 
        <> BoundedExponentialBackoffRetry: 同ExponentialBackoffRetry,增加了最大重试次数的控制.
    3. RetryLoop类: 操作重试,如果在执行一个操作时,遇到了zk链接异常,怎么办?RetryLoop可以不断重试,直到网络正常且操作执行成功为止.SessionFailRetryLoop类是一个特列,可以兼容当session失效时,如何进行操作重试.
    4. EnsembleProvider: 配置提供者,创建zk客户端时,需要指定connectionString(例如:127.0.0.1:2181),在zookeeper API中只能显式的指定,curator在这个方面提供了更加灵活性的方式,你可以通过任何方式获取或者构建connectionString.
        <> FixedEnsembleProvider: 使用固定的字符串作为connectionString.
        <> ExhibitorEnsembleProvider: 动态的获取connectionString,可以指定一个URL用来提供connectionString的输出服务.此后此Provider将会间歇性的获取最新的connectionString字符串,并保存.事实上,ExhibitorEnsembleProvider只是一个样例,展示了一种动态获取connectionString的方式,如果在真正的开发中,你可能需要参考它,来定制自己的Provider.
 
二. 通用客户端代码示例

public class ZooKeeperClient extends Thread{
protected final CuratorZookeeperClient zkClient;
protected String parent;
public static final Charset charset = Charset.forName("utf-8");
private ZNodeWatcher zNodeWatcher = new ZNodeWatcher();//自定义watcher
public ZooKeeperClient(String connectString, int sessionTimeout, String parent) throws Exception {
this.parent = parent;
zkClient = new CuratorZookeeperClient(connectString, sessionTimeout, sessionTimeout, zNodeWatcher, new ExponentialBackoffRetry(1000, Integer.MAX_VALUE));
zkClient.start();//must,but anytime before zookeeper operation
zkClient.blockUntilConnectedOrTimedOut(); //first connection should be successful
}

public boolean exist(String path,boolean watched) throws Exception{
return zkClient.getZooKeeper().exists(path,watched) == null ? false : true;
}
/**
* 此path必须存在，如果不存在则立即创建
* @param path
* @return
*/
public boolean ensurePath(final String path) throws Exception{
PathUtils.validatePath(path);
return RetryLoop.callWithRetry(zkClient, new Callable<Boolean>(){
@Override
public Boolean call() throws Exception {
EnsurePath ensure = new EnsurePath(path);
ensure.ensure(zkClient);
return true;
}
});
}
/**
*
* @param path
* @param data
* @return   如果path已经存在或者创建成功，则返回true，否则返回false。
* @throws Exception
*/
public boolean create(final String path, final String data) throws Exception {
PathUtils.validatePath(path);//if bad format,here will throw some Exception;
return RetryLoop.callWithRetry(zkClient, new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
int _current = 0;
while (_current < 3) {
_current++;
try {
//zkClient.blockUntilConnectedOrTimedOut();
//确保父节点存在
EnsurePath ensure = new EnsurePath(path).excludingLast();
//parent path should be existed.
//EnsurePath: retry + block
ensure.ensure(zkClient); //ugly API
zkClient.getZooKeeper().create(path, data.getBytes(charset), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
return true;
} catch (KeeperException.NodeExistsException e) {
return true;
}
//retry only for KeeperException，not for other runtimeException。
//other exception will be thrown，and stop retry！！
//if no Exception thrown,retry will be stopped and return successfully.
}
return false;
}
}) ;
}

public  class ZNodeWatcher implements Watcher{
@Override
public void process(WatchedEvent event) {
Event.EventType eventType = event.getType();
Event.KeeperState keeperState =  event.getState();
String path = event.getPath();
switch(event.getType()) {
case None:
//connection Error：会自动重连
logger.info("[Watcher],Connecting...");
if(keeperState == Event.KeeperState.SyncConnected){
logger.info("[Watcher],Connected...");
//检测临时节点是否失效等。
}
break;
case NodeCreated:
logger.info("[Watcher],NodeCreated:" + path);
break;
case NodeDeleted:
logger.info("[Watcher],NodeDeleted:" + path);
break;
default:
//
}
}
}
}
 
 
三. Provider代码实例
　　    本实例展示了如何使用curator-client开发简单的API,展示了RetryPolicy,RetryLoop的使用方式;实例中使用Curator自带的ExhibitorEnsembleProvider动态获取zookeeper服务器列表信息.
    1. pom.xml

<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.5</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.3.0</version>
</dependency>

   2. curator-config.propeties 

host.rest.servers=127.0.0.1,localhost
host.rest.port=8080
host.backup=127.0.0.1:2181
host.rest.path=/servers/zk
host.rest.period=180000
    3. IZkClient.java  

package com.test.demo.curator;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.apache.curator.CuratorZookeeperClient;
import org.apache.curator.RetryLoop;
import org.apache.curator.RetryPolicy;
import org.apache.curator.TimeTrace;
import org.apache.curator.drivers.TracerDriver;
import org.apache.curator.ensemble.EnsembleProvider;
import org.apache.curator.ensemble.exhibitor.DefaultExhibitorRestClient;
import org.apache.curator.ensemble.exhibitor.ExhibitorEnsembleProvider;
import org.apache.curator.ensemble.exhibitor.ExhibitorRestClient;
import org.apache.curator.ensemble.exhibitor.Exhibitors;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.EnsurePath;
import org.apache.curator.utils.PathUtils;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Transaction;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;

public class IZkClient {
private final CuratorZookeeperClient zkClient;
public IZkClient(String configLocation) throws Exception {
Properties properties = new Properties();
properties.load(ClassLoader.getSystemResourceAsStream(configLocation));
EnsembleProvider provider = buildProvider(properties);
String pTimeout = properties.getProperty("zk.timeout");
Integer timeout = 30000;
if (pTimeout != null) {
timeout = Integer.valueOf(pTimeout);
}
zkClient = new CuratorZookeeperClient(provider, timeout, timeout, null, new ExponentialBackoffRetry(1000, Integer.MAX_VALUE));
zkClient.setTracerDriver(new PrintTraceDrive());
zkClient.start();//must,but anytime before zookeeper operation
zkClient.blockUntilConnectedOrTimedOut(); //first connection should be successful
}
/**
* build provider,all of params from config-file
* @param properties
* @return
*/
private EnsembleProvider buildProvider(Properties properties) {
String servers = properties.getProperty("host.rest.servers");   //hosts.servers = 127.0.0.1,127.0.0.2
if (servers == null || servers.isEmpty()) {
throw new IllegalArgumentException("host.servers cant be empty");
}
List<String> hostnames = Arrays.asList(servers.split(","));
String port = properties.getProperty("host.rest.port");
Integer restPort = 80;   //default
if (port != null) {
restPort = Integer.valueOf(port);
}
final String backupAddress = properties.getProperty("host.backup");//127.0.0.1:2181
//if network is error,you should sepcify a backup zk-connectString
Exhibitors exhibitors = new Exhibitors(hostnames, restPort, new Exhibitors.BackupConnectionStringProvider() {
@Override
public String getBackupConnectionString() throws Exception {
return backupAddress;
}
});
//rest,as meaning of getting fresh zk-connectString list.
ExhibitorRestClient restClient = new DefaultExhibitorRestClient();
String restUriPath = properties.getProperty("host.rest.path");
String period = properties.getProperty("host.rest.period");
Integer pollingMs = 180000; //3 min
if (period != null) {
pollingMs = Integer.valueOf(period);
}
return new ExhibitorEnsembleProvider(exhibitors, restClient, restUriPath, pollingMs, new RetryNTimes(10, 1000));
}
public CuratorZookeeperClient getZkClient() {
return zkClient;
}
/**
* how to use RtryLoop ,another style
* if Znode has been existed,will delete it,and create it again.
*
*/
public boolean replace(final String path,final byte[] value){
PathUtils.validatePath(path);
boolean result = false;
try{
result = RetryLoop.callWithRetry(zkClient,new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
int _current = 0;
while(_current < 3){
_current++;
try{
zkClient.blockUntilConnectedOrTimedOut();
Transaction tx = zkClient.getZooKeeper().transaction();
tx.delete(path, -1);
tx.create(path,value,ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
tx.commit();
return true;
} catch (KeeperException.NoNodeException e){
//
} catch (KeeperException.NodeExistsException e){
//
}
}
return false;  //To change body of implemented methods use File | Settings | File Templates.
}
}) ;
}catch (Exception e){
e.printStackTrace();
}
return result;
}
//API : on for test
public String createPath(String path, byte[] value) throws Exception {
PathUtils.validatePath(path);//if bad format,here will throw some Exception;
EnsurePath ensure = new EnsurePath(path).excludingLast();
//parent path should be existed.
//EnsurePath: retry + block
ensure.ensure(zkClient); //ugly API
return zkClient.getZooKeeper().create(path, value, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
//API: on for test
public boolean createPath(String path, byte[] value,int blockTimes){
if (!zkClient.isConnected() && blockTimes == 0) {
return false;
}
TimeTrace trace = zkClient.startTracer("createPath:" + path);//log message
try{
EnsurePath ensure = new EnsurePath(path).excludingLast();
ensure.ensure(zkClient);//only for persistent node
RetryLoop loop = zkClient.newRetryLoop();
int _current = 0;
while(loop.shouldContinue()){
try{
if(_current >= blockTimes){
loop.markComplete(); //stop here.
continue;
}
//blocking
boolean isConnected = zkClient.blockUntilConnectedOrTimedOut();
if(!isConnected){
_current++;
continue;
}
zkClient.getZooKeeper().create(path, value, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
loop.markComplete();
} catch (KeeperException.NodeExistsException e){
loop.markComplete();//exist ,stop here
} catch (Exception e){
loop.takeException(e);
}
}
} catch (Exception e){
e.printStackTrace();
return false;  //cant create path
} finally{
trace.commit();
}
return true;
}
public byte[] getData(String path) throws Exception{
PathUtils.validatePath(path);
return zkClient.getZooKeeper().getData(path,false,null);
}
public void close(){
zkClient.close();
}
class PrintTraceDrive implements TracerDriver {
@Override
public void addTrace(String name, long time, TimeUnit unit) {
System.out.println("<Trace>" + name + ";time:" + TimeUnit.MILLISECONDS.convert(time, unit) + " ms");
}
@Override
public void addCount(String name, int increment) {
}
}
}

    4. IZkClientMain.java(for testing) 

public class IZkClientMain {

public static void main(String[] args) throws Exception {
String configLocation = "curator-config.properties";
IZkClient iZkClient = new IZkClient(configLocation);
String value = "curator-demo";
String path = "/curator/child/0";
iZkClient.replace(path, value.getBytes("utf-8"));
//simple method;
String nodeName = ZKPaths.getNodeFromPath(path);
System.out.print(nodeName);
//value
byte[] bytes = iZkClient.getData(path);
System.out.println(new String(bytes, "utf-8"));
Thread.sleep(180000 * 2);
iZkClient.close();
}
}

    5. ExhibitorEnsembleProvider需要使用远端的一个REST风格的Url来提供zookeeper服务器列表,如下为Spring方式:

@Controller
@RequestMapping("/servers")
public class ServersController {
@RequestMapping(value = "/zk",headers="Accept=application/x-www-form-urlencoded")
public void zk(HttpServletResponse response) throws  Exception{
FormHttpMessageConverter converter = new FormHttpMessageConverter();
converter.setCharset(Charset.forName("utf-8"));
HttpOutputMessage output = new ServletServerHttpResponse(response);
converter.write(buildServers(), MediaType.APPLICATION_FORM_URLENCODED,output);
//String servers = "count=2&port=2181&server0=127.0.0.1&server1=localhost";
}

private MultiValueMap<String,Object> buildServers(){
MultiValueMap<String,Object> map = new LinkedMultiValueMap<String, Object>();
map.add("count","2");
map.add("port","2181");
map.add("server0","127.0.0.1");
map.add("server1","localhost");
return map;
}
}
  
   备注:Curator-client可以帮助我们进行链接重连操作,重连过程中,我们不需要关注太多,不过你仍然可以通过注册Watcher的手段来活的通知.如果在操作过程中,zk的链接有异常,你可以通过RetryLoop的方式实现阻塞.