Redisson分佈式鎖淺析

針對項目中使用的分佈式鎖進行簡單的示例配置以及源碼解析，並列舉源碼中使用到的一些基礎知識點，可是沒有對redisson中使用到的netty知識進行解析。

本篇主要是對如下幾個方面進行了探索

• Maven配置
• RedissonLock簡單示例
• 源碼中使用到的Redis命令
• 源碼中使用到的lua腳本語義
• 源碼分析

Maven配置

1 2 3 4 5 6 7 8 9 10

<dependency>     <groupId>org.redisson</groupId>     <artifactId>redisson</artifactId>     <version>2.2.12</version> </dependency> <dependency>     <groupId>com.fasterxml.jackson.core</groupId>     <artifactId>jackson-annotations</artifactId>     <version>2.6.0</version> </dependency>

RedissonLock簡單示例

redission支持4種鏈接redis方式，分別爲單機、主從、Sentinel、Cluster 集羣，項目中使用的鏈接方式是Sentinel。
redis服務器不在本地的同窗請注意權限問題。

Sentinel配置

1 2 3

Config config = new Config(); config.useSentinelServers().addSentinelAddress("127.0.0.1:6479", "127.0.0.1:6489").setMasterName("master").setPassword("password").setDatabase(0); RedissonClient redisson = Redisson.create(config);

簡單使用

1 2 3 4 5 6 7 8 9 10

RLock lock = redisson.getLock("test_lock"); try{     boolean isLock=lock.tryLock();     if(isLock){         doBusiness();     } }catch(exception e){ }finally{     lock.unlock(); }

源碼中使用到的Redis命令

分佈式鎖主要須要如下redis命令，這裏列舉一下。在源碼分析部分能夠繼續參照命令的操做含義。

1. EXISTS key :當 key 存在，返回1；若給定的 key 不存在，返回0。
2. GETSET key value:將給定 key 的值設爲 value ，並返回 key 的舊值 (old value)，當 key 存在但不是字符串類型時，返回一個錯誤，當key不存在時，返回nil。
3. GET key:返回 key 所關聯的字符串值，若是 key 不存在那麼返回 nil。
4. DEL key [KEY …]:刪除給定的一個或多個 key ,不存在的 key 會被忽略,返回實際刪除的key的個數（integer）。
5. HSET key field value：給一個key 設置一個{field=value}的組合值，若是key沒有就直接賦值並返回1，若是field已有，那麼就更新value的值，並返回0.
6. HEXISTS key field:當key中存儲着field的時候返回1，若是key或者field至少有一個不存在返回0。
7. HINCRBY key field increment:將存儲在key中的哈希（Hash）對象中的指定字段field的值加上增量increment。若是鍵key不存在，一個保存了哈希對象的新建將被建立。若是字段field不存在，在進行當前操做前，其將被建立，且對應的值被置爲0，返回值是增量以後的值
8. PEXPIRE key milliseconds：設置存活時間，單位是毫秒。expire操做單位是秒。
9. PUBLISH channel message:向channel post一個message內容的消息，返回接收消息的客戶端數。

源碼中使用到的lua腳本語義

Redisson源碼中，執行redis命令的是lua腳本，其中主要用到以下幾個概念。

• redis.call() 是執行redis命令.
• KEYS[1] 是指腳本中第1個參數
• ARGV[1] 是指腳本中第一個參數的值
• 返回值中nil與false同一個意思。

須要注意的是，在redis執行lua腳本時，至關於一個redis級別的鎖，不能執行其餘操做，相似於原子操做，也是redisson實現的一個關鍵點。
另外，若是lua腳本執行過程當中出現了異常或者redis服務器直接宕掉了，執行redis的根據日誌回覆的命令，會將腳本中已經執行的命令在日誌中刪除。

源碼分析

RLOCK結構

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

public interface RLock extends Lock, RExpirable {     void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException;     boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;     void lock(long leaseTime, TimeUnit unit);     void forceUnlock();     boolean isLocked();     boolean isHeldByCurrentThread();     int getHoldCount();     Future<Void> unlockAsync();     Future<Boolean> tryLockAsync();     Future<Void> lockAsync();     Future<Void> lockAsync(long leaseTime, TimeUnit unit);     Future<Boolean> tryLockAsync(long waitTime, TimeUnit unit);     Future<Boolean> tryLockAsync(long waitTime, long leaseTime, TimeUnit unit); }

該接口主要繼承了Lock接口, 並擴展了部分方法, 好比:boolean tryLock(long waitTime, long leaseTime, TimeUnit unit)新加入的leaseTime主要是用來設置鎖的過時時間, 若是超過leaseTime尚未解鎖的話, redis就強制解鎖. leaseTime的默認時間是30s

RedissonLock獲取鎖 tryLock源碼

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

Future<Long> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId) {        internalLockLeaseTime = unit.toMillis(leaseTime);        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,                  "if (redis.call('exists', KEYS[1]) == 0) then " +                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +                      "return nil; " +                  "end; " +                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +                      "return nil; " +                  "end; " +                  "return redis.call('pttl', KEYS[1]);",                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));    }

其中
KEYS[1] 表示的是 getName() ，表明的是鎖名 test_lock
ARGV[1] 表示的是 internalLockLeaseTime 默認值是30s
ARGV[2] 表示的是 getLockName(threadId) 表明的是 id:threadId 用鎖對象id+線程id， 表示當前訪問線程，用於區分不一樣服務器上的線程.
逐句分析：

1 2 3 4 5

if (redis.call('exists', KEYS[1]) == 0) then          redis.call('hset', KEYS[1], ARGV[2], 1);          redis.call('pexpire', KEYS[1], ARGV[1]);          return nil;          end;

