springboot整合curator實現分布式鎖

理論篇：

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

自定義注解：

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface DistributedLock{
   String lockPath();  
}

 

注意事項：

   1.  CuratorFramework對象建議在應用中做單例處理，在具體使用處 注入使用, 並在應用結束前銷毀，代碼如下：

@Configration
public class CuratorConfigration{
    @Bean    
    public CuratorFramework initCuratorFramework(){
        //忽略 
       // 參照前面 CuratorFramework 對象創建部分
    }    
}

2. 在aspect部分將curatorFramework對象進行關閉

@PreDestroy
public void destroy(){
   CloseableUtils.closeQuietly(curatorFramework);
}