正文
zookeeper客户端使用第三方(Curator)封装的Api操作节点
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
- 为什么使用Curator?
- Curator本身是Netflix公司开源的zookeeper客户端；
- Curator 提供了各种应用场景的实现封装；
- curator-framework 提供了fluent风格api；
- curator-replice 提供了实现封装；
- 引入依赖：
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>2.11.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.11.0</version>
</dependency>
- 创建会话连接
package com.karat.cn.zookeeper.curator; import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry; /**
* 创建会话连接
* @author Administrator
*
*/
public class CuratorCreateSessionDemo {
private final static String CONNECTSTRING="47.107.121.215:2181"; public static void main(String args[]) {
//创建会话连接的2种方式
//正常的风格
CuratorFramework curatorFramework1=CuratorFrameworkFactory.
newClient(CONNECTSTRING,5000,5000,
new ExponentialBackoffRetry(1000, 3));//重试机制
curatorFramework1.start();
//fluent风格
CuratorFramework curatorFramework2=CuratorFrameworkFactory.builder().
connectString(CONNECTSTRING).
sessionTimeoutMs(5000).
retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
curatorFramework2.start();
System.out.println("success"); }
}
- curator连接的重试策略
ExponentialBackoffRetry() 衰减重试
RetryNTimes 指定最大重试次数
RetryOneTime 仅重试一次
RetryUnitilElapsed 一直重试知道规定的时间
- 节点操作
package com.karat.cn.zookeeper.curator; import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat; import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; /**
* curator对节点的增删改查
* @author Administrator
*
*/
public class CuratorOperatorDemo { public static void main(String[] args) throws InterruptedException {
CuratorFramework curatorFramework=CuratorClientUtils.getInstance();
System.out.println("连接成功........."); //fluent风格api增删改查操作
/**
* 创建节点
*/
/*try {
String result=curatorFramework.create()
.creatingParentsIfNeeded()//创建父节点
.withMode(CreateMode.PERSISTENT)//持久节点：节点创建后，会一直存在，不会因客户端会话失效而删除；
.forPath("/curator/curator1/curator11","123".getBytes());
System.out.println(result);
} catch (Exception e) {
e.printStackTrace();
}*/
/**
* 删除节点
*/
/*try {
//默认情况下，version为-1
curatorFramework.delete()//删除操作
.deletingChildrenIfNeeded()//删除子节点
.forPath("/node");
} catch (Exception e) {
e.printStackTrace();
}*/ /**
* 查询
*/
/*Stat stat=new Stat();
try {
byte[] bytes=curatorFramework
.getData()
.storingStatIn(stat)
.forPath("/curator/curator1/curator11");
System.out.println(new String(bytes)+"-->stat:"+stat);
} catch (Exception e) {
e.printStackTrace();
}*/
/**
* 更新
*/
/*try {
Stat stat=curatorFramework
.setData()
.forPath("/curator","lijing".getBytes());
System.out.println(stat);
} catch (Exception e) {
e.printStackTrace();
}*/ /**
* 异步操作
*/
/*ExecutorService service= Executors.newFixedThreadPool(1);//线程池(创建节点的事件由线程池处理)
CountDownLatch countDownLatch=new CountDownLatch(1);//计数器
try {
curatorFramework
.create()
.creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL)
.inBackground(new BackgroundCallback() {
@Override
public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
System.out.println(Thread.currentThread().getName()
+"->resultCode:"
+curatorEvent.getResultCode()+"->"//响应结果
+curatorEvent.getType());//当前节点操作类型
countDownLatch.countDown();
}
},service)
.forPath("/mic","123".getBytes());
} catch (Exception e) {
e.printStackTrace();
}
countDownLatch.await();//等待(让当前线程等待)
service.shutdown();//关闭线程*/
/**
* 事务操作（curator独有的）
*/
try {
Collection<CuratorTransactionResult> resultCollections=curatorFramework
.inTransaction()//开启一个事务
.create()
.forPath("/trans","111".getBytes())//创建一个节点
.and()//通过and去修改一个节点
.setData()
.forPath("/curator","111".getBytes())//当修改节点不存在，则一成功一失败，事务不会提交成功
.and()
.commit();//提交事务
for (CuratorTransactionResult result:resultCollections){
System.out.println(result.getForPath()+"->"+result.getType());
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
- 将会话连接做成工具类
package com.karat.cn.zookeeper.curator; import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry; /**
* 会话连接工具类
* @author Administrator
*
*/
public class CuratorClientUtils { private static CuratorFramework curatorFramework; private final static String CONNECTSTRING="47.107.121.215:2181"; public static CuratorFramework getInstance(){
curatorFramework= CuratorFrameworkFactory.
newClient(CONNECTSTRING,5000,5000,
new ExponentialBackoffRetry(1000,3));
curatorFramework.start();
return curatorFramework;
}
}
- 监听
package com.karat.cn.zookeeper.curator; import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.zookeeper.CreateMode; import java.util.concurrent.TimeUnit; /**
* 监听
* @author Administrator
*
*/
public class CuratorEventDemo { /**
* 三种watcher来做节点的监听
* pathcache 监视一个路径下子节点的创建、删除、节点数据更新
* NodeCache 监视一个节点的创建、更新、删除
* TreeCache pathcaceh+nodecache 的合体（监视路径下的创建、更新、删除事件），
* 缓存路径下的所有子节点的数据
*/ public static void main(String[] args) throws Exception {
CuratorFramework curatorFramework=CuratorClientUtils.getInstance();
/**
* 节点变化NodeCache
*/
//监听
/*NodeCache cache=new NodeCache(curatorFramework,"/curator",false);
cache.start(true);
//监听事件
cache.getListenable().addListener(()-> System.out.println("节点数据发生变化,变化后的结果" +
"："+new String(cache.getCurrentData().getData())));
//修改节点
curatorFramework.setData().forPath("/curator","666".getBytes());*/ /**
* PatchChildrenCache
*/ PathChildrenCache cache=new PathChildrenCache(curatorFramework,"/event",true);//参数2监听的节点，参数3是否缓存
cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
// Normal/ BUILD_INITIAL_CACHE /POST_INITIALIZED_EVENT cache.getListenable().addListener((curatorFramework1,pathChildrenCacheEvent)->{
switch (pathChildrenCacheEvent.getType()){
case CHILD_ADDED:
System.out.println("增加子节点");
break;
case CHILD_REMOVED:
System.out.println("删除子节点");
break;
case CHILD_UPDATED:
System.out.println("更新子节点");
break;
default:break;
}
});
//创建节点
curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath("/event","event".getBytes());
TimeUnit.SECONDS.sleep(1);
System.out.println("1");
//创建子节点
curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath("/event/event1","1".getBytes());
TimeUnit.SECONDS.sleep(1);
System.out.println("2");
//修改节点
curatorFramework.setData().forPath("/event/event1","222".getBytes());
TimeUnit.SECONDS.sleep(1);
System.out.println("3");
//删除节点
curatorFramework.delete().forPath("/event/event1");
System.out.println("4"); System.in.read();
}
}