• if (redis.call(‘exists’, KEYS[1]) == 0) 若是鎖名稱不存在
• then redis.call(‘hset’, KEYS[1], ARGV[2],1) 則向redis中添加一個key爲test_lock的set，而且向set中添加一個field爲線程id，值=1的鍵值對，表示此線程的重入次數爲1
• redis.call(‘pexpire’, KEYS[1], ARGV[1]) 設置set的過時時間，防止當前服務器出問題後致使死鎖，return nil; end;返回nil 結束

1 2 3 4 5

if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then          redis.call('hincrby', KEYS[1], ARGV[2], 1);          redis.call('pexpire', KEYS[1], ARGV[1]);          return nil;          end;

• if (redis.call(‘hexists’, KEYS[1], ARGV[2]) == 1) 若是鎖是存在的，檢測是不是當前線程持有鎖，若是是當前線程持有鎖
• then redis.call(‘hincrby’, KEYS[1], ARGV[2], 1)則將該線程重入的次數++
• redis.call(‘pexpire’, KEYS[1], ARGV[1]) 而且從新設置該鎖的有效時間
• return nil; end;返回nil，結束

1	return redis.call('pttl', KEYS[1]);

• 鎖存在, 但不是當前線程加的鎖，則返回鎖的過時時間

RedissonLock解鎖 unlock源碼

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

@Override     public void unlock() {         Boolean opStatus = commandExecutor.evalWrite(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,                         "if (redis.call('exists', KEYS[1]) == 0) then " +                             "redis.call('publish', KEYS[2], ARGV[1]); " +                             "return 1; " +                         "end;" +                         "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +                             "return nil;" +                         "end; " +                         "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +                         "if (counter > 0) then " +                             "redis.call('pexpire', KEYS[1], ARGV[2]); " +                             "return 0; " +                         "else " +                             "redis.call('del', KEYS[1]); " +                             "redis.call('publish', KEYS[2], ARGV[1]); " +                             "return 1; "+                         "end; " +                         "return nil;",                         Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(Thread.currentThread().getId()));         if (opStatus == null) {             throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "                     + id + " thread-id: " + Thread.currentThread().getId());         }         if (opStatus) {             cancelExpirationRenewal();         }     }

其中
KEYS[1] 表是的是getName() 表明鎖名test_lock
KEYS[2] 表示getChanelName() 表示的是發佈訂閱過程當中使用的Chanel
ARGV[1] 表示的是LockPubSub.unLockMessage 是解鎖消息，實際表明的是數字 0，表明解鎖消息
ARGV[2] 表示的是internalLockLeaseTime 默認的有效時間 30s
ARGV[3] 表示的是getLockName(thread.currentThread().getId())，是當前鎖id+線程id
語義分析:

1 2 3 4

if (redis.call('exists', KEYS[1]) == 0) then          redis.call('publish', KEYS[2], ARGV[1]);          return 1;          end;

• if (redis.call(‘exists’, KEYS[1]) == 0) 若是鎖已經不存在(多是由於過時致使不存在，也多是由於已經解鎖)
• then redis.call(‘publish’, KEYS[2], ARGV[1]) 則發佈鎖解除的消息
• return 1; end 返回1結束

1 2 3

if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then          return nil;          end;

• if (redis.call(‘hexists’, KEYS[1], ARGV[3]) == 0) 若是鎖存在，可是若果當前線程不是加鎖的線
• then return nil;end 則直接返回nil 結束

1 2 3 4 5 6 7 8 9

local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); if (counter > 0) then          redis.call('pexpire', KEYS[1], ARGV[2]);          return 0; else          redis.call('del', KEYS[1]);          redis.call('publish', KEYS[2], ARGV[1]);          return 1; end;

• local counter = redis.call(‘hincrby’, KEYS[1], ARGV[3], -1) 若是是鎖是當前線程所添加，定義變量counter，表示當前線程的重入次數-1,即直接將重入次數-1
• if (counter > 0)若是重入次數大於0，表示該線程還有其餘任務須要執行
• then redis.call(‘pexpire’, KEYS[1], ARGV[2]) 則從新設置該鎖的有效時間
• return 0 返回0結束
• else redis.call(‘del’, KEYS[1]) 不然表示該線程執行結束，刪除該鎖
• redis.call(‘publish’, KEYS[2], ARGV[1]) 而且發佈該鎖解除的消息
• return 1; end;返回1結束

1	return nil;

• 其餘狀況返回nil並結束

1 2 3 4

if (opStatus == null) {             throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "                     + id + " thread-id: " + Thread.currentThread().getId());         }

• 腳本執行結束以後，若是返回值不是0或1，即當前線程去解鎖其餘線程的加鎖時，拋出異常

RedissonLock強制解鎖源碼

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

@Override     public void forceUnlock() {         get(forceUnlockAsync());     }     Future<Boolean> forceUnlockAsync() {         cancelExpirationRenewal();         return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,                 "if (redis.call('del', KEYS[1]) == 1) then "                 + "redis.call('publish', KEYS[2], ARGV[1]); "                 + "return 1 "                 + "else "                 + "return 0 "                 + "end",                 Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage);     }

• 以上是強制解鎖的源碼,在源碼中並無找到forceUnlock()被調用的痕跡(也有多是我沒有找對),可是forceUnlockAsync()方法被調用的地方不少，大多都是在清理資源時刪除鎖。此部分比較簡單粗暴，刪除鎖成功則併發布鎖被刪除的消息，返回1結束，不然返回0結束。

總結

這裏只是簡單的一個redisson分佈式鎖的測試用例，並分析了執行lua腳本這部分，若是要繼續分析執行結束以後的操做，須要進行netty源碼分析 ，redisson使用了netty完成異步和同步的處理。