Redis分布式可重入鎖實現(xiàn)方案
前言
在單進程環(huán)境下，要保證一個代碼塊的同步執(zhí)行，直接用synchronized 關鍵字或ReetrantLock 即可。在分布式環(huán)境下，要保證多個節(jié)點的線程對代碼塊的同步訪問，就必須要用到分布式鎖方案。
分布式鎖實現(xiàn)方案有很多，有基于關系型數(shù)據(jù)庫行鎖實現(xiàn)的；有基于ZooKeeper臨時順序節(jié)點實現(xiàn)的；還有基于 Redis setnx 命令實現(xiàn)的。本文介紹一下基于 Redis 實現(xiàn)的分布式鎖方案。
理解分布式鎖
實現(xiàn)分布式鎖有幾個要求
- 互斥性：任意時刻，最多只會有一個客戶端線程可以獲得鎖
- 可重入：同一客戶端的同一線程，獲得鎖后能夠再次獲得鎖
- 避免死鎖：客戶端獲得鎖后即使宕機，后續(xù)客戶端也可以獲得鎖
- 避免誤解鎖：客戶端A加的鎖只能由A自己釋放
- 釋放鎖通知：持有鎖的客戶端釋放鎖后，最好可以通知其它客戶端繼續(xù)搶鎖
- 高性能和高可用
Redis 服務端命令是單線程串行執(zhí)行的，天生就是原子的，并且支持執(zhí)行自定義的 lua 腳本，功能上更加強大。
關于互斥性，我們可以用 setnx 命令實現(xiàn)，Redis 可以保證只會有一個客戶端 set 成功。但是由于我們要實現(xiàn)的是一個分布式的可重入鎖，數(shù)據(jù)結構得用 hash，用客戶端ID+線程ID作為 field，value 記作鎖的重入次數(shù)即可。
關于死鎖，代碼里建議把鎖的釋放寫在 finally 里面確保一定執(zhí)行，針對客戶端搶到鎖后宕機的場景，可以給 redis key 設置一個超時時間來解決。
關于誤解鎖，客戶端在釋放鎖時，必須判斷 field 是否當前客戶端ID以及線程ID一致，不一致就不執(zhí)行刪除，這里需要用到 lua 腳本判斷。
關于釋放鎖通知，可以利用 Redis 發(fā)布訂閱模式，給每個鎖創(chuàng)建一個頻道，釋放鎖的客戶端負責往頻道里發(fā)送消息通知等待搶鎖的客戶端。
最后關于高性能和高可用，因為 Redis 是基于內(nèi)存的，天生就是高性能的。但是 Redis 服務本身一旦出現(xiàn)問題，分布式鎖也就不可用了，此時可以多部署幾臺獨立的示例，使用 RedLock 算法來解決高可用的問題。
設計實現(xiàn)
首先我們定義一個 RedisLock 鎖對象的抽象接口，只有嘗試加鎖和釋放鎖方法
public interface RedisLock {
boolean tryLock();
boolean tryLock(long waitTime, long leaseTime, TimeUnit unit);
void unlock();
}然后提供一個默認實現(xiàn) DefaultRedisLock
public class DefaultRedisLock implements RedisLock {
// 客戶端ID UUID
private final String clientId;
private final StringRedisTemplate redisTemplate;
// 鎖頻道訂閱器 接收釋放鎖通知
private final LockSubscriber lockSubscriber;
// 加鎖的key
private final String lockKey;
}關于tryLock() ，首先執(zhí)行l(wèi)ua腳本嘗試獲取鎖，如果加鎖失敗則返回其它客戶端持有鎖的過期時間，客戶端訂閱鎖對應的頻道，然后sleep，直到收到鎖釋放的通知再繼續(xù)搶鎖。最終不管有沒有搶到鎖，都會在 finally 取消頻道訂閱。
@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) {
final long timeout = System.currentTimeMillis() + unit.toMillis(waitTime);
final long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(leaseTime, unit, threadId);
if (ttl == null) {
return true;
}
if (System.currentTimeMillis() >= timeout) {
return false;
}
final Semaphore semaphore = lockSubscriber.subscribe(getChannel(lockKey), threadId);
try {
while (true) {
if (System.currentTimeMillis() >= timeout) {
return false;
}
ttl = tryAcquire(leaseTime, unit, threadId);
if (ttl == null) {
return true;
}
if (System.currentTimeMillis() >= timeout) {
return false;
}
semaphore.tryAcquire(timeout - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lockSubscriber.unsubscribe(getChannel(lockKey), threadId);
}
return false;
}tryAcquire() 就是執(zhí)行l(wèi)ua腳本來加鎖，解釋一下這段腳本的邏輯：首先判斷 lockKey 是否存在，不存在則直接設置 lockKey并且設置過期時間，返回空，表示加鎖成功。存在則判斷 field 是否和當前客戶端ID+線程ID一致，一致則代表鎖重入，遞增一下value即可，不一致代表加鎖失敗，返回鎖的過期時間
private Long tryAcquire(long leaseTime, TimeUnit timeUnit, long threadId) {
return redisTemplate.execute(RedisScript.of(
"if (redis.call('exists', KEYS[1]) == 0) 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;" +
"return redis.call('pttl', KEYS[1]);", Long.class), Collections.singletonList(lockKey),
String.valueOf(timeUnit.toMillis(leaseTime)), getLockName(threadId));
}lockName是由客戶端ID和線程ID組成的：
private String getLockName(long threadId) {
return clientId + ":" + threadId;
}如果加鎖失敗，客戶端會嘗試訂閱對應的頻道，名稱規(guī)則是：
private String getChannel(String lockKey) {
return "__lock_channel__:" + lockKey;
}訂閱方法是LockSubscriber#subscribe ，同一個頻道無需訂閱多個監(jiān)聽器，所以用一個 Map 記錄。訂閱成功以后，會返回當前線程對應的一個 Semaphore 對象，默認許可數(shù)是0，當前線程會調用Semaphore#tryAcquire 等待許可數(shù)，監(jiān)聽器在收到鎖釋放消息后會給 Semaphore 對象增加許可數(shù)，喚醒線程繼續(xù)搶鎖。
@Component
public class LockSubscriber {
@Autowired
private RedisMessageListenerContainer messageListenerContainer;
private final Map<String, Map<Long, Semaphore>> channelSemaphores = new HashMap<>();
private final Map<String, MessageListener> listeners = new HashMap<>();
private final StringRedisSerializer serializer = new StringRedisSerializer();
public synchronized Semaphore subscribe(String channelName, long threadId) {
MessageListener old = listeners.put(channelName, new MessageListener() {
@Override
public void onMessage(Message message, byte[] pattern) {
String channel = serializer.deserialize(message.getChannel());
String ignore = serializer.deserialize(message.getBody());
Map<Long, Semaphore> semaphoreMap = channelSemaphores.get(channel);
if (semaphoreMap != null && !semaphoreMap.isEmpty()) {
semaphoreMap.values().stream().findFirst().ifPresent(Semaphore::release);
}
}
});
if (old == null) {
messageListenerContainer.addMessageListener(listeners.get(channelName), new ChannelTopic(channelName));
}
Semaphore semaphore = new Semaphore(0);
Map<Long, Semaphore> semaphoreMap = channelSemaphores.getOrDefault(channelName, new HashMap<>());
semaphoreMap.put(threadId, semaphore);
channelSemaphores.put(channelName, semaphoreMap);
return semaphore;
}
public synchronized void unsubscribe(String channelName, long threadId) {
Map<Long, Semaphore> semaphoreMap = channelSemaphores.get(channelName);
if (semaphoreMap != null) {
semaphoreMap.remove(threadId);
if (semaphoreMap.isEmpty()) {
MessageListener listener = listeners.remove(channelName);
if (listener != null) {
messageListenerContainer.removeMessageListener(listener);
}
}
}
}
}對于 unlock，就只是一段 lua 腳本，這里解釋一下：判斷當前客戶端ID+線程ID 這個 field 是否存在，存在說明是自己加的鎖，可以釋放。不存在說明不是自己加的鎖，無需做任何處理。因為是可重入鎖，每次 unlock 都只是遞減一下 value，只有當 value 等于0時才是真正的釋放鎖。釋放鎖的時候會 del lockKey，再 publish 發(fā)送鎖釋放通知，讓其他客戶端可以繼續(xù)搶鎖。
@Override
public void unlock() {
long threadId = Thread.currentThread().getId();
redisTemplate.execute(RedisScript.of(
"if (redis.call('hexists', KEYS[1], ARGV[1]) == 0) then " +
"return nil;end;" +
"local counter = redis.call('hincrby', KEYS[1], ARGV[1], -1); " +
"if (counter > 0) then " +
"return 0; " +
"else " +
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], 1); " +
"return 1; " +
"end; " +
"return nil;", Long.class), Arrays.asList(lockKey, getChannel(lockKey)),
getLockName(threadId));
}最后，我們需要一個 RedisLockFactory 來創(chuàng)建鎖對象，它同時會生成客戶端ID
@Component
public class RedisLockFactory {
private static final String CLIENT_ID = UUID.randomUUID().toString();
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private LockSubscriber lockSubscriber;
public RedisLock getLock(String lockKey) {
return new DefaultRedisLock(CLIENT_ID, redisTemplate, lockSubscriber, lockKey);
}
}至此，一個基于 Redis 實現(xiàn)的分布式可重入鎖就完成了。
尾巴
目前這個版本的分布式鎖，保證了互斥性、可重入、避免死鎖和誤解鎖、實現(xiàn)了釋放鎖通知，但是并沒有高可用的保證。如果 Redis 是單實例部署，就會存在單點問題，Redis 一旦故障，整個分布式鎖將不可用。如果 Redis 是主從集群模式部署，雖然有主從自動切換，但是 Master 和 Slave 之間的數(shù)據(jù)同步是存在延遲的，分布式鎖可能會出現(xiàn)問題。比如：客戶端A加鎖成功，lockKey 寫入了 Master，此時 Master 宕機，其它 Slave 升級成了 Master，但是還沒有同步到 lockKey，客戶端B來加鎖也會成功，這就沒有保證互斥性。針對這個問題，可以參考 RedLock 算法，部署多個單獨的 Redis 示例，只要一半以上的Redis節(jié)點加鎖成功就算成功，來盡可能的保證服務高可用。
到此這篇關于Redis分布式可重入鎖實現(xiàn)方案的文章就介紹到這了,更多相關Redis重入鎖內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
本文主要介紹了Redis三種特殊數(shù)據(jù)類型的具體使用，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-02-02
CentOS系統(tǒng)下Redis安裝和自啟動配置的步驟
相信大家都知道Redis是一個C實現(xiàn)的基于內(nèi)存、可持久化的鍵值對數(shù)據(jù)庫，在分布式服務中常作為緩存服務。所以這篇文章將詳細介紹在CentOS系統(tǒng)下如何從零開始安裝到配置啟動服務。有需要的可以參考借鑒。2016-09-09
redis計數(shù)器與數(shù)量控制的實現(xiàn)
使用Redis計數(shù)器可以輕松地解決數(shù)量控制的問題,同時還能有效地提高應用的性能,本文主要介紹了redis計數(shù)器與數(shù)量控制的實現(xiàn),具有一定的參考價值,感興趣的可以了解一下2023-12-12
redis用做緩存是一種非常常見的手段,然而由于內(nèi)存大小的限制，會導致redis在內(nèi)存空間滿了以后需要處理繼續(xù)存入的數(shù)據(jù)，所以就需要淘汰策略，本文就詳細的介紹一下2021-08-08
session存放在服務器，關閉瀏覽器不會失效，本文主要介紹了redis實現(xiàn)分布式session的解決方案，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-03-03
我們在使用Redis分片集群時，集群最好的狀態(tài)就是每個實例可以處理相同或相近比例的請求，但如果不是這樣，則會出現(xiàn)某些實例壓力特別大，而某些實例特別空閑的情況發(fā)生，本文就一起來看下這種情況是如何發(fā)生的以及如何處理2022-12-12
分布式鎖是控制分布式系統(tǒng)不同進程共同訪問共享資源的一種鎖的實現(xiàn)。如果不同的系統(tǒng)或同一個系統(tǒng)的不同主機之間共享了某個臨界資源，往往需要互斥來防止彼此干擾，以保證一致性2022-12-12
基于Redis?Set輕松實現(xiàn)簡單的抽獎系統(tǒng)
Redis Set集合是無序且元素唯一的String類型數(shù)據(jù)結構,支持高效增刪查改操作,包括獲取所有值、判斷包含關系、計算交并差集等,底層基于Hash表實現(xiàn)O(1)時間復雜度,這篇文章主要介紹了基于Redis?Set輕松實現(xiàn)簡單的抽獎系統(tǒng)的相關資料,需要的朋友可以參考下2026-03-03
這篇文章主要介紹了redis由于目標計算機積極拒絕,無法連接的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-07-07
本文主要討論了延時雙刪策略,用于解決緩存與數(shù)據(jù)庫數(shù)據(jù)不一致的問題,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2025-08-08