为了更好的实现Java操作zookeeper服务器，后来出现了Curator框架，非常的强大，目前已经是Apache的顶级项目，里面提供了更多丰富的操作，例如session超时重连、主从选举、分布式计数器、分布式锁等等适用于各种复杂的zookeeper场景的API封装 |
Curator框架中使用链式编程风格，易读性更强，使用工厂方法创建连接对象。 1.使用CuratorFrameworkFactory的两个静态工厂方法（参数不同）来实现 1.1 connectString：连接串 1.2 retryPolicy：重试连接策略。有四种实现，分别是：ExponentialBackoffRetry、RetryNTimes、RetryOneTimes、RetryUntilElapsed 1.3sessionTimeoutMs：会话超时时间，默认为60000ms 1.4connectionTimeoutMs连接超时时间，默认为15000ms
注意对于retryPolicy策略通过一个接口来让用户自定义实现 |
2.1创建连接
/** 重试策略: 初始时间为1s, 重试10次 */ RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
/** 通过工厂创建连接 */ CuratorFramework cf = CuratorFrameworkFactory.builder() .connectString(ZK_ADDR) .sessionTimeoutMs(SESSION_TIMEOUT) .retryPolicy(retryPolicy) .build();
/** 开启连接 */ cf.start(); |
2.2 新增节点
/** * 新增节点:指定节点类型(不加withMode默认为持久类型节点)、路径、数据内容 * 1.creatingParentsIfNeeded() 递归创建父目录 * 2.withMode() 节点类型(持久|临时) * 3.forPath() 指定路径 */ cf.create() .creatingParentsIfNeeded() .withMode(CreateMode.PERSISTENT) .forPath("/super/c1", "c1内容".getBytes()); |
2.3 删除节点
/** * 删除节点 * 1.deletingChildrenIfNeeded() 递归删除 * 2.guaranteed() 确保节点被删除 * 3. withVersion(int version) //特定版本号 */ cf.delete().deletingChildrenIfNeeded().forPath("/super"); |
2.4 读取和修改数据
/** * 读取和修改数据 : getData()和setData() */ cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c1", "c1内容".getBytes()); cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c2", "c2内容".getBytes());
/** 读取节点内容 */ String c2_data = new String(cf.getData().forPath("/super/c2")); System.out.println("c2_data-->"+c2_data);
/** 修改节点内容 */ cf.setData().forPath("/super/c2", "修改c2的内容".getBytes()); String update_c2_data = new String(cf.getData().forPath("/super/c2")); System.out.println("update_c2_data-->"+update_c2_data); |
2.5 绑定回调函数
ExecutorService pool = Executors.newCachedThreadPool();
cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT) .inBackground(new BackgroundCallback() { @Override public void proce***esult(CuratorFramework cf, CuratorEvent event) throws Exception { System.out.println("code-->" + event.getResultCode()); System.out.println("type-->" + event.getType()); System.out.println("线程为-->" + Thread.currentThread().getName()); } }, pool).forPath("/super/c3", "c2的内容".getBytes());
System.out.println("主线程-->" + Thread.currentThread().getName());
Thread.sleep(Integer.MAX_VALUE); |
2.6 读取子节点和判断节点是否存在
/** * 读取子节点的方法: getChildren() * 判断节点是否存在: checkExists() */ List<String> list = cf.getChildren().forPath("/super"); for (String p: list) { System.out.println(p); }
//如果为null标识不存在 Stat stat = cf.checkExists().forPath("/super/c4"); System.out.println(stat); |
如果要使用类似Wather的监听功能Curator必须依赖一个jar包，Maven依赖 <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>2.4.2</version> </dependency> 有了这个依赖包，使用NodeCache的方式去客户端实例中注册一个监听缓存，然后实现对应的监听方法即可，这里主要有两种监听方式 NodeCacheListener：监听节点的新增、修改操作 PathChildrenCacheListener：监听子节点的新增、修改、删除操作 |
4.1 分布式锁
在分布式场景中，为了保证数据的一致性，经常在程序运行的某一个点需要进行同步操作(java提供了synchronized或者Reentrantlock实现)比如看一个小示例，这个示例出现分布式不同步的问题 比如：之前是在高并发下访问一个程序，现在则是在高并发下访问多个服务器节点（分布式）
使用Curator基于zookeeper的特性提供的分布式锁来处理分布式场景的数据一致性，zookeeper本身的分布式是有写问题的，之前实现的时候遇到过，这里强烈推荐使用Curator分布式锁 public class Lock2 {
/** zk地址 */
private static final String ZK_ADDR = "192.168.1.220:2181,192.168.1.127:2181,192.168.1.128:2181";
/** session超时时间 */
private static final int SESSION_TIMEOUT = 5000; //MS
static int count = 10;
public static CuratorFramework createCuratorFramework(){
CuratorFramework cf = CuratorFrameworkFactory.builder()
.connectString(ZK_ADDR)
.sessionTimeoutMs(SESSION_TIMEOUT)
.retryPolicy(new ExponentialBackoffRetry(1000, 10))
.build();
return cf;
}
public static void main(String[] args) throws Exception {
final CountDownLatch countDown = new CountDownLatch(1);
for (int i =0; i < 10; i++) {
new Thread(new Runnable() {
@Override
public void run() {
CuratorFramework cf = createCuratorFramework();
cf.start();
//锁对象 client 锁节点
final InterProcessMutex lock = new InterProcessMutex(cf, "/super");
try {
countDown.await();
lock.acquire(); //获得锁
number();
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
lock.release();//释放锁
} catch (Exception e) {
e.printStackTrace();
}
}
}
},"t" + i).start();;
}
Thread.sleep(2000);
countDown.countDown();
}
public static void number() {
count--;
System.out.println(Thread.currentThread().getName() + "-->" + count);
}
} |
4.2 分布式计数器功能
一说到分布式计数器，可能脑海里想到AtomicInteger（原子累加）这种经典方式，如果针对一个JVM的场景当然没问题，但是现在是在分布式场景下，就需要利用Curator框架的DistributedAtomicInteger了 public class CuratorAtomicInteger {
/** zk地址 */
private static final String ZK_ADDR = "192.168.1.220:2181,192.168.1.127:2181,192.168.1.128:2181";
/** session超时时间 */
private static final int SESSION_TIMEOUT = 5000; //MS
public static void main(String[] args) throws Exception {
CuratorFramework cf = CuratorFrameworkFactory.builder()
.connectString(ZK_ADDR)
.sessionTimeoutMs(SESSION_TIMEOUT)
.retryPolicy(new ExponentialBackoffRetry(1000, 10))
.build();
cf.start();
//使用DistributedAtomicInteger
DistributedAtomicInteger atomicInteger = new DistributedAtomicInteger(cf, "/superM", new RetryNTimes(3, 1000));
//atomicInteger.increment();
atomicInteger.add(1);
AtomicValue<Integer> atomicValue = atomicInteger.get();
System.out.println("atomicValue.succeeded()-->" + atomicValue.succeeded());
System.out.println("atomicValue.postValue()-->" + atomicValue.postValue());
System.out.println("atomicValue.preValue()-->" + atomicValue.preValue());
}
} |
4.3 Barrier
4.3.1 DistributedDoubleBarrier
分布式Barrier 类DistributedDoubleBarrier： 它会阻塞所有节点上的等待进程，直到某一个被满足， 然后所有的节点同时开始，中间谁先运行完毕，谁后运行完毕不关心，但是最终一定是一块退出运行的
public class CuratorBarrier {
/** zk地址 */
private static final String ZK_ADDR = "192.168.1.220:2181,192.168.1.127:2181,192.168.1.128:2181";
/** session超时时间 */
private static final int SESSION_TIMEOUT = 5000; //MS
public static void main(String[] args) throws Exception{
for (int i =0; i < 5; i++) {
new Thread(new Runnable() {
@Override
public void run() {
try {
/** 实例化5个客户端对象 */
CuratorFramework cf = CuratorFrameworkFactory.builder()
.connectString(ZK_ADDR)
.sessionTimeoutMs(SESSION_TIMEOUT)
.retryPolicy(new ExponentialBackoffRetry(1000, 10))
.build();
cf.start();
DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(cf, "/superBarrier", 5);
Thread.sleep(1000 * (new Random()).nextInt(3));
System.out.println(Thread.currentThread().getName() + " 已准备好！");
barrier.enter();
System.out.println("同时开始运行...");
Thread.sleep(1000 * (new Random()).nextInt(3));
System.out.println("运行完毕...");
barrier.leave();
System.out.println("同时退出运行...");
} catch (Exception e) {
e.printStackTrace();
}
}
},"t" + i).start();;
}
}
} |
4.3.2 DistributedBarrier
分布式Barrier 类DistributedBarrier： 它会阻塞所有节点上的等待进程（所有节点进入待执行状态），直到“某一个人吹哨”说开始执行， 然后所有的节点同时开始 public class CuratorBarrier2 {
/** zk地址 */
private static final String ZK_ADDR = "192.168.1.220:2181,192.168.1.127:2181,192.168.1.128:2181";
/** session超时时间 */
private static final int SESSION_TIMEOUT = 5000; //MS
static DistributedBarrier barrier = null;
public static void main(String[] args) throws Exception{
for (int i =0; i < 5; i++) {
new Thread(new Runnable() {
@Override
public void run() {
try {
/** 实例化5个客户端对象 */
CuratorFramework cf = CuratorFrameworkFactory.builder()
.connectString(ZK_ADDR)
.sessionTimeoutMs(SESSION_TIMEOUT)
.retryPolicy(new ExponentialBackoffRetry(1000, 10))
.build();
cf.start();
barrier = new DistributedBarrier(cf, "/superBarrier");
System.out.println(Thread.currentThread().getName() + " 设置barrier");
barrier.setBarrier(); //设置
barrier.waitOnBarrier(); //等待
System.out.println("开始执行程序...");
} catch (Exception e) {
e.printStackTrace();
}
}
},"t" + i).start();;
}
Thread.sleep(5000);
barrier.removeBarrier(); //释放
}
} |
Curator内部实现的几种重试策略: 1.ExponentialBackoffRetry:重试指定的次数, 且每一次重试之间停顿的时间逐渐增加. 2.RetryNTimes:指定最大重试次数的重试策略 3.RetryOneTime:仅重试一次 4.RetryUntilElapsed:一直重试直到达到规定的时间 |
5.1 ExponentialBackoffRetry
ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries) ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs)
参数说明 1.baseSleepTimeMs 初始sleep时间 2.maxRetries 最大重试次数 3.maxSleepMs 最大重试时间 |
5.2 RetryNTimes
RetryNTimes(int n, int sleepMsBetweenRetries) 参数说明 1.n 最大重试次数 2.sleepMsBetweenRetries 每次重试的间隔时间 |
5.3 RetryOneTime
RetryOneTime(int sleepMsBetweenRetry)
参数说明 1.sleepMsBetweenRetry为重试间隔的时间 |
5.4 RetryUntilElapsed
RetryUntilElapsed(int maxElapsedTimeMs, int sleepMsBetweenRetries)
参数说明 1.maxElapsedTimeMs 最大重试时间 2.sleepMsBetweenRetries 每次重试的间隔时间 |
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
