理论篇：
Curator是Netflix开源的一套ZooKeeper客户端框架. Netflix在使用ZooKeeper的过程中发现ZooKeeper自带的客户端太底层, 应用方在使用的时候需要自己处理很多事情, 于是在它的基础上包装了一下, 提供了一套更好用的客户端框架. Netflix在用ZooKeeper的过程中遇到的问题, 我们也遇到了, 所以开始研究一下, 首先从他在github上的源码, wiki文档以及Netflix的技术blog入手.
看完官方的文档之后, 发现Curator主要解决了三类问题:
- 封装ZooKeeper client与ZooKeeper server之间的连接处理;
- 提供了一套Fluent风格的操作API;
- 提供ZooKeeper各种应用场景(recipe, 比如共享锁服务, 集群领导选举机制)的抽象封装.
Curator列举的ZooKeeper使用过程中的几个问题
初始化连接的问题: 在client与server之间握手建立连接的过程中, 如果握手失败, 执行所有的同步方法(比如create, getData等)将抛出异常
自动恢复(failover)的问题: 当client与一台server的连接丢失,并试图去连接另外一台server时, client将回到初始连接模式
session过期的问题: 在极端情况下, 出现ZooKeeper session过期, 客户端需要自己去监听该状态并重新创建ZooKeeper实例 .
对可恢复异常的处理:当在server端创建一个有序ZNode, 而在将节点名返回给客户端时崩溃, 此时client端抛出可恢复的异常, 用户需要自己捕获这些异常并进行重试
使用场景的问题:Zookeeper提供了一些标准的使用场景支持, 但是ZooKeeper对这些功能的使用说明文档很少, 而且很容易用错. 在一些极端场景下如何处理, zk并没有给出详细的文档说明. 比如共享锁服务, 当服务器端创建临时顺序节点成功, 但是在客户端接收到节点名之前挂掉了, 如果不能很好的处理这种情况, 将导致死锁.
Curator主要从以下几个方面降低了zk使用的复杂性:
重试机制:提供可插拔的重试机制, 它将给捕获所有可恢复的异常配置一个重试策略, 并且内部也提供了几种标准的重试策略(比如指数补偿).
连接状态监控: Curator初始化之后会一直的对zk连接进行监听, 一旦发现连接状态发生变化, 将作出相应的处理.
zk客户端实例管理:Curator对zk客户端到server集群连接进行管理. 并在需要的情况, 重建zk实例, 保证与zk集群的可靠连接
各种使用场景支持:Curator实现zk支持的大部分使用场景支持(甚至包括zk自身不支持的场景), 这些实现都遵循了zk的最佳实践, 并考虑了各种极端情况.
Curator通过以上的处理, 让用户专注于自身的业务本身, 而无需花费更多的精力在zk本身.
CuratorFrameworkFactory类提供了两个方法, 一个工厂方法newClient, 一个构建方法build. 使用工厂方法newClient可以创建一个默认的实例, 而build构建方法可以对实例进行定制. 当CuratorFramework实例构建完成, 紧接着调用start()方法, 在应用结束的时候, 需要调用close()方法. CuratorFramework是线程安全的. 在一个应用中可以共享同一个zk集群的CuratorFramework.
核心对象CuratorFramework的创建如下：
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString("")
.sessionTimeoutMs(5000)
.connectionTimeoutMs(5000)
.retryPolicy(retryPolicy)
.build();
client.start();
需要使用分布式锁的地方，代码如下：
String lockOn= "test"; InterProcessMutex mutex = new InterProcessMutex(curatorFramework,lockOn); boolean locked =mutex.acquire(0,TimeUnit.SECONDS); //finally部分 mutex.release();
分布式锁常用于定时任务，使用自定义注解，使用spring aspect around， 在真正的代码执行之前尝试获取锁，获取不到直接退出，获取到锁的，执行具体业务，代码如下：
@Aspect
public class DistributedLockAspect{
@Pointcut("@annotation(com.**.**.DistributedLock")
public void methodAspect(){};
@Around("methodAspect()")
public Object execute(ProceedingJoinPoint joinPoint) throws Exception{
String lockPath = "/opt/zookeeper/lock";
InterProcessMutex mutex = new InterProcessMutex(cruatorFramework,lockPath);
try{
boolean locked = mutex.acquire(0,TimeUnit.SECONDS);
if(!locked){
return null;
}else{
return joinPoint.proceed();
}
}catch(Exception e){
e.printStackTrace();
}finally{
mutex.release();
}
}
}
自定义注解：
1 @Target(ElementType.METHOD)
2 @Retention(RetentionPolicy.RUNTIME)
3 public @interface DistributedLock{
4 String lockPath();
5 }
注意事项：
1. CuratorFramework对象建议在应用中做单例处理，在具体使用处 注入使用, 并在应用结束前销毁，代码如下：
@Configration
public class CuratorConfigration{
@Bean
public CuratorFramework initCuratorFramework(){
//忽略
// 参照前面 CuratorFramework 对象创建部分
}
}
2. 在aspect部分将curatorFramework对象进行关闭
@PreDestroy
public void destroy(){
CloseableUtils.closeQuietly(curatorFramework);
}
原文地址：https://www.cnblogs.com/xifenglou/p/10455553.html