分布式锁-curator实现

Posted 2021-02-11 xifenglou

理论篇：

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);
}