前言
分布式鎖一般有三種實現方式：
- 數據庫樂觀鎖；2. 基於Redis的分布式鎖；3. 基於ZooKeeper的分布式鎖。
本篇博客將介紹第二種方式，基於Redis實現分布式鎖。
雖然網上已經有各種介紹Redis分布式鎖實現的博客，然而他們的實現卻有着各種各樣的問題，為了避免誤人子弟，本篇博客將詳細介紹如何正確地實現Redis分布式鎖。
可靠性
首先，為了確保分布式鎖可用，我們至少要確保鎖的實現同時滿足以下四個條件：
互斥性：在任意時刻，只有一個客戶端能持有鎖。
不會發生死鎖：即使有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證后續其他客戶端能加鎖。
具有容錯性：只要大部分的Redis節點正常運行，客戶端就可以加鎖和解鎖。
解鈴還須系鈴人：加鎖和解鎖必須是同一個客戶端，客戶端自己不能把別人加的鎖給解了。
代碼實現
一、引入redis 依賴
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
備注：根據版本不同，jedis 的set 方法也有所不同
二、配置文件增加redis 配置
# redis.properties 配置文件：
# region Redis jedis
# redis配置開始
# Redis數據庫索引（默認為0）
spring.redis.database=0
# Redis服務器地址
spring.redis.host=localhost
# Redis服務器連接端口
spring.redis.port=6379
# Redis服務器連接密碼（默認為空）
spring.redis.password=redis123456
# 連接池最大連接數（使用負值表示沒有限制）
spring.redis.jedis.pool.max-active=1024
# 連接池最大阻塞等待時間（使用負值表示沒有限制）
spring.redis.jedis.pool.max-wait=10000
# 連接池中的最大空閑連接
spring.redis.jedis.pool.max-idle=100
# 連接池中的最小空閑連接
spring.redis.jedis.pool.min-idle=5
# 連接超時時間（毫秒）
spring.redis.timeout=10000
# 連接耗盡時是否阻塞, false報異常,ture阻塞直到超時
spring.redis.block-when-exhausted=true
# endregion
三、利用Spring 把JedisPool 加入Bean 工廠
@Configuration
@PropertySource("classpath:application.properties")
@Slf4j
public class RedisConfig {
@Value("${spring.redis.host}")
private String host;
@Value("${spring.redis.port}")
private int port;
@Value("${spring.redis.timeout}")
private int timeout;
@Value("${spring.redis.jedis.pool.max-idle}")
private int maxIdle;
@Value("${spring.redis.jedis.pool.max-wait}")
private long maxWaitMillis;
@Value("${spring.redis.password}")
private String password;
@Value("${spring.redis.block-when-exhausted}")
private boolean blockWhenExhausted;
@Bean
public JedisPool redisPoolFactory() throws Exception {
log.info("JedisPool注入開始！！");
log.info("redis地址：" + host + ":" + port);
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMaxIdle(maxIdle);
jedisPoolConfig.setMaxWaitMillis(maxWaitMillis);
// 連接耗盡時是否阻塞, false報異常,ture阻塞直到超時, 默認true
jedisPoolConfig.setBlockWhenExhausted(blockWhenExhausted);
// 是否啟用pool的jmx管理功能, 默認true
jedisPoolConfig.setJmxEnabled(true);
JedisPool jedisPool = new JedisPool(jedisPoolConfig, host, port, timeout, password);
log.info("JedisPool注入成功！！");
return jedisPool;
}
}
四、封裝RedisService 對外提供服務
@Service
@Slf4j
public class RedisService {
//Redis 成功返回結果標識
private static final String LOCK_SUCCESS = "OK";
//Reis 操作返回成功
private static final Long RELEASE_SUCCESS = 1L;
//Redis鎖不存在時才設置成功
private static final String SET_IF_NOT_EXIST = "NX";
//Redis鎖超時時間單位
private static final String SET_WITH_EXPIRE_TIME = "PX";
@Autowired
private JedisPool jedisPool;
/**
* 嘗試獲取分布式鎖
*
* @param lockKey 鎖
* @param requestId 請求唯一標識(可以通過uuid等方式獲取唯一ID)
* @param expireTime 超期時間(毫秒)
* @return 是否獲取成功
*/
public boolean tryGetDistributedLock(String lockKey, String requestId, int expireTime) {
Jedis jedis = jedisPool.getResource();
//Jedis 3.0.0 以前的版本
//String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
//Jedis 3.0.0 及以后的版本
SetParams setParams = SetParams.setParams().nx().px(expireTime);
String result = jedis.set(lockKey, requestId, setParams);
return LOCK_SUCCESS.equals(result);
}
/**
* 釋放分布式鎖
*
* @param lockKey 鎖
* @param requestId 請求唯一標識(加鎖時用的唯一標識)
* @return 是否釋放成功
*/
public boolean releaseDistributedLock(String lockKey, String requestId) {
Jedis jedis = jedisPool.getResource();
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
return RELEASE_SUCCESS.equals(result);
}
}
五、簡單解釋
1. 利用redis 的set 命令的 5個參數保證操作的原子性
2. 利用Lua 腳本保證在釋放鎖時的原子性
3. 利用requestId 唯一標識保證不會釋放別人的鎖