Redisson分布式鎖在Java應用中的使用

一、引入Redisson依賴，並配置相關的Bean

a. Spring 應用

通過Maven引入依賴

<dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson</artifactId>
   <version>3.14.0</version>
</dependency>

配置相關的Bean

創建配置類的Bean：

Config config = new Config();
config.useClusterServers()
       // use "rediss://" for SSL connection
      .addNodeAddress("redis://127.0.0.1:7181");

創建 Redisson 實例：

// 同步與異步API
RedissonClient redisson = Redisson.create(config);

此外還有Reactive API和RxJava2 API相關的客戶端，具體查看Redisson在GitHub上的說明。

b. Spring Boot 應用

通過Maven引入依賴

<dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson-spring-boot-starter</artifactId>
   <version>3.14.0</version>
</dependency>

注意，如果通過其他方式引入了redisson-spring-data模塊，則需要根據Spring Boot的版本，調整redisson-spring-data的版本，具體的版本適配見這里。

在application.properties中添加配置

基本的Redis配置（其中host、port、password必須配置）：

spring:
  redis:
    database: 
    host:
    port:
    password:
    ssl: 
    timeout:
    cluster:
      nodes:
    sentinel:
      master:
      nodes:

根據需要添加Redisson相關的配置：

  # path to config - redisson.yaml
  redisson: 
    file: classpath:redisson.yaml
    config:
      clusterServersConfig:
        idleConnectionTimeout: 10000
        connectTimeout: 10000
        timeout: 3000
        retryAttempts: 3
        retryInterval: 1500
        failedSlaveReconnectionInterval: 3000
        failedSlaveCheckInterval: 60000
        password: null
        subscriptionsPerConnection: 5
        clientName: null
        loadBalancer: !<org.redisson.connection.balancer.RoundRobinLoadBalancer> {}
        subscriptionConnectionMinimumIdleSize: 1
        subscriptionConnectionPoolSize: 50
        slaveConnectionMinimumIdleSize: 24
        slaveConnectionPoolSize: 64
        masterConnectionMinimumIdleSize: 24
        masterConnectionPoolSize: 64
        readMode: "SLAVE"
        subscriptionMode: "SLAVE"
        nodeAddresses:
        - "redis://127.0.0.1:7004"
        - "redis://127.0.0.1:7001"
        - "redis://127.0.0.1:7000"
        scanInterval: 1000
        pingConnectionInterval: 0
        keepAlive: false
        tcpNoDelay: false
      threads: 16
      nettyThreads: 32
      codec: !<org.redisson.codec.FstCodec> {}
      transportMode: "NIO"

獲取並使用Redisson客戶端

通過上述配置，就可以在代碼中通過自動裝配，直接獲取並使用RedissonClient 或RedisTemplate/ReactiveRedisTemplate 等Bean了。

二、使用Redisson分布式鎖

用鎖的一般步驟：

1. 獲取鎖實例（只是獲得一把鎖的引用，並不是占有鎖）
2. 通過鎖實例加鎖（占有了這把鎖）
3. 通過鎖實例釋放鎖

Redisson提供很多種類型的鎖，其中最常用的就是可重入鎖（Reentrant Lock）了。

Redisson中的可重入鎖

1. 獲取鎖實例

RLock lock = redissonClient.getLock(String lockName);

獲取的鎖實例實現了RLock接口，而該接口擴展了JUC包中的Lock接口，以及異步鎖接口RLockAsync。

2. 通過鎖實例加鎖

從同步與異步特性來區分，加鎖方法可分為同步加鎖和異步加鎖兩類。異步加鎖方法的名稱一般是在相應的同步加鎖方法后加上“Async”后綴。
從阻塞與非阻塞特性來區分，加鎖方法可分為阻塞加鎖和非阻塞加鎖兩類。非阻塞加鎖方法的名稱一般是“try”開頭。

下面以比較常用的同步加鎖方法來說明加鎖的一些細節。

阻塞加鎖的方法：

1. void lock(): （JUC中Lock接口定義的方法）如果當前鎖可用，則加鎖成功，並立即返回；如果當前鎖不可用，則阻塞等待直至鎖可用，然后返回。
2. void lock(long leaseTime, TimeUnit unit): 加鎖機制與void lock()相同，只是增加了鎖的有效（租賃）時長leaseTime。加鎖成功后，可以在程序中顯式調用unlock()方法進行釋放；如果未顯式釋放，則經過leaseTime時間，該鎖會自動釋放。如果leaseTime傳入-1，則會一直持有，直至調用unlock()。

非阻塞加鎖的方法：

