Apache curator-client详解
Posted bcombettter
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Apache curator-client详解相关的知识,希望对你有一定的参考价值。
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的方式实现阻塞.
以上是关于Apache curator-client详解的主要内容,如果未能解决你的问题,请参考以下文章
你真的了解LAMP/LNMP架构吗?(详解PHP和Apache和MariaDB之间的工作联系)以及安装php7,8版本。