Redis之分散式鎖實現
點贊再看，養成習慣，微信搜尋【三太子敖丙】關注這個網際網路苟且偷生的工具人。
本文 GitHub https://github.com/JavaFamily 已收錄，有一線大廠面試完整考點、資料以及我的系列文章。
前言
上一章節我提到了基於zk分散式鎖的實現，這章節就來說一下基於Redis的分散式鎖實現吧。
zk實現分散式鎖的傳送門：zk分散式鎖
在開始提到Redis分散式鎖之前，我想跟大家聊點Redis的基礎知識。
說一下Redis的兩個命令：
SETNX key value
setnx 是SET if Not eXists(如果不存在，則 SET)的簡寫。
用法如圖，如果不存在set成功返回int的1，這個key存在了返回0。
SETEX key seconds value
將值 value 關聯到 key ，並將 key 的生存時間設為 seconds (以秒為單位)。
如果 key 已經存在，setex命令將覆寫舊值。
有小夥伴肯定會疑惑萬一set value 成功 set time失敗，那不就傻了麼，這啊Redis官網想到了。
setex是一個原子性(atomic)操作，關聯值和設定生存時間兩個動作會在同一時間內完成。
我設定了10秒的失效時間，ttl命令可以檢視倒計時，負的說明已經到期了。
跟大家講這兩個命名也是有原因的，因為他們是Redis實現分散式鎖的關鍵。
正文
開始前還是看看場景:
我依然是建立了很多個執行緒去扣減庫存inventory，不出意外的庫存扣減順序變了，最終的結果也是不對的。
單機加synchronized或者Lock這些常規操作我就不說了好吧，結果肯定是對的。
我先實現一個簡單的Redis鎖，然後我們再實現分散式鎖，可能更方便大家的理解。
還記得上面我說過的命令麼，實現一個單機的其實比較簡單，你們先思考一下，別往下看。
setnx
可以看到，第一個成功了，沒釋放鎖，後面的都失敗了，至少順序問題問題是解決了，只要加鎖，縮放後面的拿到，釋放如此迴圈，就能保證按照順序執行。
但是你們也發現問題了，還是一樣的，第一個仔set成功了，但是突然掛了，那鎖就一直在那無法得到釋放，後面的執行緒也永遠得不到鎖，又死鎖了。
所以....
setex
知道我之前說這個命令的原因了吧，設定一個過期時間，就算執行緒1掛了，也會在失效時間到了，自動釋放。
我這裡就用到了nx和px的結合引數，就是set值並且加了過期時間，這裡我還設定了一個過期時間，就是這時間內如果第二個沒拿到第一個的鎖，就退出阻塞了，因為可能是客戶端斷連了。
加鎖
整體加鎖的邏輯比較簡單，大家基本上都能看懂，不過我拿到當前時間去減開始時間的操作感覺有點笨， System.currentTimeMillis()消耗很大的。
/**
* 加鎖
*
* @param id
* @return
*/
public boolean lock(String id) {
Long start = System.currentTimeMillis();
try {
for (; ; ) {
//SET命令返回OK ，則證明獲取鎖成功
String lock = jedis.set(LOCK_KEY, id, params);
if ("OK".equals(lock)) {
return true;
}
//否則迴圈等待，在timeout時間內仍未獲取到鎖，則獲取失敗
long l = System.currentTimeMillis() - start;
if (l >= timeout) {
return false;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} finally {
jedis.close();
}
}
System.currentTimeMillis消耗大，每個執行緒進來都這樣，我之前寫程式碼，就會在伺服器啟動的時候，開一個執行緒不斷去拿，呼叫方直接獲取值就好了，不過也不是最優解，日期類還是有很多好方法的。
@Service
public class TimeServcie {
private static long time;
static {
new Thread(new Runnable(){
@Override
public void run() {
while (true){
try {
Thread.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
long cur = System.currentTimeMillis();
setTime(cur);
}
}
}).start();
}
public static long getTime() {
return time;
}
public static void setTime(long time) {
TimeServcie.time = time;
}
}
解鎖
解鎖的邏輯更加簡單，就是一段Lua的拼裝，把Key做了刪除。
你們發現沒，我上面加鎖解鎖都用了UUID，這就是為了保證，誰加鎖了誰解鎖，要是你刪掉了我的鎖，那不亂套了嘛。
LUA是原子性的，也比較簡單，就是判斷一下Key和我們引數是否相等，是的話就刪除，返回成功1，0就是失敗。
/**
* 解鎖
*
* @param id
* @return
*/
public boolean unlock(String id) {
String script =
"if redis.call('get',KEYS[1]) == ARGV[1] then" +
" return redis.call('del',KEYS[1]) " +
"else" +
" return 0 " +
"end";
try {
String result = jedis.eval(script, Collections.singletonList(LOCK_KEY), Collections.singletonList(id)).toString();
return "1".equals(result) ? true : false;
} finally {
jedis.close();
}
}
驗證
我們可以用我們寫的Redis鎖試試效果，可以看到都按照順序去執行了
思考
大家是不是覺得完美了，但是上面的鎖，有不少瑕疵的，我沒思考很多點，你或許可以思考一下，原始碼我都開源到我的GItHub了。
而且，鎖一般都是需要可重入行的，上面的執行緒都是執行完了就釋放了，無法再次進入了，進去也是重新加鎖了，對於一個鎖的設計來說肯定不是很合理的。
我不打算手寫，因為都有現成的，別人幫我們寫好了。
redisson
redisson的鎖，就實現了可重入了，但是他的原始碼比較晦澀難懂。
使用起來很簡單，因為他們底層都封裝好了，你連線上你的Redis客戶端，他幫你做了我上面寫的一切，然後更完美。
簡單看看他的使用吧，跟正常使用Lock沒啥區別。
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(inventory, inventory, 10L, SECONDS, linkedBlockingQueue);
long start = System.currentTimeMillis();
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
final RedissonClient client = Redisson.create(config);
final RLock lock = client.getLock("lock1");
for (int i = 0; i <= NUM; i++) {
threadPoolExecutor.execute(new Runnable() {
public void run() {
lock.lock();
inventory--;
System.out.println(inventory);
lock.unlock();
}
});
}
long end = System.currentTimeMillis();
System.out.println("執行執行緒數:" + NUM + " 總耗時:" + (end - start) + " 庫存數為:" + inventory);
上面可以看到我用到了getLock，其實就是獲取一個鎖的例項。
RedissionLock也沒做啥，就是熟悉的初始化。
public RLock getLock(String name) {
return new RedissonLock(connectionManager.getCommandExecutor(), name);
}
public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
super(commandExecutor, name);
//命令執行器
this.commandExecutor = commandExecutor;
//UUID字串
this.id = commandExecutor.getConnectionManager().getId();
//內部鎖過期時間
this.internalLockLeaseTime = commandExecutor.
getConnectionManager().getCfg().getLockWatchdogTimeout();
this.entryName = id + ":" + name;
}
加鎖
有沒有發現很多跟Lock很多相似的地方呢？
嘗試加鎖，拿到當前執行緒，然後我開頭說的ttl也看到了，是不是一切都是那麼熟悉？
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
//當前執行緒ID
long threadId = Thread.currentThread().getId();
//嘗試獲取鎖
Long ttl = tryAcquire(leaseTime, unit, threadId);
// 如果ttl為空，則證明獲取鎖成功
if (ttl == null) {
return;
}
//如果獲取鎖失敗，則訂閱到對應這個鎖的channel
RFuture<RedissonLockEntry> future = subscribe(threadId);
commandExecutor.syncSubscription(future);
try {
while (true) {
//再次嘗試獲取鎖
ttl = tryAcquire(leaseTime, unit, threadId);
//ttl為空，說明成功獲取鎖，返回
if (ttl == null) {
break;
}
//ttl大於0 則等待ttl時間後繼續嘗試獲取
if (ttl >= 0) {
getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
getEntry(threadId).getLatch().acquire();
}
}
} finally {
//取消對channel的訂閱
unsubscribe(future, threadId);
}
//get(lockAsync(leaseTime, unit));
}
獲取鎖
獲取鎖的時候，也比較簡單，你可以看到，他也是不斷重新整理過期時間，跟我上面不斷去拿當前時間，校驗過期是一個道理，只是我比較粗糙。
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
//如果帶有過期時間，則按照普通方式獲取鎖
if (leaseTime != -1) {
return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
//先按照30秒的過期時間來執行獲取鎖的方法
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
//如果還持有這個鎖，則開啟定時任務不斷重新整理該鎖的過期時間
ttlRemainingFuture.addListener(new FutureListener<Long>() {
@Override
public void operationComplete(Future<Long> future) throws Exception {
if (!future.isSuccess()) {
return;
}
Long ttlRemaining = future.getNow();
// lock acquired
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
底層加鎖邏輯
你可能會想這麼多操作，在一起不是原子性不還是有問題麼？
大佬們肯定想得到呀，所以還是LUA，他使用了Hash的資料結構。
主要是判斷鎖是否存在，存在就設定過期時間，如果鎖已經存在了，那對比一下執行緒，執行緒是一個那就證明可以重入，鎖在了，但是不是當前執行緒，證明別人還沒釋放，那就把剩餘時間返回，加鎖失敗。
是不是有點繞，多理解一遍。
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit,
long threadId, RedisStrictCommand<T> command) {
//過期時間
internalLockLeaseTime = unit.toMillis(leaseTime);
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
//如果鎖不存在，則通過hset設定它的值，並設定過期時間
"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; " +
//如果鎖已存在，並且鎖的是當前執行緒，則通過hincrby給數值遞增1
"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; " +
//如果鎖已存在，但並非本執行緒，則返回過期時間ttl
"return redis.call('pttl', KEYS[1]);",
Collections.<Object>singletonList(getName()),
internalLockLeaseTime, getLockName(threadId));
}
解鎖
鎖的釋放主要是publish釋放鎖的資訊，然後做校驗，一樣會判斷是否當前執行緒，成功就釋放鎖，還有個hincrby遞減的操作，鎖的值大於0說明是可重入鎖，那就重新整理過期時間。
如果值小於0了，那刪掉Key釋放鎖。
是不是又和AQS很像了？
AQS就是通過一個volatile修飾status去看鎖的狀態，也會看數值判斷是否是可重入的。
所以我說程式碼的設計，最後就萬劍歸一，都是一樣的。
public RFuture<Void> unlockAsync(final long threadId) {
final RPromise<Void> result = new RedissonPromise<Void>();
//解鎖方法
RFuture<Boolean> future = unlockInnerAsync(threadId);
future.addListener(new FutureListener<Boolean>() {
@Override
public void operationComplete(Future<Boolean> future) throws Exception {
if (!future.isSuccess()) {
cancelExpirationRenewal(threadId);
result.tryFailure(future.cause());
return;
}
//獲取返回值
Boolean opStatus = future.getNow();
//如果返回空，則證明解鎖的執行緒和當前鎖不是同一個執行緒，丟擲異常
if (opStatus == null) {
IllegalMonitorStateException cause =
new IllegalMonitorStateException("
attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + threadId);
result.tryFailure(cause);
return;
}
//解鎖成功，取消重新整理過期時間的那個定時任務
if (opStatus) {
cancelExpirationRenewal(null);
}
result.trySuccess(null);
}
});
return result;
}
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, EVAL,
//如果鎖已經不存在， 釋出鎖釋放的訊息
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end;" +
//如果釋放鎖的執行緒和已存在鎖的執行緒不是同一個執行緒，返回null
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
//通過hincrby遞減1的方式，釋放一次鎖
//若剩餘次數大於0 ，則重新整理過期時間
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
//否則證明鎖已經釋放，刪除key併發布鎖釋放的訊息
"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(threadId));
}
總結
這個寫了比較久，但是不是因為複雜什麼的，是因為個人工作的原因，最近事情很多嘛，還是那句話，程式設計師才是我的本職寫文章只是個愛好，不能本末倒置了。
大家會發現，你學懂一個技術棧之後，學新的會很快，而且也能發現他們的設計思想和技巧真的很巧妙，也總能找到相似點，和讓你驚歎的點。
就拿Doug Lea寫的AbstractQueuedSynchronizer（AQS）來說，他寫了一行程式碼，你可能看幾天才能看懂，大佬們的思想是真的牛。
我看原始碼有時候也頭疼，但是去谷歌一下，自己理解一下，突然恍然大悟的時候覺得一切又很值。
學習就是一條時而鬱鬱寡歡，時而開環大笑的路，大家加油，我們成長路上一起共勉。
我是敖丙，一個在網際網路苟且偷生的工具人。
最好的關係是互相成就，大家的**「三連」**就是丙丙創作的最大動力，我們下期見！
注：如果本篇部落格有任何錯誤和建議，歡迎人才們留言，你快說句話啊！
文章持續更新，可以微信搜尋「 三太子敖丙 」第一時間閱讀，回覆【資料】【面試】【簡歷】有我準備的一線大廠面試資料和簡歷模板，本文 GitHub https://github.com/JavaFamily 已經收錄，有大廠面試完整考點，歡迎Star。