1. boolean tryLock(): （JUC中Lock接口定義的方法）調用該方法會立刻返回。返回值為true則表示鎖可用，加鎖成功；返回值為false則表示鎖不可用，加鎖失敗。
2. boolean tryLock(long time, TimeUnit unit): （JUC中Lock接口定義的方法）如果鎖可用則立刻返回true，否則最多等待time長的時間（如果time<=0，則不會等待）。在time時間內鎖可用則立刻返回true，time時間之后返回false。如果在等待期間線程被其他線程中斷，則會拋出nterruptedException 異常。
3. boolean tryLock(long waitTime, long leaseTime, TimeUnit unit): 與boolean tryLock(long time, TimeUnit unit)類似，只是增加了鎖的使用（租賃）時長leaseTime。

3. 通過鎖實例釋放鎖

1. void unlock(): 釋放鎖。如果當前線程是鎖的持有者（即在該鎖實例上加鎖成功的線程），則會釋放成功，否則會拋出異常。

一般編程范式

同步阻塞加鎖

String lockName = ...
RLock lock = redissonClient.getLock(lockName);
// 阻塞式加鎖
lock.lock();
try {
    // 操作受鎖保護的資源

} finally {
    // 釋放鎖
    lock.unlock();
}

同步非阻塞加鎖

String lockName = ...
RLock lock = redissonClient.getLock(lockName);
if (lock.tryLock()) {
    try {
        // 操作受鎖保護的資源
    } finally {
        lock.unlock();
    }
} else {
    // 執行其他業務操作
}

三、分布式鎖分析

優秀的分布式鎖需要具備以下特性：

• 互斥性：在任意時刻，只有一個客戶端（線程）能持有鎖，這是鎖的基本要求。
• 鎖的可重入：同一個客戶端能多次持有同一把鎖。實現上只要檢查鎖的持有者是否為當前客戶端，若是則重入鎖成功，並將鎖的持有數加1。一般通過給每個客戶端分配一個唯一的ID，並在加鎖成功時向鎖中寫入該ID即可。
• 不會因客戶端異常而長久鎖住：當客戶端在持有鎖期間崩潰而未主動解鎖時，鎖也會在一定時間后自動釋放，即鎖有超時自動釋放的特性。
• 解鎖的安全性：加鎖和解鎖必須是同一個客戶端，客戶端不能把別人加的鎖給釋放了，即不能誤解鎖。實現上與鎖的可重入類似，在釋放鎖時檢查客戶端ID與鎖中保存的ID是否一致即可。

Redisson的分布式鎖除了實現上述幾個特性外，還具有鎖的自動續期功能。即當我們加鎖而未指定鎖的有效時長時，Redisson會按一定的周期，定時檢查當前線程是否活躍，若是則自動為鎖續期，這一特性稱為watchdog（看門狗）機制。
有了這個特性，我們就可以不必為設定鎖的有效時間而糾結了（設得太長，則會在客戶端崩潰后仍長時間占有鎖；設得太短，則可能在業務邏輯執行完成前，鎖自動釋放），Redisson分布式鎖可以在客戶端崩壞時自動釋放，業務邏輯未執行完時自動續期。

用 Redis 實現分布式鎖的最佳實踐

假設沒有Redisson，需要我們自己用Redis實現分布式鎖，以下是一些不錯的編程實踐：

• set 命令要用set key value px milliseconds nx，替代 setnx + expire 需要分兩次執行命令的方式，保證原子性。
• 客戶端的ID一般保存在線程本地變量（ThreadLocal）中。客戶端ID要求全局唯一，可以考慮【IP+線程ID】組合，或者用UUID。
• 阻塞式子加鎖可用Object對象的wait+notify機制實現。
• 釋放鎖時，需要檢查當前客戶端是否為鎖的持有者，因此有compare and set的邏輯。為了保證兩個操作的原子性，用單個Lua腳本來執行多個Redis操作（利用了eval命令執行Lua腳本的原子性，參考這里及這里）。另外，如果鎖有自動續期的定時任務，在解鎖的時候需要停掉該任務。

其他主題（TODO）

• 鎖的高可用（容錯性）：只要大多數Redis節點正常運行，客戶端就能夠獲取和釋放鎖（參考這里）。
• 死鎖問題：當因客戶端編程邏輯問題導致兩個線程死鎖時，如何檢測並解決死鎖問題？
• Redisson中其他類型的鎖（參考這里）
• 鎖的續期機制分析（參考這里）
• Redisson 支持4種鏈接redis的方式：Cluster(集群)、Sentinel servers(哨兵)、Master/Slave servers(主從)、Single server(單機)（參考這里）
• 自己動手實現注解式加鎖（參考這里）

參考文檔

1. Redisson的GitHub倉庫
2. redisson-spring-boot-starter 使用說明
3. Redisson: Distributed locks and synchronizers
4. JDK & Redisson 中的源碼注釋
5. 慢談 Redis 實現分布式鎖 以及 Redisson 源碼解析
6. 使用Redisson實現分布式鎖
7. Redisson 實現分布式鎖原理分析（對Redisson的源碼進行分析）
8. Redis官方文檔：[用Redis構建分布式鎖]（http://ifeve.com/redis-lock/）
9. Redisson分布式鎖實現
10. redisson實現分布式鎖原理